要想通过 Airflow 将大量数据插入到 BigQuery 中,可以尝试使用 BigQueryHook 和 BigQueryOperator 实现。示例代码如下:
from airflow.contrib.hooks.bigquery_hook import BigQueryHook
from airflow.operators.bash_operator import BashOperator
from airflow.operators.bigquery_operator import BigQueryOperator
# 设置 BigQueryHook
bq_hook = BigQueryHook(bigquery_conn_id='my_bigquery_conn')
# 插入 1 亿行数据到 BigQuery
task = BigQueryOperator(
task_id='bigquery_task',
sql='SELECT * FROM my_data_table',
destination_dataset_table='project_id.dataset_id.my_new_table',
write_disposition='WRITE_TRUNCATE',
create_disposition='CREATE_IF_NEEDED',
bigquery_conn_id='my_bigquery_conn',
)
# 执行任务
task.execute(context=None)
这里使用了 BigQueryHook 连接到 BigQuery,然后使用 BigQueryOperator 插入数据。注意设置 write_disposition 为 WRITE_TRUNCATE,这将覆盖已存在的表格,因此可以使用 truncate table 方法避免在重复插入时导致数据重复。