如何从Sensor任务获取Xcom值并用于Dataflow任务参数

云原生阿岚

2026-7-28

解决方法

步骤1：导入XComArg

首先确保导入Airflow的XComArg（适用于Airflow 2.x及以上版本）：

from airflow.utils.xcom import XComArg

步骤2：从Sensor任务提取目标字符串

假设你的Sensor任务task_id为sensor_task，根据XCom的实际结构提取'df-00...'字符串：

如果Sensor返回的XCom是字典，目标字符串在指定键下（比如file_name，需根据你的XCom结构调整键名）：
```
target_file = XComArg('sensor_task')['file_name']
```
如果Sensor直接返回的就是'df-00...'字符串：
```
target_file = XComArg('sensor_task')
```

步骤3：传入Dataflow任务参数

修改DataflowTemplatedJobStartOperator的配置，将提取到的字符串赋值给inputFilePattern：

dataflow_gcs_to_bq = DataflowTemplatedJobStartOperator(
    task_id='dataflow_gcs_to_bq',
    template='gs://your-template-path',
    job_name='your-job-name',
    location='your-region',
    project_id='your-project-id',
    parameters={
        'javascriptTextTransformFunctionName': 'your-transform-function',
        'JSONPath': 'gs://your-json-path',
        'javascriptTextTransformGcsPath': 'gs://your-js-path',
        'inputFilePattern': target_file,
        'outputTable': 'your-project:your-dataset.your-table',
        'bigQueryLoadingTemporaryDirectory': 'gs://your-temp-bucket',
    },        
)