### 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)

"on_retry_callback": None,
            "on_success_callback": None,
            "outlets": [],
            "owner": "airflow",
            "params": {},
            "pool": "default_pool",
            "pool_slots": 1,
            "priority_weight": 1,
            "queue": "default",
            "resources": None,
            "retries": 0,
            "retry_delay": timedelta(0, 300),
            "retry_exponential_backoff": False,
            "run_as_user": None,
            "sla": None,
            "task_id": "10",
            "trigger_rule": "all_success",
            "wait_for_downstream": False,
            "wait_for_past_depends_before_skipping": False,
            "weight_rule": "downstream",
        }, """
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
     ACTION NEEDED! PLEASE READ THIS CAREFULLY AND CORRECT TESTS CAREFULLY
 Some fields were added to the BaseOperator! Please add them to the list above and make sure that
 you add support for DAG serialization - you should add the field to
 `airflow/serialization/schema.json` - they should have correct type defined there.
 Note that we do not support versioning yet so you should only add optional fields to BaseOperator.
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
                         """
    def test_operator_deserialize_old_names(self):