使用Google Cloud Composer实现BigQuery跨数据集查询加载时遭遇DAG语法错误求助

你遇到的这个语法错误很直观，我梳理了你的代码，问题出在这几个关键地方：

1. 多余代码片段直接触发语法错误 #

在BigQueryOperator的参数列表里，bql字符串结束后多了一行`enter code here`，这行完全是无关的冗余内容，而且它出现在函数参数的定义位置，直接破坏了Python的语法结构——这就是错误提示指向第47行的核心原因。

2. destination_dataset_table参数格式不合法 #

这个参数要求传入项目ID.数据集ID.表名的完整字符串格式，比如'your-project.your-dataset.table_B'。你写的dataset_id.table_B既没有指定项目，也不是合法的字符串（如果dataset_id是自定义变量，你需要提前定义它并拼接成完整路径）。

3. 遗漏任务与DAG的关联参数 #

你的BigQueryOperator任务没有添加dag=dag参数，这会导致Airflow无法识别该任务属于哪个DAG，虽然不会触发语法错误，但部署后任务不会显示在DAG列表中，是个隐性问题。

另外补充个小细节：新版本Airflow中，BigQueryOperator的查询参数推荐用sql代替旧的bql，兼容性和规范性更好。

下面是修正后的完整代码：

import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
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))

# 修正后的BigQuery任务
t1 = BigQueryOperator(
    task_id='bq_recent_questions_query',
    sql="""
        SELECT * FROM `projectid.dataset_id.table_A`;
    """,
    use_legacy_sql=False,
    destination_dataset_table='projectid.dataset_id.table_B',
    dag=dag
)

替换原代码后重新部署，应该就能顺利解决这个问题了。

内容的提问来源于stack exchange，提问作者Dilipan Muthuramalingam