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

assert getattr(simple_task, "bash_command") == bash_command
        
        
        
        
        assert serialized_dag["dag"]["tasks"][0]["_operator_extra_links"] == serialized_links
        
        assert set(simple_task.extra_links) == {*links, "airflow", "github", "google"}
        dr = dag_maker.create_dagrun(execution_date=test_date)
        (ti,) = dr.task_instances
        XCom.set(
            key="search_query",
            value=bash_command,
            task_id=simple_task.task_id,
            dag_id=simple_task.dag_id,
            run_id=dr.run_id,
        )
        
        for name, expected in links.items():
            link = simple_task.get_extra_links(ti, name)
            assert link == expected
        
        link = simple_task.get_extra_links(ti, GoogleLink.name)
        assert "https://www.google.com" == link
    def test_extra_operator_links_logs_error_for_non_registered_extra_links(self, caplog):
        """
        Assert OperatorLinks not registered via Plugins and if it is not an inbuilt Operator Link,
        it can still deserialize the DAG (does not error) but just logs an error
        """
        class TaskStateLink(BaseOperatorLink):