### What do you see as an issue?


The [documentation for Dynamic Task Mapping](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/dynamic-task-mapping.html#simple-mapping
) does not include an example of a "reduce" task (e.g. `sum_it` in the [examples](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/dynamic-task-mapping.html#simple-mapping)) using the classic (or non-TaskFlow) operators. It only includes an example that uses the TaskFlow operators.

When I attempted to write a "reduce" task using classic operators for my DAG, I found that there wasn't an obvious approach.

### Solving the problem

We should add an example of a "reduce" task that uses the classic (non-TaskFlow) operators.

For example, for the given `sum_it` example:

```
"""Example DAG demonstrating the usage of dynamic task mapping reduce using classic
operators.
"""
from __future__ import annotations

from datetime import datetime

from airflow import DAG
from airflow.decorators import task
from airflow.operators.python import PythonOperator


def add_one(x: int):
    return x + 1

def sum_it(values):
    total = sum(values)
    print(f"Total was {total}")


with DAG(dag_id="example_dynamic_task_mapping_reduce", start_date=datetime(2022, 3, 4)):
    add_one_task = PythonOperator.partial(
        task_id="add_one",
        python_callable=add_one,
    ).expand(
        op_kwargs=[
            {"x": 1},
            {"x": 2},
            {"x": 3},
        ]
    )

    sum_it_task = PythonOperator(
        task_id="sum_it",
        python_callable=sum_it,
        op_kwargs={"values": add_one_task.output},
    )
```

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

"""Verify serialization fails without timetable registration."""
        dag = get_timetable_based_simple_dag(CustomSerializationTimetable("bar"))
        with pytest.raises(SerializationError) as ctx:
            SerializedDAG.to_dict(dag)
        message = (
            "Failed to serialize DAG 'simple_dag': Timetable class "
            "'tests.test_utils.timetables.CustomSerializationTimetable' "
            "is not registered or "
            "you have a top level database access that disrupted the session. "
            "Please check the airflow best practices documentation."
        )
        assert str(ctx.value) == message
    def validate_serialized_dag(self, json_dag, ground_truth_dag):
        """Verify serialized DAGs match the ground truth."""
        assert json_dag["dag"]["fileloc"].split("/")[-1] == "test_dag_serialization.py"
        json_dag["dag"]["fileloc"] = None
        def sorted_serialized_dag(dag_dict: dict):
            """
            Sorts the "tasks" list and "access_control" permissions in the
            serialised dag python dictionary. This is needed as the order of
            items should not matter but assertEqual would fail if the order of
            items changes in the dag dictionary
            """
            dag_dict["dag"]["tasks"] = sorted(dag_dict["dag"]["tasks"], key=lambda x: sorted(x.keys()))
            dag_dict["dag"]["_access_control"]["__var"]["test_role"]["__var"] = sorted(
                dag_dict["dag"]["_access_control"]["__var"]["test_role"]["__var"]
            )
            return dag_dict
        assert sorted_serialized_dag(ground_truth_dag) == sorted_serialized_dag(json_dag)
    def test_deserialization_across_process(self):