如何明确声明气流连接的字符集=utf8

https://issues.apache.org/jira/browse/AIRFLOW-4824

目前,我有以下代码的变通方法:

def get_uri(hook):
    conn = hook.get_connection(getattr(hook, hook.conn_name_attr))
    login = ''
    if conn.login:
        login = '{conn.login}:{conn.password}@'.format(conn=conn)
    host = conn.host
    if conn.port is not None:
        host += ':{port}'.format(port=conn.port)
    charset = ''
    if conn.extra_dejson.get('charset', False):
        chrs = conn.extra_dejson["charset"]
        if chrs.lower() == 'utf8' or chrs.lower() == 'utf-8':
            charset = '?charset=utf8'
    return '{conn.conn_type}://{login}{host}/{conn.schema}{charset}'.format(
        conn=conn, login=login, host=host, charset=charset)

然后按如下方式使用:

url = get_uri(sql_hook)
from sqlalchemy import create_engine
engine = create_engine(url)

from sqlalchemy import create_engine
from airflow.hooks.mysql_hook import MySqlHook

conn = MySqlHook(mysql_conn_id='conn_id')
uri = conn.get_uri()
engine = create_engine(uri+'?charset=utf8')
df.to_sql('test_table', engine, if_exists='append', index=False)

我用上面的代码解决了这个问题。

我顺便解决了这个问题,效果很好(在文件aiffort.cfg中编辑):

sql_alchemy_conn = mysql://user:password@host:port/airflow?charset=utf8