从 BigQuery 导出到 MySQL 时出错

Posted 2023-02-16

【中文标题】从 BigQuery 导出到 MySQL 时出错

【英文标题】：Error when exporting from BigQuery to MySQL

【发布时间】：2021-09-30 08:51:52

【问题描述】：

我正在尝试将表从 BigQuery 导出到 Google Cloud mysql 数据库。

我发现这个运算符叫做BigQueryToMySqlOperator（记录在这里https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_api/airflow/providers/google/cloud/transfers/bigquery_to_mysql/index.html?highlight=bigquerytomysqloperator#module-airflow.providers.google.cloud.transfers.bigquery_to_mysql）

当我将包含此任务的 DAG 部署到 Cloud Composer 上时，该任务总是失败并出现错误

Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1113, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1287, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1317, in _execute_task
    result = task_copy.execute(context=context)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/transfers/bigquery_to_mysql.py", line 166, in execute
    for rows in self._bq_get_data():
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/transfers/bigquery_to_mysql.py", line 138, in _bq_get_data
    response = cursor.get_tabledata(
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/bigquery.py", line 2508, in get_tabledata
    return self.hook.get_tabledata(*args, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/bigquery.py", line 1284, in get_tabledata
    rows = self.list_rows(dataset_id, table_id, max_results, selected_fields, page_token, start_index)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/common/hooks/base_google.py", line 412, in inner_wrapper
    raise AirflowException(
airflow.exceptions.AirflowException: You must use keyword arguments in this methods rather than positional

我真的不明白为什么它会抛出这个错误。谁能帮我弄清楚出了什么问题或者我应该如何将数据从 BigQuery 导出到 MySQL DB？非常感谢您的帮助！

编辑：我的操作员代码基本上看起来像这样

transfer_data = BigQueryToMySqlOperator(
            task_id='task_id',
            dataset_table='origin_bq_table',
            mysql_table='dest_table_name',
            replace=True,
        )

【问题讨论】：

发布您的运营商代码

@Elad 我已经在我的操作员代码中进行了编辑，我相信用法足够标准

正如艾伦在他的回答中解释的那样，这可能是一个错误github.com/apache/airflow/issues/17198

不幸的是，它看起来像这样，。谢谢你仍然看看

【参考方案1】：

根据堆栈跟踪，您最有可能使用apache-airflow-providers-google==2.2.0。

airflow.exceptions.AirflowException：您必须在 这个方法而不是位置方法

这个错误源于GoogleBaseHook，可以追溯到BigQueryToMySqlOperator。

BigQueryToMySqlOperator > BigQueryHook > BigQueryConnection > BigQueryCursor > get_tabledata

你得到 AirflowException 的原因是因为get_tabledata 作为执行方法的一部分被调用。

不幸的是，运算符的test 并不全面，因为它只检查调用的方法是否是正确的参数。

我认为这将需要新版本的 google 提供程序，其中 BigQueryToMySqlOperator 中的光标使用关键字参数调用 list_rows，而不是使用位置参数调用 list_rows 的 get_tabledata。

我还在 Airflow 存储库中创建了一个Github Issue。

【讨论】：

非常感谢艾伦，看来我在不久的将来会遇到这个问题

解决此问题的一种方法是使用自定义运算符，该运算符本质上是相同的运算符，但将 get_tabledata 替换为具有命名参数的正确 list_row 方法。