使用 Airflow 将 mysql 数据加载到 bigquery 的 dag 出现“无效参数传递”错误

我运行一个 DAG，提取 MySQL 数据并将其加载到气流中的 BigQuery。我目前收到以下错误：

/usr/local/lib/python2.7/dist-packages/airflow/models.py:1927：PendingDeprecationWarning：无效参数传递给 MySqlToGoogleCloudStorageOperator。 Airflow 2.0 中将不再支持传递此类参数。无效参数是：

*参数：()

**kwargs：{'google_cloud_storage_connn_id'：'podioGCPConnection'}类别= PendingDeprecationWarning

/usr/local/lib/python2.7/dist-packages/airflow/models.py:1927：PendingDeprecationWarning：无效参数传递给 GoogleCloudStorageToBigQueryOperator。 Airflow 2.0 中将不再支持传递此类参数。无效参数是：

*参数：()

**kwargs: {'project_id': 'podio-data'} 类别=PendingDeprecationWarning

dag 的代码在这里：

my_connections = [
    'podiotestmySQL'
]

my_tables = [
    'logistics_orders',
    'logistics_waybills',
    'logistics_shipping_lines',
    'logistics_info_requests'
]

default_args = {
    'owner' : 'tia',
    'start_date' : datetime(2018, 1, 2),
    'depends_on_past' : False,
    'retries' : 1,
    'retry_delay':timedelta(minutes=5),
}

dag = DAG('etl', default_args=default_args,schedule_interval=timedelta(days=1))

slack_notify = SlackAPIPostOperator (
    task_id = 'slack_notfiy',
    token = 'xxxxxx',
    channel='data-status',
    username = 'airflow',
    text = 'Successfully performed podio ETL operation',
    dag=dag)

for connection in my_connections:
    for table in my_tables: 
        extract = MySqlToGoogleCloudStorageOperator(
           task_id="extract_mysql_%s_%s"%(connection,table),
           mysql_conn_id = connection,
           google_cloud_storage_connn_id = 'podioGCPConnection',
           sql = "SELECT *, '%s' as source FROM podiodb.%s"%(connection,table),
           bucket='podio-reader-storage',
           filename= '%s/%s/%s{}.json'%(connection,table,table),
           schema_filename='%s/schemas/%s.json'%(connection,table),
           dag=dag)

       load =GoogleCloudStorageToBigQueryOperator(
           task_id = "load_bg_%s_%s"%(connection,table),
           bigquery_conn_id = 'podioGCPConnection',
           google_cloud_storage_conn_id = 'podioGCPConnection',
           bucket = 'podio-reader-storage',
           destination_project_dataset_table = "Podio_Data1.%s/%s"%(connection,table),
           source_objects = ["%s/%s/%s*.json"%(connection,table,table)],
           schema_object = "%s/schemas/%s.json"%(connection,table),
           source_format = 'NEWLINE_DELIMITED_JSON',
           create_disposition = 'CREATE_IF_NEEDED',
           write_disposition = 'WRITE_TRUNCATE',
           project_id = 'podio-data',
           dag=dag)

      load.set_upstream(extract)
      slack_notify.set_upstream(load)

在这里阅读源码：https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/operators/gcs_to_bq.py https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/operators/gcs_to_bq.py

请从默认参数中删除这些参数：

google_cloud_storage_connn_id = 'podioGCPConnection'
project_id = 'podio-data',

您需要在 Airflow 仪表板中创建连接。