エアフローmysqlからgcpダグエラー

Question

私は最近Airflowでの作業を開始しました。

1. は、MySQLデータベース

BigQueryのにJSONファイルとして

アップロード保存されたJSONファイルをクラウドストレージバケット内のクエリと格納を抽出する問い合わせ：私はそれをDAGに取り組んでいます私は空気の流れ1.8.0、Pythonの3、およびパンダ0.19.0を使用していますMySqlOperator、MySqlToGoogleCloudStorageOperatorとGoogleCloudStorageToBigQueryOperator

：

ダグは、3つの演算子をインポートします。ここで

は私のダグ・コードです：私はそれを実行したときに

sql2gcp_csv = MySqlToGoogleCloudStorageOperator(

    task_id='sql2gcp_csv', 
    sql='airflow_gcp/aws_sql_extract_7days.sql', 
    bucket='gs://{{var.value.gcs_bucket}}/{{ ds_nodash }}/', 
    filename='{{ ds_nodash }}-account-*.json', 
    schema_filename='support/file.json', 
    approx_max_file_size_bytes=1900000000, 
    mysql_conn_id='aws_mysql', 
    google_cloud_storage_conn_id='airflow_gcp', 

) 

しかし、私は次のエラーが表示されます。この例外がスローされる理由

[2017-07-20 22:38:07,478] {models.py:1441} INFO - Marking task as FAILED. 

[2017-07-20 22:38:07,490] {models.py:1462} ERROR - a bytes-like object is required, not 'str' 

/home/User/airflow/workspace/env/lib/python3.5/site-packages/airflow/models.py:1927: PendingDeprecationWarning: Invalid arguments were passed to MySqlOperator. Support for passing such arguments will be dropped in Airflow 2.0. Invalid arguments were: 
*args:() 
**kwargs: {'database': 'test'} 
category=PendingDeprecationWarning 

/home/User/airflow/workspace/env/lib/python3.5/site- 
packages/airflow/ti_deps/deps/base_ti_dep.py:94: PendingDeprecationWarning: generator '_get_dep_statuses' raised StopIteration 

for dep_status in self._get_dep_statuses(ti, session, dep_context): 
Traceback (most recent call last): 

File "/home/User/airflow/workspace/env/bin/airflow", line 28, in <module> args.func(args) 

File "/home/User/airflow/workspace/env/lib/python3.5/site-packages/airflow/bin/cli.py", line 422, in run pool=args.pool, 

File "/home/User/airflow/workspace/env/lib/python3.5/site-packages/airflow/utils/db.py", line 53, in wrapper result = func(*args, **kwargs) 

File "/home/User/airflow/workspace/env/lib/python3.5/site-packages/airflow/models.py", line 1374, in run result = task_copy.execute(context=context) 

File "/home/User/airflow/workspace/env/lib/python3.5/site-packages/airflow/contrib/operators/mysql_to_gcs.py", line 91, in execute files_to_upload = self._write_local_data_files(cursor) 

File "/home/User/airflow/workspace/env/lib/python3.5/site-packages/airflow/contrib/operators/mysql_to_gcs.py", line 136, in _write_local_data_files 
json.dump(row_dict, tmp_file_handle) 

File "/usr/lib/python3.5/json/__init__.py", line 179, in dump 

TypeError: a bytes-like object is required, not 'str' 

誰でも知っていますか？

Answer 1

トレースバックによると、コードはthis pointに違反しています。

json.dump(row_dict, tmp_file_handle) 

tmp_file_handleは、デフォルトの入力引数とNamedTemporaryFileinitializedは、つまり、それはw+bモードで開かれたファイルをシミュレートしている（したがって、唯一の入力としてバイトのようなデータを受け入れ）：あなたが見ることができるように、それはコードを処理します。

問題は、Python 2ではすべての文字列がバイトであるのに対し、Python 3では文字列（デフォルトではutf-8）です。

あなたは、Python 2開いて、このコードを実行する場合：

In [1]: from tempfile import NamedTemporaryFile 
In [2]: tmp_f = NamedTemporaryFile(delete=True) 
In [3]: import json 
In [4]: json.dump({'1': 1}, tmp_f) 

それは正常に動作します。

しかし、あなたは、Python 3を開いて、同じコード実行した場合：

In [54]: from tempfile import NamedTemporaryFile 
In [55]: tmp_f = NamedTemporaryFile(delete=True) 
In [56]: import json 
In [57]: json.dump({'1': 1}, tmp_f) 
--------------------------------------------------------------------------- 
TypeError         Traceback (most recent call last) 
<ipython-input-57-81743b9013c4> in <module>() 
----> 1 json.dump({'1': 1}, tmp_f) 

/usr/local/lib/python3.6/json/__init__.py in dump(obj, fp, skipkeys, ensure_ascii, check_circular, allow_nan, cls, indent, separators, default, sort_keys, **kw) 
    178  # a debuggability cost 
    179  for chunk in iterable: 
--> 180   fp.write(chunk) 
    181 
    182 

/usr/local/lib/python3.6/tempfile.py in func_wrapper(*args, **kwargs) 
    481    @_functools.wraps(func) 
    482    def func_wrapper(*args, **kwargs): 
--> 483     return func(*args, **kwargs) 
    484    # Avoid closing the file as long as the wrapper is alive, 
    485    # see issue #18879. 

TypeError: a bytes-like object is required, not 'str' 

を私たちはあなたと同じエラーを取得します。

これは、まだPython 3では完全にサポートされていないことを意味しています（test coverageのように、モジュールairflow/contrib/operators/mysql_to_gcs.pyはまだPython 2または3でテストされていません）。これを確認する1つの方法は、Python 2を使用してコードを実行し、それが動作するかどうかを確認することです。

their JIRAで、どちらのバージョンのPythonでも移植性を要求することをお勧めします。