A closer look at Airflow’s KubernetesPodOperator and XCom

The KubernetesPodOperator handles communicating XCom values differently than other operators. The basics are described in the operator documentation under the xcom_push parameter. I’ve written up a more detailed example that expands on that documentation.

An Airflow task instance described by the KubernetesPodOperator can write a dict to the file /airflow/xcom/return.json (always the same file) that will be read and turned into an XCom value by a airflow-xcom-sidecar sidecar container. This sidecar gets created when the KubernetesPodOperator parameter xcom_push is True.

For instance, if a KubernetesPodOperator task instance executes the following Python code in a Docker container example_image:

import json

xcom_return = {"key1": value1, "key2": value2}

with open("/airflow/xcom/return.json", "w") as file:
 json.dump(xcom_return, file)

This is the KubernetesPodOperator task instance:

from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator


example_kubernetes_task = KubernetesPodOperator(
 namespace="airflow",
 name="exampletask",
 task_id="example_kubernetes_task",
 image="docker_repo/example_image",
 xcom_push=True,
 ...
)

The XCom values will be made available to downstream task instances under the ‘return_value’ key:

xcom_data = context['ti'].xcom_pull(task_ids='k8s_test_task',
                                    key='return_value')
print(xcom_data["key1"]) # value1
print(xcom_data["key2"]) # value2

The default for xcom_pull‘s key parameter is ‘return_value’, so key is an optional parameter in this example.

XCom values can also be pulled using Jinja templates in operator parameters that support templates, which are listed in operator documentation. Here is the list of parameters for the KubernetesPodOperator, and also for the PythonOperator.

This is how to get the value of ‘key1’ in a Jinja template:

{{ task_instance.xcom_pull(task_ids='example_kubernetes_task', key='return_value')['key1'] }}

A useful way to use XCom values with templates is to pass arguments to a Docker container:

from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator 


another_kubernetes_task = KubernetesPodOperator(
 namespace="airflow",
 name="anotherexampletask",
 task_id="another_kubernetes_task",
 image="docker_repo/another_example_image",
 arguments=["--myarg",
  "{{ task_instance.xcom_pull(task_ids='example_kubernetes_task', key='return_value')['key1'] }}"
 ],
 ...
)