所以我已经为此苦苦挣扎了几个小时。这是我的运营商的代码:
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)都成功了。
有没有人有任何想法?先感谢您。
最佳答案
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 来在本地解决这个问题。 (并将其部署到您的插件文件夹中)
关于python - Airflow GKEPodOperator xcom_push 返回 None,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/58349627/