python - Airflow GKEPodOperator xcom_push 返回 None

Question

所以我已经为此苦苦挣扎了几个小时。这是我的运营商的代码:

task1 = GKEPodOperator(
        task_id="task1",
        project_id="proj",
        location="location",
        cluster_name="cluster_name",
        name="cluster-calculator",
        namespace="default",
        image=Variable.get("cluster_calculator_image"),
        arguments=['--name clustercalculator'],
        env_vars=env_vars,
        xcom_push=True,
        is_delete_operator_pod=True,
        get_logs=True,
        dag=dag
    )

该 Pod 运行一个简单的 docker 容器，其中包含 Java 应用程序，执行一些操作并将结果写入默认值 /airflow/xcom/result.json 文件。

这是我尝试获取 xcom_push 结果的方式:

def print_xcom_result(*op_args, **kwargs):
        print(op_args)
        print(kwargs['task_instance'].xcom_pull(task_ids='task1'))

    test_values = PythonOperator(
        task_id="task1_test",
        python_callable=print_xcom_result,
        provide_context=True,
        op_args=["{{task_instance.xcom_pull(task_ids='task1')}}"],
        dag=dag
    )

无论我尝试什么，它总是打印无。

[2019-10-12 00:06:23,061] {{logging_mixin.py:95}} INFO - ('None',)
[2019-10-12 00:06:23,072] {{logging_mixin.py:95}} INFO - None

当我在 Airflow UI 上访问 XCOM 时，它什么也没显示。
我也试过这里的例子: Failed to extract xcom from airflow pod - Kubernetes Pod Operator它也不起作用。

Sidecar 容器已确定创建，我在日志中看到了它的输出:

Running command... [1mcat /airflow/xcom/return.json[0m
Running command... [1mkill -s SIGINT 1[0m
INFO[0m - {"clusterSize":2}[0m

我什至尝试在外部运行 docker 容器，验证结果是否正确写入 xcom 目录，但在 DAG 执行期间没有运气获得此结果。

Airflow 版本是最新的。 python 3.7

如果这很重要，我有 6 个容器运行 Airflow(webserver、flower、worker、scheduler、postgre、rabbitmq)。 Celery 是执行者。 Pod 在 Google Cloud 上的 Kubernetes Engine 中运行。

没有错误，两个运算符(operator)都成功了。

有没有人有任何想法？先感谢您。

Answer 1

Airflow 中存在一个错误，即 GKEPodOperator 的执行没有返回语句

super(GKEPodOperator, self).execute(context)

应该

return super(GKEPodOperator, self).execute(context)

该错误已在 Airflow 2.0 ( https://issues.apache.org/jira/browse/AIRFLOW-4072 ) 中修复

一个建议可能是创建一个插件 YourOwnGKEOperator 来在本地解决这个问题。 (并将其部署到您的插件文件夹中)