无法从气流吊舱中提取xcom

本文介绍了无法从气流吊舱中提取xcom-Kubernetes Pod Operator的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

在运行DAG时，该DAG使用docker映像运行jar， 给出了 xcom_push = True ，它可以在一个容器中创建另一个容器以及Docker镜像.

While running a DAG which runs a jar using a docker image, xcom_push=True is given which creates another container along with the docker image in a single pod.

DAG:

jar_task = KubernetesPodOperator( namespace='test', image="path to image", image_pull_secrets="secret", image_pull_policy="Always", node_selectors={"d-type":"na-node-group"}, cmds=["sh","-c",..~running jar here~..], secrets=[secret_file], env_vars=environment_vars, labels={"k8s-app": "airflow"}, name="airflow-pod", config_file=k8s_config_file, resources=pod.Resources(request_cpu=0.2,limit_cpu=0.5,request_memory='512Mi',limit_memory='1536Mi'), in_cluster=False, task_id="run_jar", is_delete_operator_pod=True, get_logs=True, xcom_push=True, dag=dag)

这是成功执行JAR时的错误.

Here are the errors when the JAR is executed successfully..

[2018-11-27 11:37:21,605] {{logging_mixin.py:95}} INFO - [2018-11-27 11:37:21,605] {{pod_launcher.py:166}} INFO - Running command... cat /airflow/xcom/return.json [2018-11-27 11:37:21,605] {{logging_mixin.py:95}} INFO - [2018-11-27 11:37:21,647] {{logging_mixin.py:95}} INFO - [2018-11-27 11:37:21,646] {{pod_launcher.py:173}} INFO - cat: can't open '/airflow/xcom/return.json': No such file or directory [2018-11-27 11:37:21,647] {{logging_mixin.py:95}} INFO - [2018-11-27 11:37:21,647] {{logging_mixin.py:95}} INFO - [2018-11-27 11:37:21,647] {{pod_launcher.py:166}} INFO - Running command... kill -s SIGINT 1 [2018-11-27 11:37:21,647] {{logging_mixin.py:95}} INFO - [2018-11-27 11:37:21,702] {{models.py:1760}} ERROR - Pod Launching failed: Failed to extract xcom from pod: airflow-pod-hippogriff-a4628b12 Traceback (most recent call last): File "/usr/local/airflow/operators/kubernetes_pod_operator.py", line 126, in execute get_logs=self.get_logs) File "/usr/local/airflow/operators/pod_launcher.py", line 90, in run_pod return self._monitor_pod(pod, get_logs) File "/usr/local/airflow/operators/pod_launcher.py", line 110, in _monitor_pod result = self._extract_xcom(pod) File "/usr/local/airflow/operators/pod_launcher.py", line 161, in _extract_xcom raise AirflowException('Failed to extract xcom from pod: {}'.format(pod.name)) airflow.exceptions.AirflowException: Failed to extract xcom from pod: airflow-pod-hippogriff-a4628b12 During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.6/site-packages/airflow/models.py", line 1659, in _run_raw_task result = task_copy.execute(context=context) File "/usr/local/airflow/operators/kubernetes_pod_operator.py", line 138, in execute raise AirflowException('Pod Launching failed: {error}'.format(error=ex)) airflow.exceptions.AirflowException: Pod Launching failed: Failed to extract xcom from pod: airflow-pod-hippogriff-a4628b12 [2018-11-27 11:37:21,704] {{models.py:1789}} INFO - All retries failed; marking task as FAILED

推荐答案

如果xcom_push为True，则KubernetesPodOperator在Pod中与基础容器(实际工作容器)一起再创建一个sidecar容器(airflow-xcom-sidecar). 该sidecar容器从/airflow/xcom/return.json读取数据并作为xcom值返回. 因此，您需要在基本容器中将要返回的数据写入/airflow/xcom/return.json文件.

If xcom_push is True then KubernetesPodOperator creates one more sidecar container (airflow-xcom-sidecar) in Pod along with the base container(actual worker container). This sidecar container reads data from /airflow/xcom/return.json and returns as xcom value. So in your base container you need to write the data you want to return in /airflow/xcom/return.json file.