python之无法从 airflow pod 中提取 xcom之Kubernetes Pod Operator

在运行使用 docker 镜像运行 jar 的 DAG 时，
xcom_push=True 已给出，它会在单个 pod 中创建另一个容器以及 docker 镜像。

有向无环图:

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 执行成功后出现以下错误。

    [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) 以及基础容器 (实际 worker 容器)。 此 sidecar 容器从 /airflow/xcom/return.json 读取数据并作为 xcom 值返回。 因此，在您的基础容器中，您需要在 /airflow/xcom/return.json 文件中写入要返回的数据。