google-cloud-composer BigQuery 跨数据集加载

Posted 2023-03-25

【中文标题】google-cloud-composer BigQuery 跨数据集加载

【英文标题】：google-cloud-composer BigQuery Cross Dataset load

【发布时间】：2021-04-09 14:50:21

【问题描述】：

我正在尝试运行一个简单的选择查询（来自 biqquery）并使用 Composer 将结果集加载到另一个 bq 表中。但是，我在代码的最后一行出现错误。

损坏的 DAG：[/home/airflow/gcs/dags/es_tc_etl_wkf_mtly.py] 语法无效（es_tc_etl_wkf_mtly.py，第 47 行）

代码：

import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators import bigquery_operator
from datetime import timedelta

default_args = 
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)


dag = DAG(
    'es_tc_etl_wkf',
    default_args=default_args,
    description='Job monitoring dag',
    schedule_interval=None,
    dagrun_timeout=timedelta(minutes=20))

# priority_weight has type int in Airflow DB, uses the maximum.
t1 = bigquery_operator.BigQueryOperator(
        task_id='bq_recent_questions_query',
        bql="""
        select * from bigquery.table.`projectid`.dataset_id.table_A;
        """,`enter code here`
        use_legacy_sql=False,
        destination_dataset_table=dataset_id.table_B)

【参考方案1】：

您有一个额外的enter code here 未分配给任何可识别的参数。 bql已经是deprecated，建议使用sql。此外，查询语句中的整个数据集应包含在反引号 (`) 中。

bql（可以接收一个str代表一个sql语句，一个str的列表 （sql 语句），或对模板文件的引用。模板参考 被以 '.sql' 结尾的 str 识别。） – （已弃用。使用 sql 参数代替）要执行的sql代码（模板化）

应用我上面提到的建议后，你的 t1 应该是这样的。

t1 = bigquery_operator.BigQueryOperator(
        task_id='bq_recent_questions_query',
        sql="""
        select * from `bigquery.table.projectid.dataset_id.table_A`;
        """,
        use_legacy_sql=False,
        destination_dataset_table=dataset_id.table_B)

请参阅此documentation，了解 BigQueryOperator 的使用示例。