You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/06/16 06:58:18 UTC
[GitHub] [airflow] uranusjr opened a new pull request, #24489: Documentation on task mapping additions
uranusjr opened a new pull request, #24489:
URL: https://github.com/apache/airflow/pull/24489
None of these actually exist yet. Instead of implementing things and discuss what the UI should be, let’s describe the UI first (and decide if they make sense), and try to make the UI happen later.
These are the additions planned for 2.5 on AIP-42 (task mapping), namely `expand_kwargs()`, `map()`, filter (using Airflow's existing mechanism), and `zip()`. The examples for `expand_kwargs()` and `zip()` are pretty bad now. Suggestions are much welcomed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #24489:
URL: https://github.com/apache/airflow/pull/24489#issuecomment-1161367574
The PR is likely ready to be merged. No tests are needed as no important environment files, nor python files were modified by it. However, committers might decide that full test matrix is needed and add the 'full tests needed' label. Then you should rebase it to the latest main or amend the last commit of the PR, and push it with --force-with-lease.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #24489:
URL: https://github.com/apache/airflow/pull/24489#discussion_r899768327
##########
docs/apache-airflow/concepts/dynamic-task-mapping.rst:
##########
@@ -208,27 +210,156 @@ In this example you have a regular data delivery to an S3 bucket and want to app
with DAG(dag_id="mapped_s3", start_date=datetime(2020, 4, 7)) as dag:
- files = S3ListOperator(
+ list_filenames = S3ListOperator(
task_id="get_input",
bucket="example-bucket",
prefix='incoming/provider_a/{{ data_interval_start.strftime("%Y-%m-%d") }}',
)
@task
- def count_lines(aws_conn_id, bucket, file):
+ def count_lines(aws_conn_id, bucket, filename):
hook = S3Hook(aws_conn_id=aws_conn_id)
- return len(hook.read_key(file, bucket).splitlines())
+ return len(hook.read_key(filename, bucket).splitlines())
@task
def total(lines):
return sum(lines)
- counts = count_lines.partial(aws_conn_id="aws_default", bucket=files.bucket).expand(
- file=XComArg(files)
- )
+ counts = count_lines.partial(
+ aws_conn_id="aws_default", bucket=list_filenames.bucket
+ ).expand(filename=XComArg(list_filenames))
+
total(lines=counts)
+Assigning multiple parameters to a non-TaskFlow operator
+========================================================
+
+Sometimes an upstream needs to specify multiple arguments to a downstream operator. To do this, you can use the ``expand_kwargs`` function, which takes a sequence of mappings to map against.
+
+.. code-block:: python
+
+ BashOperator.partial(task_id="bash").expand_kwargs(
+ [
+ {"bash_command": "echo $ENV1", "env": {"ENV1": "1"}},
+ {"bash_command": "printf $ENV2", "env": {"ENV2": "2"}},
+ ],
+ )
+
+This produces two task instances at run-time printing ``1`` and ``2`` respectively.
+
+Similar to ``expand``, you can also map against a XCom that returns a list of dicts, or a list of XComs each returning a dict. Re-using the S3 example above, you can use a mapped task to perform “branching” and copy files to different buckets:
+
+.. code-block:: python
+
+ list_filenames = S3ListOperator(...) # Same as the above example.
+
+
+ @task
+ def create_copy_kwargs(filename):
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ dest_bucket_name = "my_text_bucket"
+ else:
+ dest_bucket_name = "my_other_bucket"
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": dest_bucket_name,
+ }
+
+
+ copy_kwargs = create_copy_kwargs.expand(filename=XComArg(list_filenames))
+
+ # Copy files to another bucket, based on the file's extension.
+ copy_filenames = S3CopyObjectOperator.partial(
+ task_id="copy_files", source_bucket_name=list_filenames.bucket
+ ).expand_kwargs(copy_kwargs)
+
+Filtering items from an expanded task
+=====================================
+
+A mapped task can remove any elements from being passed on to its downstream tasks by returning ``None``. For example, if we want to *only* copy files from an S3 bucket to another with certain extensions, we could implement ``create_copy_kwargs`` like this instead:
+
+.. code-block:: python
+
+ @task
+ def create_copy_kwargs(filename):
+ # Skip files not ending with these suffixes.
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ return None
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": "my_other_bucket",
+ }
+
+
+ # copy_kwargs and copy_files are implemented the same.
+
+This makes ``copy_files`` only expand against ``.txt``, ``.json``, and ``.yml`` files, while ignoring the rest.
Review Comment:
```suggestion
This makes ``copy_files`` only expand against ``.json`` and ``.yml`` files, while ignoring the rest.
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a diff in pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #24489:
URL: https://github.com/apache/airflow/pull/24489#discussion_r903324990
##########
docs/apache-airflow/concepts/dynamic-task-mapping.rst:
##########
@@ -208,27 +210,156 @@ In this example you have a regular data delivery to an S3 bucket and want to app
with DAG(dag_id="mapped_s3", start_date=datetime(2020, 4, 7)) as dag:
- files = S3ListOperator(
+ list_filenames = S3ListOperator(
task_id="get_input",
bucket="example-bucket",
prefix='incoming/provider_a/{{ data_interval_start.strftime("%Y-%m-%d") }}',
)
@task
- def count_lines(aws_conn_id, bucket, file):
+ def count_lines(aws_conn_id, bucket, filename):
hook = S3Hook(aws_conn_id=aws_conn_id)
- return len(hook.read_key(file, bucket).splitlines())
+ return len(hook.read_key(filename, bucket).splitlines())
@task
def total(lines):
return sum(lines)
- counts = count_lines.partial(aws_conn_id="aws_default", bucket=files.bucket).expand(
- file=XComArg(files)
- )
+ counts = count_lines.partial(
+ aws_conn_id="aws_default", bucket=list_filenames.bucket
+ ).expand(filename=XComArg(list_filenames))
+
total(lines=counts)
+Assigning multiple parameters to a non-TaskFlow operator
+========================================================
+
+Sometimes an upstream needs to specify multiple arguments to a downstream operator. To do this, you can use the ``expand_kwargs`` function, which takes a sequence of mappings to map against.
+
+.. code-block:: python
+
+ BashOperator.partial(task_id="bash").expand_kwargs(
+ [
+ {"bash_command": "echo $ENV1", "env": {"ENV1": "1"}},
+ {"bash_command": "printf $ENV2", "env": {"ENV2": "2"}},
+ ],
+ )
+
+This produces two task instances at run-time printing ``1`` and ``2`` respectively.
+
+Similar to ``expand``, you can also map against a XCom that returns a list of dicts, or a list of XComs each returning a dict. Re-using the S3 example above, you can use a mapped task to perform “branching” and copy files to different buckets:
+
+.. code-block:: python
+
+ list_filenames = S3ListOperator(...) # Same as the above example.
+
+
+ @task
+ def create_copy_kwargs(filename):
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ dest_bucket_name = "my_text_bucket"
+ else:
+ dest_bucket_name = "my_other_bucket"
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": dest_bucket_name,
+ }
+
+
+ copy_kwargs = create_copy_kwargs.expand(filename=XComArg(list_filenames))
+
+ # Copy files to another bucket, based on the file's extension.
+ copy_filenames = S3CopyObjectOperator.partial(
+ task_id="copy_files", source_bucket_name=list_filenames.bucket
+ ).expand_kwargs(copy_kwargs)
+
+Filtering items from an expanded task
+=====================================
+
+A mapped task can remove any elements from being passed on to its downstream tasks by returning ``None``. For example, if we want to *only* copy files from an S3 bucket to another with certain extensions, we could implement ``create_copy_kwargs`` like this instead:
+
+.. code-block:: python
+
+ @task
+ def create_copy_kwargs(filename):
+ # Skip files not ending with these suffixes.
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ return None
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": "my_other_bucket",
+ }
+
+
+ # copy_kwargs and copy_files are implemented the same.
+
+This makes ``copy_files`` only expand against ``.json`` and ``.yml`` files, while ignoring the rest.
+
+Transforming mapped data
+========================
+
+Since it is common to want to transform the output data format for task mapping, especially from a non-TaskFlow operator, where the output format is pre-determined and cannot be easily converted (such as ``create_copy_kwargs`` in the above example), a special ``map()`` function can be used to easily perform this kind of transformation. The above example can therefore be modified like this:
+
+.. code-block:: python
+
+ from airflow.exceptions import AirflowSkipException
+
+ filenames = S3ListOperator(...) # Unchanged.
+
+
+ def create_copy_kwargs(filename):
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ raise AirflowSkipException(f"skipping {filename!r}; unexpected suffix")
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": "my_other_bucket",
+ }
+
+
+ copy_kwargs = XComArg(list_filenames).map(create_copy_kwargs)
+
+ # Unchanged.
+ copy_filenames = S3CopyObjectOperator.partial(...).expand_kwargs(copy_kwargs)
+
+There are a couple of things to note:
+
+#. The callable argument of ``map()`` (``create_copy_kwargs`` in the example) **must not** be a task, but a plain Python function. The transformation is as a part of the “pre-processing” of the downstream task (i.e. ``copy_files``), not a standalone task in the DAG.
+#. The callable always take exactly one positional argument. This function is called for each item in the iterable used for task-mapping, similar to how Python’s built-in ``map()`` works.
+#. Since the callable is executed as a part of the downstream task, you can use any existing techniques to write the task function. To mark a component as skipped, for example, you should raise ``AirflowSkipException``. Note that returning ``None`` **does not** work here.
Review Comment:
It would set the value to None, e.g. if you have `def foo(x)` and return None in the pre-processing transform, `foo` would simply receive `x=None`. (Note that since you can’t push None to XCom—an unfortunate limitation inherited from long ago—`x` cannot be None without pre-processing.)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jedcunningham commented on pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
jedcunningham commented on PR #24489:
URL: https://github.com/apache/airflow/pull/24489#issuecomment-1169067276
Removing from the milestone and setting to draft, as we should hold this until we are ready to get it in the next minor release. Feel free to adjust this as necessary @uranusjr.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a diff in pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #24489:
URL: https://github.com/apache/airflow/pull/24489#discussion_r903324990
##########
docs/apache-airflow/concepts/dynamic-task-mapping.rst:
##########
@@ -208,27 +210,156 @@ In this example you have a regular data delivery to an S3 bucket and want to app
with DAG(dag_id="mapped_s3", start_date=datetime(2020, 4, 7)) as dag:
- files = S3ListOperator(
+ list_filenames = S3ListOperator(
task_id="get_input",
bucket="example-bucket",
prefix='incoming/provider_a/{{ data_interval_start.strftime("%Y-%m-%d") }}',
)
@task
- def count_lines(aws_conn_id, bucket, file):
+ def count_lines(aws_conn_id, bucket, filename):
hook = S3Hook(aws_conn_id=aws_conn_id)
- return len(hook.read_key(file, bucket).splitlines())
+ return len(hook.read_key(filename, bucket).splitlines())
@task
def total(lines):
return sum(lines)
- counts = count_lines.partial(aws_conn_id="aws_default", bucket=files.bucket).expand(
- file=XComArg(files)
- )
+ counts = count_lines.partial(
+ aws_conn_id="aws_default", bucket=list_filenames.bucket
+ ).expand(filename=XComArg(list_filenames))
+
total(lines=counts)
+Assigning multiple parameters to a non-TaskFlow operator
+========================================================
+
+Sometimes an upstream needs to specify multiple arguments to a downstream operator. To do this, you can use the ``expand_kwargs`` function, which takes a sequence of mappings to map against.
+
+.. code-block:: python
+
+ BashOperator.partial(task_id="bash").expand_kwargs(
+ [
+ {"bash_command": "echo $ENV1", "env": {"ENV1": "1"}},
+ {"bash_command": "printf $ENV2", "env": {"ENV2": "2"}},
+ ],
+ )
+
+This produces two task instances at run-time printing ``1`` and ``2`` respectively.
+
+Similar to ``expand``, you can also map against a XCom that returns a list of dicts, or a list of XComs each returning a dict. Re-using the S3 example above, you can use a mapped task to perform “branching” and copy files to different buckets:
+
+.. code-block:: python
+
+ list_filenames = S3ListOperator(...) # Same as the above example.
+
+
+ @task
+ def create_copy_kwargs(filename):
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ dest_bucket_name = "my_text_bucket"
+ else:
+ dest_bucket_name = "my_other_bucket"
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": dest_bucket_name,
+ }
+
+
+ copy_kwargs = create_copy_kwargs.expand(filename=XComArg(list_filenames))
+
+ # Copy files to another bucket, based on the file's extension.
+ copy_filenames = S3CopyObjectOperator.partial(
+ task_id="copy_files", source_bucket_name=list_filenames.bucket
+ ).expand_kwargs(copy_kwargs)
+
+Filtering items from an expanded task
+=====================================
+
+A mapped task can remove any elements from being passed on to its downstream tasks by returning ``None``. For example, if we want to *only* copy files from an S3 bucket to another with certain extensions, we could implement ``create_copy_kwargs`` like this instead:
+
+.. code-block:: python
+
+ @task
+ def create_copy_kwargs(filename):
+ # Skip files not ending with these suffixes.
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ return None
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": "my_other_bucket",
+ }
+
+
+ # copy_kwargs and copy_files are implemented the same.
+
+This makes ``copy_files`` only expand against ``.json`` and ``.yml`` files, while ignoring the rest.
+
+Transforming mapped data
+========================
+
+Since it is common to want to transform the output data format for task mapping, especially from a non-TaskFlow operator, where the output format is pre-determined and cannot be easily converted (such as ``create_copy_kwargs`` in the above example), a special ``map()`` function can be used to easily perform this kind of transformation. The above example can therefore be modified like this:
+
+.. code-block:: python
+
+ from airflow.exceptions import AirflowSkipException
+
+ filenames = S3ListOperator(...) # Unchanged.
+
+
+ def create_copy_kwargs(filename):
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ raise AirflowSkipException(f"skipping {filename!r}; unexpected suffix")
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": "my_other_bucket",
+ }
+
+
+ copy_kwargs = XComArg(list_filenames).map(create_copy_kwargs)
+
+ # Unchanged.
+ copy_filenames = S3CopyObjectOperator.partial(...).expand_kwargs(copy_kwargs)
+
+There are a couple of things to note:
+
+#. The callable argument of ``map()`` (``create_copy_kwargs`` in the example) **must not** be a task, but a plain Python function. The transformation is as a part of the “pre-processing” of the downstream task (i.e. ``copy_files``), not a standalone task in the DAG.
+#. The callable always take exactly one positional argument. This function is called for each item in the iterable used for task-mapping, similar to how Python’s built-in ``map()`` works.
+#. Since the callable is executed as a part of the downstream task, you can use any existing techniques to write the task function. To mark a component as skipped, for example, you should raise ``AirflowSkipException``. Note that returning ``None`` **does not** work here.
Review Comment:
It would set the value to None, e.g. if you have `def foo(x)` and return None in the pre-processing transform, `foo` would simply receive `x=None`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
uranusjr commented on PR #24489:
URL: https://github.com/apache/airflow/pull/24489#issuecomment-1204877898
Now that we’ve got things implemented[^1], this can be merged.
[^1]: I did miss one thing—checking `map()` does not receive a `@task`—so I added it in this PR.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #24489:
URL: https://github.com/apache/airflow/pull/24489#issuecomment-1205430706
Some whitespace :scream_cat:
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #24489:
URL: https://github.com/apache/airflow/pull/24489#discussion_r902880469
##########
docs/apache-airflow/concepts/dynamic-task-mapping.rst:
##########
@@ -208,27 +210,156 @@ In this example you have a regular data delivery to an S3 bucket and want to app
with DAG(dag_id="mapped_s3", start_date=datetime(2020, 4, 7)) as dag:
- files = S3ListOperator(
+ list_filenames = S3ListOperator(
task_id="get_input",
bucket="example-bucket",
prefix='incoming/provider_a/{{ data_interval_start.strftime("%Y-%m-%d") }}',
)
@task
- def count_lines(aws_conn_id, bucket, file):
+ def count_lines(aws_conn_id, bucket, filename):
hook = S3Hook(aws_conn_id=aws_conn_id)
- return len(hook.read_key(file, bucket).splitlines())
+ return len(hook.read_key(filename, bucket).splitlines())
@task
def total(lines):
return sum(lines)
- counts = count_lines.partial(aws_conn_id="aws_default", bucket=files.bucket).expand(
- file=XComArg(files)
- )
+ counts = count_lines.partial(
+ aws_conn_id="aws_default", bucket=list_filenames.bucket
+ ).expand(filename=XComArg(list_filenames))
+
total(lines=counts)
+Assigning multiple parameters to a non-TaskFlow operator
+========================================================
+
+Sometimes an upstream needs to specify multiple arguments to a downstream operator. To do this, you can use the ``expand_kwargs`` function, which takes a sequence of mappings to map against.
+
+.. code-block:: python
+
+ BashOperator.partial(task_id="bash").expand_kwargs(
+ [
+ {"bash_command": "echo $ENV1", "env": {"ENV1": "1"}},
+ {"bash_command": "printf $ENV2", "env": {"ENV2": "2"}},
+ ],
+ )
+
+This produces two task instances at run-time printing ``1`` and ``2`` respectively.
+
+Similar to ``expand``, you can also map against a XCom that returns a list of dicts, or a list of XComs each returning a dict. Re-using the S3 example above, you can use a mapped task to perform “branching” and copy files to different buckets:
+
+.. code-block:: python
+
+ list_filenames = S3ListOperator(...) # Same as the above example.
+
+
+ @task
+ def create_copy_kwargs(filename):
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ dest_bucket_name = "my_text_bucket"
+ else:
+ dest_bucket_name = "my_other_bucket"
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": dest_bucket_name,
+ }
+
+
+ copy_kwargs = create_copy_kwargs.expand(filename=XComArg(list_filenames))
+
+ # Copy files to another bucket, based on the file's extension.
+ copy_filenames = S3CopyObjectOperator.partial(
+ task_id="copy_files", source_bucket_name=list_filenames.bucket
+ ).expand_kwargs(copy_kwargs)
+
+Filtering items from an expanded task
+=====================================
+
+A mapped task can remove any elements from being passed on to its downstream tasks by returning ``None``. For example, if we want to *only* copy files from an S3 bucket to another with certain extensions, we could implement ``create_copy_kwargs`` like this instead:
+
+.. code-block:: python
+
+ @task
+ def create_copy_kwargs(filename):
+ # Skip files not ending with these suffixes.
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ return None
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": "my_other_bucket",
+ }
+
+
+ # copy_kwargs and copy_files are implemented the same.
+
+This makes ``copy_files`` only expand against ``.json`` and ``.yml`` files, while ignoring the rest.
+
+Transforming mapped data
+========================
+
+Since it is common to want to transform the output data format for task mapping, especially from a non-TaskFlow operator, where the output format is pre-determined and cannot be easily converted (such as ``create_copy_kwargs`` in the above example), a special ``map()`` function can be used to easily perform this kind of transformation. The above example can therefore be modified like this:
+
+.. code-block:: python
+
+ from airflow.exceptions import AirflowSkipException
+
+ filenames = S3ListOperator(...) # Unchanged.
+
+
+ def create_copy_kwargs(filename):
+ if filename.rsplit(".", 1)[-1] not in ("json", "yml"):
+ raise AirflowSkipException(f"skipping {filename!r}; unexpected suffix")
+ return {
+ "source_bucket_key": filename,
+ "dest_bucket_key": filename,
+ "dest_bucket_name": "my_other_bucket",
+ }
+
+
+ copy_kwargs = XComArg(list_filenames).map(create_copy_kwargs)
+
+ # Unchanged.
+ copy_filenames = S3CopyObjectOperator.partial(...).expand_kwargs(copy_kwargs)
+
+There are a couple of things to note:
+
+#. The callable argument of ``map()`` (``create_copy_kwargs`` in the example) **must not** be a task, but a plain Python function. The transformation is as a part of the “pre-processing” of the downstream task (i.e. ``copy_files``), not a standalone task in the DAG.
+#. The callable always take exactly one positional argument. This function is called for each item in the iterable used for task-mapping, similar to how Python’s built-in ``map()`` works.
+#. Since the callable is executed as a part of the downstream task, you can use any existing techniques to write the task function. To mark a component as skipped, for example, you should raise ``AirflowSkipException``. Note that returning ``None`` **does not** work here.
Review Comment:
what happens when you return None?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk merged pull request #24489: Documentation on task mapping additions
Posted by GitBox <gi...@apache.org>.
potiuk merged PR #24489:
URL: https://github.com/apache/airflow/pull/24489
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org