企业官网建设流程全解析

1. 项目概述：为什么要在 Airflow DAG 中接收命令行参数

Airflow 1.10.11 是一个被大量中大型企业稳定采用的长期支持版本，尤其在金融、电商、数据中台等对系统稳定性要求极高的场景中，至今仍有大量生产环境未升级至 2.x。这个版本的 DAG 调度机制默认是“静态定义+定时触发”，所有任务逻辑、输入路径、日期范围、业务标识都硬编码在 Python 文件里，导致同一个 DAG 每次复用都要复制粘贴改代码——我去年在某银行的数据治理平台维护期间，就因为一个清洗任务要适配 7 个不同分行的账期口径，硬生生写了 7 个几乎一模一样的 DAG 文件，光是后续修一个 SQL 小 bug 就得同步改 7 处，上线前还得逐个校验，出过两次因漏改导致某分行数据延迟 12 小时的事故。

而“DAG 接收命令行参数”这件事，本质不是为了炫技，而是要把 DAG 从“一次性脚本”升级为“可复用服务接口”。它解决的是三个真实痛点：第一，避免重复造轮子——同一套 ETL 逻辑，只需一个 DAG，通过传参切换 source_system=corebanking 或 source_system=creditcard；第二，支撑人工干预场景——比如某天上游数据异常，需要重跑 2023-08-15 到 2023-08-20 的全量数据，但又不想动调度周期或临时改代码，这时直接airflow trigger_dag -d my_etl_dag --conf '{"start_date":"2023-08-15","end_date":"2023-08-20"}'就能精准触发；第三，打通外部系统调用链路——比如和 BI 平台集成，用户在前端点“生成月报”，后端自动拼装{"report_type":"monthly","month":"2023-08"}并调用 Airflow API 触发 DAG，整个流程零人工介入。

很多人误以为“加参数=改 DAG 文件”，其实完全不是。Airflow 1.10.11 的--conf参数机制，是在不修改任何 Python 代码的前提下，把 JSON 字符串注入到 DAG 运行上下文里，再由PythonOperator或自定义 Operator 在执行时动态读取。它不依赖 Web UI 配置、不修改数据库表结构、不重启 Scheduler，属于“热插拔式”能力扩展。我实测过，在千级 DAG 规模的集群上，单次带 conf 的 trigger_dag 命令平均耗时 120ms，比普通触发只多 15ms，性能损耗几乎可忽略。关键在于，这个能力必须配合正确的 DAG 编写范式才能生效——不是所有写法都能接收到参数，也不是所有 Operator 都能正确解析 conf。接下来我会从设计思路、核心细节、完整实操到排障技巧，一层层拆给你看。

2. 整体设计与思路拆解：为什么必须用--conf而不是环境变量或全局变量

2.1 三种常见参数传递方式的对比与淘汰逻辑

刚接触这个问题的开发者，通常会本能想到三种方案：改环境变量（os.environ）、用全局变量（global VAR）、或者直接在 DAG 定义里写死default_args。但我在某保险公司的数据平台迁移项目中，亲手踩过这三类坑，最终全部放弃，原因非常具体：

• 环境变量方案：你可能会想export MY_PARAM=value && airflow trigger_dag my_dag。问题在于，Airflow 的 Scheduler 和 Worker 是分离进程，Scheduler 收到 trigger 请求后，会把 DAG 实例化信息写入数据库，Worker 从数据库拉取任务时，根本看不到你当初在 Scheduler 机器上设的环境变量。更致命的是，如果 Worker 部署在 Docker 容器或 Kubernetes Pod 里，环境变量根本跨不过容器边界。我试过用kubectl exec -it worker-pod -- sh -c "export MY_PARAM=test && airflow run..."，结果任务日志里打印os.environ.get('MY_PARAM')始终是 None——因为airflow run命令启动的是新进程，继承的是容器启动时的环境，不是你 exec 进去临时设的那个。

• 全局变量方案：有人会在 DAG 文件顶部写RUNTIME_PARAM = None，然后幻想在 trigger 时能动态赋值。这是对 Airflow 执行模型的根本误解。DAG 文件在 Scheduler 启动时就被 import 一次，所有顶层代码（包括RUNTIME_PARAM = None）只执行一遍，之后每次 trigger 都是复用内存里已加载的 DAG 对象，不会重新执行文件。你改的只是本地变量副本，Scheduler 根本感知不到。我曾看到有同事在 DAG 里写RUNTIME_PARAM = json.loads(os.environ.get('AIRFLOW_CONF', '{}'))，结果发现所有任务都共享同一个RUNTIME_PARAM引用，当两个并发任务同时修改它时，出现数据错乱——比如任务 A 设了{"date":"2023-01"}，任务 B 设了{"date":"2023-02"}，最后两个任务都读到了"2023-02"。

• default_args 硬编码方案：比如default_args={'params': {'date': '2023-01'}}。这看似可行，但它把参数固化在 DAG 定义里，每次要换参数就得改代码、提交 Git、触发 CI/CD、等待部署，完全违背“快速响应业务需求”的初衷。某电商大促期间，运营临时要求把双十一流量分析窗口从 7 天改成 14 天，按这个方案至少要 2 小时才能上线，而实际业务等不了。

所以最终我们锁定--conf方案，不是因为它最炫，而是它唯一满足四个硬性条件：进程隔离安全（每个任务实例独享 conf）、跨进程可靠（conf 通过数据库持久化传递）、零代码变更（DAG 文件完全不动）、API 友好（支持 CLI、REST API、Python SDK 全渠道调用）。它的底层原理其实很清晰：当你执行airflow trigger_dag -d my_dag --conf '{"key":"value"}'时，CLI 工具会把 JSON 字符串存进dag_run.conf字段（对应数据库表dag_run的conf列），Scheduler 在调度时把这个字段反序列化成 Python dict，再注入到context中，最终在PythonOperator的python_callable函数里，通过kwargs['dag_run'].conf就能拿到原始 JSON。

2.2 DAG 编写范式的强制约定：为什么必须用PythonOperator+provide_context=True

在 Airflow 1.10.11 中，--conf的参数只有在特定 Operator 且特定配置下才能被正确传递。我翻过源码（airflow/operators/python_operator.py第 92 行），确认了关键逻辑：PythonOperator的execute方法会检查self.provide_context是否为 True，如果是，就把context（含dag_run.conf）作为**kwargs传给你的python_callable函数。如果设成 False，kwargs就是空字典，dag_run.conf直接丢失。

更隐蔽的坑是 Operator 类型限制。BashOperator、EmailOperator这些内置 Operator，压根不提供dag_run上下文访问入口。你不能指望在bash_command='echo {{ dag_run.conf.date }}'里直接 Jinja 渲染 conf，因为dag_run.conf不在 Jinja 的context变量白名单里（1.10.11 的 Jinja context 默认只包含ds,ts,params等预定义变量，dag_run是被刻意过滤的）。我试过强行在BashOperator里写bash_command='python -c "import json; print(json.loads(\'{{ ti.xcom_pull(key=\'conf_json\') }}\').get(\'date\'))"'，结果发现ti.xcom_pull根本拿不到 conf——因为 conf 是 DAG Run 级别的，不是 Task Instance 级别，XCom 默认不自动传递。

所以标准解法只有一个：所有需要读取 conf 的逻辑，必须封装在PythonOperator里，且必须设置provide_context=True。这是不可妥协的范式。你可以把它理解成一道“参数安检门”——只有经过这道门的任务，才能合法拿到 conf。至于其他 Operator，它们的角色是“执行者”，不是“决策者”；真正的参数解析、分支判断、动态 SQL 拼接，必须在 PythonOperator 里做完，再把结果通过 XCom 传给下游 BashOperator 或 SparkOperator。

2.3 参数安全边界的划定：为什么 conf 必须是合法 JSON，且 key 名要避开保留字

--conf后面跟的必须是严格格式的 JSON 字符串，不是任意字符串。我见过最典型的错误是airflow trigger_dag -d my_dag --conf "{date:2023-01}"，少引号、少花括号、用单引号，结果触发失败，日志里只显示json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes。Airflow 内部用的是标准json.loads()，只认双引号，不认单引号，且 key 和 string value 都必须用双引号包裹。

更深层的安全约束在于 key 名的命名规范。dag_run.conf是一个普通的 Python dict，但当你在 Jinja 模板里想间接使用它时（比如传给下游任务的op_kwargs），就会撞上 Airflow 的变量名黑名单。1.10.11 的源码里明确定义了RESERVED_MACRO_NAMES = ['macros', 'params', 'dag', 'task', 'ts', 'ds', 'yesterday_ds']，如果你的 conf 里用了{"params": "xxx"}，那么在PythonOperator的python_callable函数里，kwargs['dag_run'].conf['params']是能正常取到的，但一旦你试图在BashOperator的bash_command里写echo {{ params }}，Jinja 会优先返回default_args['params']（如果设置了的话），而不是你传的 conf。我在线上环境因此出过一次严重事故：conf 里传了{"params": {"env": "prod"}}，下游任务却读到了测试环境的params，导致数据写进了 UAT 库。

所以我的经验是：conf 的 key 名必须满足三个条件——全小写、下划线分隔、避开所有 Airflow 保留字。推荐用run_id,data_source,execution_mode,override_date这类无歧义的名称。data_source比source好，因为source可能和某些 Operator 的source参数冲突；override_date比date好，因为date是 Python 内置模块名，虽然不常用，但为防万一。另外，value 值尽量避免嵌套过深，conf 是存在数据库里的，Airflow 对conf字段的数据库类型是 TEXT，但过长的 JSON（> 64KB）在 MySQL 里可能触发max_allowed_packet限制，我建议单次 conf 总大小控制在 16KB 以内，足够覆盖绝大多数业务参数。

3. 核心细节解析与实操要点：从 DAG 定义到参数解析的完整链路

3.1 DAG 文件的标准结构：一个可直接抄作业的模板

下面是一个经过生产环境千次验证的 DAG 模板，专为 Airflow 1.10.11 设计，所有注释都是基于真实踩坑经验写的，不是照搬文档：

# -*- coding: utf-8 -*- """ my_etl_dag.py 功能：通用数据抽取 DAG，支持通过 --conf 动态指定数据源、日期范围、执行模式 作者：某数据平台组 最后更新：2023-08-20 """ from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python_operator import PythonOperator from airflow.operators.bash_operator import BashOperator from airflow.hooks.base_hook import BaseHook import logging import json # 设置默认参数，注意：这里不放任何业务参数，只放调度相关参数 default_args = { 'owner': 'data_engineer', 'depends_on_past': False, 'start_date': datetime(2023, 1, 1), 'email_on_failure': True, 'email': ['alert@company.com'], 'retries': 1, 'retry_delay': timedelta(minutes=5), } # 创建 DAG 实例，注意：schedule_interval 设为 None，表示不自动调度，只手动触发 dag = DAG( 'my_etl_dag', default_args=default_args, description='通用 ETL DAG，支持命令行参数', schedule_interval=None, # 关键！必须设为 None，否则会和 --conf 冲突 catchup=False, # 关键！必须设为 False，否则会补跑历史周期，conf 参数无法覆盖 max_active_runs=1, # 可选，防止并发混乱 ) def parse_conf_and_prepare(**kwargs): """ 【核心函数】解析 dag_run.conf，并做基础校验和默认值填充 kwargs: 由 provide_context=True 自动注入的上下文 """ # 1. 从上下文中提取 conf，这是唯一合法入口 dag_run_conf = kwargs.get('dag_run', {}).conf if kwargs.get('dag_run') else {} # 2. 日志记录原始 conf，方便排障（线上必须加，否则出问题找不到源头） logging.info(f"[parse_conf_and_prepare] Raw conf received: {dag_run_conf}") # 3. 定义参数 schema，强制校验必要字段 required_keys = ['data_source', 'start_date', 'end_date'] missing_keys = [k for k in required_keys if k not in dag_run_conf] if missing_keys: raise ValueError(f"Missing required conf keys: {missing_keys}. " f"Please pass --conf '{{\"data_source\":\"xxx\",\"start_date\":\"YYYY-MM-DD\",\"end_date\":\"YYYY-MM-DD\"}}'") # 4. 设置默认值，避免空值导致下游报错 conf = { 'data_source': dag_run_conf['data_source'].strip(), 'start_date': dag_run_conf['start_date'].strip(), 'end_date': dag_run_conf['end_date'].strip(), 'execution_mode': dag_run_conf.get('execution_mode', 'full').strip(), # 默认 full 'override_table': dag_run_conf.get('override_table'), # 可选，None 表示不覆盖 } # 5. 业务校验：日期格式必须合法，且 start <= end try: start_dt = datetime.strptime(conf['start_date'], '%Y-%m-%d') end_dt = datetime.strptime(conf['end_date'], '%Y-%m-%d') if start_dt > end_dt: raise ValueError(f"start_date {conf['start_date']} > end_date {conf['end_date']}") except ValueError as e: raise ValueError(f"Invalid date format in conf: {e}") # 6. 将校验后的 conf 存入 XCom，供下游任务使用（XCom 是跨任务传参的唯一安全通道） kwargs['ti'].xcom_push(key='parsed_conf', value=conf) logging.info(f"[parse_conf_and_prepare] Parsed and validated conf: {conf}") def generate_dynamic_sql(**kwargs): """ 【核心函数】根据 conf 动态生成 SQL，支持不同 data_source 的方言 """ # 从 XCom 拉取上一步解析好的 conf conf = kwargs['ti'].xcom_pull(key='parsed_conf', task_ids='parse_conf_task') # 1. 根据 data_source 选择不同的 SQL 模板 sql_templates = { 'mysql_corebanking': """ INSERT OVERWRITE TABLE dwd.fact_transaction PARTITION(ds='{{ ds }}') SELECT trans_id, amount, currency, FROM_UNIXTIME(CAST(create_time AS BIGINT)) as create_time FROM ods.corebanking_transaction WHERE ds BETWEEN '{{ conf.start_date }}' AND '{{ conf.end_date }}' AND status = 'SUCCESS'; """, 'hive_creditcard': """ INSERT OVERWRITE TABLE dwd.fact_transaction PARTITION(ds='{{ ds }}') SELECT id as trans_id, amt as amount, cur as currency, from_unixtime(cast(create_ts as bigint)) as create_time FROM ods.creditcard_transaction WHERE ds >= '{{ conf.start_date }}' AND ds <= '{{ conf.end_date }}' AND result_code = '0000'; """ } # 2. 获取对应模板，如果没匹配到，抛出明确错误 template = sql_templates.get(conf['data_source']) if not template: raise ValueError(f"Unsupported data_source: {conf['data_source']}. " f"Supported: {list(sql_templates.keys())}") # 3. Jinja 渲染，注意：这里用的是 Python 的 str.format，不是 Airflow 的 Jinja # 因为我们要在运行时动态替换，不是在 DAG 解析时 rendered_sql = template.replace("{{ conf.start_date }}", conf['start_date']) \ .replace("{{ conf.end_date }}", conf['end_date']) \ .replace("{{ ds }}", kwargs['ds']) # ds 是 Airflow 自带的宏 # 4. 把渲染好的 SQL 存入 XCom，供下游执行 kwargs['ti'].xcom_push(key='final_sql', value=rendered_sql) logging.info(f"[generate_dynamic_sql] Final SQL length: {len(rendered_sql)} chars") def execute_sql_with_spark(**kwargs): """ 【核心函数】调用 Spark 提交 SQL，支持不同集群配置 """ conf = kwargs['ti'].xcom_pull(key='parsed_conf', task_ids='parse_conf_task') final_sql = kwargs['ti'].xcom_pull(key='final_sql', task_ids='generate_sql_task') # 1. 根据 execution_mode 选择 Spark 提交模式 if conf['execution_mode'] == 'dry_run': logging.info(f"[execute_sql_with_spark] DRY RUN MODE: SQL would be:\n{final_sql}") return # 直接返回，不执行 # 2. 构建 spark-submit 命令 spark_cmd = f""" spark-sql \ --master yarn \ --deploy-mode client \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.adaptive.coalescePartitions.enabled=true \ -e "{final_sql.replace('"', '\\"')}" """ # 3. 记录命令（脱敏处理，不打日志敏感信息） logging.info(f"[execute_sql_with_spark] Executing spark-sql with mode: {conf['execution_mode']}") # 4. 实际执行（这里用 BashOperator 封装，但核心逻辑在 Python 里） import subprocess result = subprocess.run(spark_cmd, shell=True, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError(f"Spark SQL execution failed: {result.stderr}") logging.info(f"[execute_sql_with_spark] Spark SQL executed successfully") # 定义任务链 parse_conf_task = PythonOperator( task_id='parse_conf_task', python_callable=parse_conf_and_prepare, provide_context=True, # 关键！必须为 True dag=dag, ) generate_sql_task = PythonOperator( task_id='generate_sql_task', python_callable=generate_dynamic_sql, provide_context=True, dag=dag, ) execute_sql_task = PythonOperator( task_id='execute_sql_task', python_callable=execute_sql_with_spark, provide_context=True, dag=dag, ) # 设置任务依赖 parse_conf_task >> generate_sql_task >> execute_sql_task

这个模板的关键设计点，我都用注释标出来了。它不是一个“能跑就行”的玩具，而是为生产环境打磨过的：

• schedule_interval=None和catchup=False是硬性要求，否则--conf会被自动调度覆盖；
• parse_conf_and_prepare函数里做了完整的参数校验、类型转换、业务规则检查，把错误拦截在第一步，避免下游任务执行到一半才报错；
• 所有参数传递都走 XCom，不依赖全局变量或环境变量，保证每个 Task Instance 数据隔离；
• generate_dynamic_sql里用str.replace而不是 Jinja，是因为 Jinja 在 DAG 解析阶段就渲染了，而我们需要在任务运行时（runtime）才动态决定 SQL 内容；
• execute_sql_with_spark里加入了dry_run模式，这是线上必备的安全开关，任何新参数上线前，先用--conf '{"execution_mode":"dry_run"}'看 SQL 是否符合预期。

3.2 CLI 触发的完整命令与参数组合策略

airflow trigger_dag是最常用的触发方式，但它的参数组合有很多门道。我整理了一份“参数组合速查表”，覆盖 95% 的线上场景：

特别提醒两个高频陷阱：
第一，JSON 字符串里的双引号必须转义。Linux bash 里，--conf '{"key":"value"}'是合法的，但如果你的 value 本身含双引号，比如{"sql":"SELECT * FROM t WHERE name=\"test\""}"，就必须写成--conf '{\"sql\":\"SELECT * FROM t WHERE name=\\\"test\\\"\"}'。我建议统一用 Python 脚本生成 conf 字符串，避免手写出错。
第二，-e和--conf的优先级关系。-e设置的execution_date会影响所有 Airflow 内置宏（ds,next_ds,yesterday_ds），但它不会覆盖conf里的start_date和end_date。这两个是完全独立的维度：execution_date是调度时间戳，conf是业务参数。比如你-e 2023-08-05，但conf里start_date="2023-07-01"，那么任务会以2023-08-05为ds运行，但 SQL 查询的是2023-07-01到2023-07-31的数据——这是完全合理的设计，意味着“用今天的计算资源，处理上个月的数据”。

3.3 Web UI 和 REST API 的替代触发方式

虽然 CLI 是最直接的方式，但在自动化集成场景，Web UI 和 REST API 更实用。Airflow 1.10.11 的 REST API（需开启webserver_config.py中的enable_proxy_fix = True）提供了/api/experimental/dags/<DAG_ID>/dag_runs接口，支持 POST 请求触发 DAG 并传 conf。

一个真实的 Python 调用示例（用 requests 库）：

import requests import json # Airflow Web Server 地址，需配置认证（Basic Auth 或 Kerberos） airflow_url = "http://airflow-webserver:8080" auth = ("admin", "password") # 生产环境必须用更安全的 token 认证 # 构造请求体 payload = { "conf": { "data_source": "hive_creditcard", "start_date": "2023-08-01", "end_date": "2023-08-05", "execution_mode": "full" } } # 发送 POST 请求 response = requests.post( f"{airflow_url}/api/experimental/dags/my_etl_dag/dag_runs", auth=auth, headers={"Content-Type": "application/json"}, data=json.dumps(payload) ) if response.status_code == 200: run_info = response.json() print(f"DAG triggered successfully! Run ID: {run_info['dag_run_id']}") else: print(f"Trigger failed: {response.status_code} {response.text}")

Web UI 的操作路径是：进入 DAG 页面 → 点击右上角 “Trigger DAG” 按钮 → 在弹出框里粘贴 JSON 到 “Configuration JSON” 文本框 → 点击 “Trigger”。注意，Web UI 的文本框对 JSON 格式非常敏感，少一个逗号或引号都会报错，而且错误提示极其简陋（就一个红色 “Bad Request”），所以建议先用json.loads()在本地校验 JSON 合法性，再粘贴。

REST API 的最大优势是可编程，可以轻松集成到 CI/CD 流水线、BI 平台、甚至钉钉/企微机器人里。比如某零售公司的 BI 系统，用户在报表页面点“刷新数据”，后端就调用这个 API，把用户选择的日期范围、门店 ID 等参数组装成 conf，触发对应的 DAG。整个过程对用户透明，体验接近“点击即生效”。

4. 实操过程与核心环节实现：从零部署到线上验证的全流程

4.1 环境准备与版本确认（避坑第一步）

在动手写代码前，必须确认你的 Airflow 环境确实是 1.10.11，且配置正确。我见过太多人因为版本混淆而浪费数小时。执行以下命令逐项验证：

# 1. 确认 Airflow 版本（必须精确到 patch version） $ airflow version 1.10.11 # 2. 确认 Scheduler 和 Worker 进程都在运行（--conf 依赖两者协同） $ ps aux | grep airflow | grep -E "(scheduler|worker)" # 应该看到类似：/usr/bin/python ... airflow scheduler # /usr/bin/python ... airflow worker # 3. 检查数据库连接（conf 存储在 dag_run 表，必须可写） $ airflow db check [2023-08-05 10:23:45,123] {db.py:371} INFO - Checking database connectivity... [2023-08-05 10:23:45,125] {db.py:374} INFO - Database connection successful. # 4. 验证 Web UI 是否启用实验性 API（REST API 依赖此项） # 检查 airflow.cfg，确认以下两项为 True： # enable_api = True # expose_config = False # 生产环境必须为 False，防信息泄露

最关键的一步是检查airflow.cfg中的executor配置。--conf在SequentialExecutor下工作正常，但在CeleryExecutor下，必须确保celery_result_backend配置正确（通常是redis://localhost:6379/0），否则dag_run.conf可能无法被 Worker 正确读取。我在线上遇到过一次，Scheduler 写入了 conf，但 Worker 从 Redis 拉取时超时，导致kwargs['dag_run']是 None，任务直接失败。解决方案是：celery_result_backend = redis://<redis-host>:6379/0，并确保 Redis 服务健康。

4.2 DAG 文件部署与语法校验（零错误上线）

把上面的模板保存为my_etl_dag.py，放到 Airflow 的dags_folder（通常在~/airflow/dags/）。部署后，不要急着触发，先做三步校验：

第一步：DAG 语法校验

$ airflow list_dags # 查看输出中是否有 my_etl_dag，且没有 ERROR $ airflow list_tasks my_etl_dag # 应该输出：parse_conf_task, generate_sql_task, execute_sql_task

第二步：DAG 结构校验

$ airflow show_dag my_etl_dag # 检查输出中的 'schedule_interval' 是否为 'None'，'catchup' 是否为 'False' # 检查 'tasks' 列表里每个 task 的 'operator' 是否为 'PythonOperator' # 检查 'params' 字段是否为空（应该为空，因为我们没设 default_args['params']）

第三步：模拟执行校验（最有效）

# 使用 airflow test 模拟单次任务执行，不依赖 Scheduler $ airflow test my_etl_dag parse_conf_task 2023-08-05 --conf '{"data_source":"mysql_corebanking","start_date":"2023-08-01","end_date":"2023-08-05"}' # 如果成功，会输出详细日志，最后一行是 "Done."，且没有 Traceback # 如果失败，会直接打印错误堆栈，比如 "ValueError: Missing required conf keys..."

airflow test是神命令，它绕过整个调度引擎，在本地直接运行任务函数，是调试--conf解析逻辑的最快方式。我习惯在写完parse_conf_and_prepare函数后，立刻用airflow test验证各种边界 case：空 conf、缺字段、日期格式错误、SQL 注入字符（如"; DROP TABLE）等，确保防御性编程到位。

4.3 首次触发与日志追踪（如何读懂 Airflow 日志）

假设一切校验通过，现在执行首次触发：

$ airflow trigger_dag -d my_etl_dag --conf '{"data_source":"mysql_corebanking","start_date":"2023-08-01","end_date":"2023-08-05"}'

触发成功后，立即打开 Web UI，进入my_etl_dag页面，找到刚创建的 DAG Run（run_id类似scheduled__2023-08-05T10-23-45.123456+00-00），点击进入，再点击parse_conf_task任务，查看日志。一个健康的日志流应该包含这些关键行：

[2023-08-05 10:23:45,123] {logging_mixin.py:95} INFO - [parse_conf_and_prepare] Raw conf received: {'data_source': 'mysql_corebanking', 'start_date': '2023-08-01', 'end_date': '2023-08-05'} [2023-08-05 10:23:45,124] {logging_mixin.py:95} INFO - [parse_conf_and_prepare] Parsed and validated conf: {'data_source': 'mysql_corebanking', 'start_date': '2023-08-01', 'end_date': '2023-08-05', 'execution_mode': 'full', 'override_table': None} [2023-08-05 10:23:45,125] {base_task_runner.py:118} INFO - Job 123: Subtask parse_conf_task Done.

如果看到Raw conf received: {}，说明--conf没传进去，检查 CLI 命令的 JSON 格式；如果看到KeyError: 'dag_run'，说明provide_context=False，回去改 DAG；如果看到ValueError: Missing required conf keys，说明传的 JSON 缺字段。

下游任务的日志同样重要。点开generate_sql_task，应该能看到类似：

[2023-08-05 10:23:46,789] {logging_mixin.py:95} INFO - [generate_dynamic_sql] Final SQL length: 328 chars

再点开 `execute_sql