dheerajturaga commented on code in PR #71808:
URL: https://github.com/apache/airflow/pull/71808#discussion_r3889943205
##########
providers/standard/docs/operators/bash.rst:
##########
@@ -226,6 +226,58 @@ Here's how you can use the result_processor with the
BashOperator:
)
+Multiple XCom outputs
+---------------------
+
+Pair ``output_processor`` with ``multiple_outputs=True`` to push more than one
XCom from a single task. When
+the processed output is a dictionary, each key is pushed as its own XCom,
which lets downstream tasks pull
+individual values by name instead of pulling the whole dictionary and indexing
into it.
+
+.. tab-set::
+
+ .. tab-item:: @task.bash
+ :sync: taskflow
+
+ .. exampleinclude::
/../src/airflow/providers/standard/example_dags/example_bash_decorator.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_decorator_bash_multiple_outputs]
+ :end-before: [END howto_decorator_bash_multiple_outputs]
+
+ .. tab-item:: BashOperator
+ :sync: operator
+
+ .. exampleinclude::
/../src/airflow/providers/standard/example_dags/example_bash_operator.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_bash_multiple_outputs]
+ :end-before: [END howto_operator_bash_multiple_outputs]
+
+The producing task above pushes an XCom for ``dag_folder`` and one for
``file_count``. The full dictionary is
+*also* pushed as the task's return value, so ``{{
ti.xcom_pull(task_ids="describe_dag_folder") }}`` still
+resolves to ``{"dag_folder": ..., "file_count": ...}``.
+
+.. important::
+
+ Only the **last line** written by the command is captured, so the
dictionary must be the final thing the
+ command emits. A few consequences worth designing around:
Review Comment:
Good question — it's worth spelling out, because the naming does suggest
otherwise. `output_processor` receives only the **last** line, not the whole
stream.
The capture happens in `SubprocessHook.run_command`, which loops over stdout
rebinding a single `line` variable and returns only whatever it holds when the
loop ends:
```python
line = ""
...
for raw_line in iter(self.sub_process.stdout.readline, b""):
line = raw_line.decode(output_encoding,
errors="backslashreplace").rstrip()
self.log.info("%s", line)
...
return SubprocessResult(exit_code=return_code, output=line)
```
—
[`hooks/subprocess.py:93-106`](https://github.com/apache/airflow/blob/ffaeb1696768743939f1ca15ad3b329f0c8d6c4b/providers/standard/src/airflow/providers/standard/hooks/subprocess.py#L93-L106).
Every line is logged (that's the `self.log.info` inside the loop, which is
probably what makes it look like the full output is retained), but only the
last one is returned.
The method's own docstring states it: *":return: `namedtuple` containing
`exit_code` and `output`, the last line from stderr or stdout"*
([`subprocess.py:75-76`](https://github.com/apache/airflow/blob/ffaeb1696768743939f1ca15ad3b329f0c8d6c4b/providers/standard/src/airflow/providers/standard/hooks/subprocess.py#L75-L76)).
`BashOperator.execute` then passes exactly that value through: `return
self.output_processor(result.output)`
([`operators/bash.py:229`](https://github.com/apache/airflow/blob/ffaeb1696768743939f1ca15ad3b329f0c8d6c4b/providers/standard/src/airflow/providers/standard/operators/bash.py#L229)).
There's also an existing test pinning it —
[`test_subprocess.py::test_return_value`](https://github.com/apache/airflow/blob/ffaeb1696768743939f1ca15ad3b329f0c8d6c4b/providers/standard/tests/unit/standard/hooks/test_subprocess.py#L68-L80)
asserts that `"test-val\ntest-val\n"` yields `""` and `"test-val\ntest-val"`
yields `"test-val"`.
You're right that the *rest* of the pipeline is as you describe: the
`output_processor` return value is what the task returns, and that's the input
for pushing XComs. The constraint is purely upstream of the processor.
---
Drafted-by: Claude Code (Opus 5); reviewed by @dheerajturaga before posting
##########
providers/standard/docs/operators/bash.rst:
##########
@@ -226,6 +226,58 @@ Here's how you can use the result_processor with the
BashOperator:
)
+Multiple XCom outputs
+---------------------
+
+Pair ``output_processor`` with ``multiple_outputs=True`` to push more than one
XCom from a single task. When
+the processed output is a dictionary, each key is pushed as its own XCom,
which lets downstream tasks pull
+individual values by name instead of pulling the whole dictionary and indexing
into it.
+
+.. tab-set::
+
+ .. tab-item:: @task.bash
+ :sync: taskflow
+
+ .. exampleinclude::
/../src/airflow/providers/standard/example_dags/example_bash_decorator.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_decorator_bash_multiple_outputs]
+ :end-before: [END howto_decorator_bash_multiple_outputs]
+
+ .. tab-item:: BashOperator
+ :sync: operator
+
+ .. exampleinclude::
/../src/airflow/providers/standard/example_dags/example_bash_operator.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_bash_multiple_outputs]
+ :end-before: [END howto_operator_bash_multiple_outputs]
+
+The producing task above pushes an XCom for ``dag_folder`` and one for
``file_count``. The full dictionary is
+*also* pushed as the task's return value, so ``{{
ti.xcom_pull(task_ids="describe_dag_folder") }}`` still
+resolves to ``{"dag_folder": ..., "file_count": ...}``.
+
+.. important::
+
+ Only the **last line** written by the command is captured, so the
dictionary must be the final thing the
+ command emits. A few consequences worth designing around:
+
+ * A trailing ``echo`` with no arguments emits an empty line, which becomes
the captured output instead of
+ your dictionary.
+ * ``stderr`` is merged into ``stdout``, so a subcommand that writes to
``stderr`` last will overwrite the
+ captured value. Redirect noisy subcommands (for example ``2>/dev/null``)
to avoid this.
+ * The dictionary must fit on a single line. Use ``jq -c`` rather than
pretty-printed output, and prefer
+ ``printf`` over a multi-line ``printf`` format string.
Review Comment:
Answered in the thread above — the bullets follow from
`SubprocessHook.run_command` returning only the last line.
Mapping each one to the code, for completeness:
* **Trailing empty `echo`** — the loop's last iteration binds `line` to
`""`, which is exactly the `("test-val\ntest-val\n", "")` case in
[`test_return_value`](https://github.com/apache/airflow/blob/ffaeb1696768743939f1ca15ad3b329f0c8d6c4b/providers/standard/tests/unit/standard/hooks/test_subprocess.py#L68-L80).
* **`stderr` overwriting the value** — the subprocess is spawned with
`stderr=STDOUT`
([`subprocess.py:85`](https://github.com/apache/airflow/blob/ffaeb1696768743939f1ca15ad3b329f0c8d6c4b/providers/standard/src/airflow/providers/standard/hooks/subprocess.py#L85)),
so stderr writes interleave into the same stream the loop reads and can land
last.
* **Single-line requirement** — a pretty-printed dict spans several
`readline` iterations, so the processor would only ever see its closing `}`.
Happy to reword if any of these read as more alarming than they need to be,
but I'd rather keep them: the failure mode is silent (you get `""` or a stray
line, not an error) and it bit me while writing the example Dag.
---
Drafted-by: Claude Code (Opus 5); reviewed by @dheerajturaga before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
