任务组中的拉列表 xcom 不起作用答案

【问题标题】：Pull list xcoms in TaskGroups not working任务组中的拉列表 xcom 不起作用
【发布时间】：2022-10-02 02:37:15
【问题描述】：

我的气流代码具有以下可调用的 Python 运算符，我在其中创建列表并将其推送到 xcoms：

keys = []
values = []

def attribute_count_check(e_run_id,**context):
    
    job_run_id = int(e_run_id)
    da = \"select count (distinct row_num) from dds_metadata.dds_temp_att_table where run_id ={}\".format(job_run_id)
    cursor.execute(da)
    res = cursor.fetchall()
    view_res = [x for res in res for x in res]
    count_of_sql = view_res[0]
    print(count_of_sql)
    if count_of_sql < 1:
        print(\"deleting of cluster\")
        return \'delete_cluster\'    
    else :
        print(\"triggering attr_check\")
        num_attributes_per_task = num_attr #job_config
        diff = math.ceil (count_of_sql / num_attributes_per_task)
        instance = int(diff)
        n = num_attributes_per_task
        global values
        global keys
        for r in range(1, instance+1):
            #a = r
            keys.append(r)
            lower_ranges =(n*(r-1)) +1
            upper_range = (n*(r - 1)) + n
            b =(lower_ranges,upper_range)
            values.append(b)
            task_instance = context[\'task_instance\']
            task_instance.xcom_push(key=\"di_keys\", value=keys)
            task_instance.xcom_push(key=\"di_values\", value=values)

工作中的 xcoms 如下图所示：

现在我正在尝试从 xcoms 获取值以使用以下代码动态创建集群：

with TaskGroup(\'dataproc_create_cluster\',prefix_group_id=False) as dataproc_create_clusters:

    for i in zip(\'{{ ti.xcom_pull(key=\"di_keys\")}}\',\'{{ ti.xcom_pull(key=\"di_values\")}}\'):
        dynmaic_create_cluster = DataprocCreateClusterOperator(
        task_id=\"create_cluster_{}\".format(list(eval(str(i)))[0]),
        project_id=\'{0}\'.format(PROJECT),
        cluster_config=CLUSTER_GENERATOR_CONFIG,
        region=\'{0}\'.format(REGION),
        cluster_name=\"dataproc-cluster-{}-sit\".format(str(i[0])),
    )

但我收到以下错误：

Broken DAG: [/opt/airflow/dags/Cluster_config.py] Traceback (most recent call last):
  File \"/usr/local/lib/python3.6/site-packages/airflow/models/baseoperator.py\", line 547, in __init__
    validate_key(task_id)
  File \"/usr/local/lib/python3.6/site-packages/airflow/utils/helpers.py\", line 56, in validate_key
    \"dots and underscores exclusively\".format(k=k)
airflow.exceptions.AirflowException: The key (create_cluster_{) has to be made of alphanumeric characters, dashes, dots and underscores exclusively

所以我改变了task_id如下：

task_id=\"create_cluster_\"+re.sub(r\'\\W+\', \'\', str(list(eval(str(i)))[0])),

之后我收到以下错误：

airflow.exceptions.DuplicateTaskIdFound: Task id \'create_cluster_\' has already been added to the DAG

这让我觉得 Xcoms 中的值一次被解析一个文字，所以我使用了 render_template_as_native_obj=True, 。

但我仍然收到重复的任务 ID 错误

标签： airflow google-cloud-composer airflow-2.x

【解决方案1】：

首先，您只能在模板化字段中使用 jinja2 模板。例如，PythonOperator 包含以下模板化字段； https://github.com/apache/airflow/blob/2.4.1/airflow/operators/python.py#L130

template_fields: Sequence[str] = ('templates_dict', 'op_args', 'op_kwargs')

其次，一旦创建了 DAG，就无法根据任务的输出更改它包含的任务数。有一个例外；这是使用映射任务。最好的例子是； https://airflow.apache.org/docs/apache-airflow/2.3.0/concepts/dynamic-task-mapping.html#task-generated-mapping

@task
def make_list():
    # This can also be from an API call, checking a database, -- almost anything you like, as long as the
    # resulting list/dictionary can be stored in the current XCom backend.
    return [1, 2, {"a": "b"}, "str"]


@task
def consumer(arg):
    print(list(arg))


with DAG(dag_id="dynamic-map", start_date=datetime(2022, 4, 2)) as dag:
    consumer.expand(arg=make_list())

【讨论】：