Hi - I am having a hard time understanding how to implement a mapped task with a target. Running locally with an S3Result at the flow level works without any special configuration. When using Docker Storage an error occurs :

Traceback (most recent call last):
  File "/opt/prefect/healthcheck.py", line 136, in <module>
    result_check(flows)
  File "/opt/prefect/healthcheck.py", line 64, in result_check
    _check_mapped_result_templates(flow)
  File "/opt/prefect/healthcheck.py", line 58, in _check_mapped_result_templates
    "Mapped tasks with custom result locations must include {filename} as a template in their location - see <https://docs.prefect.io/core/advanced_tutorials/using-results.html#specifying-a-location-for-mapped-or-looped-tasks>"
ValueError: Mapped tasks with custom result locations must include {filename} as a template in their location - see <https://docs.prefect.io/core/advanced_tutorials/using-results.html#specifying-a-location-for-mapped-or-looped-tasks>

The documentation says "When configuring results for a mapped pipeline, if you choose to configure the location it is required that you include `{filename}`". Is filename a context (I couldn't find it)? https://docs.prefect.io/core/advanced_tutorials/using-results.html#mapping In short how does a configuration for a mapped task look like? With no result configured on the task level. Or is that required?

Hi @itay livni - thanks for the question. Let me check this one for you.

@Jenny So when I run the example given I get a

KeyError: 'filename

. I looked for filename in the Context documentation but could not find it

[2020-05-22 19:46:07] INFO - prefect.FlowRunner | Beginning Flow run for 'blah'
[2020-05-22 19:46:07] INFO - prefect.FlowRunner | Starting flow run.
[2020-05-22 19:46:07] INFO - prefect.TaskRunner | Task 'return_list': Starting task run...
[2020-05-22 19:46:07] INFO - prefect.TaskRunner | Task 'return_list': finished task run for task with final state: 'Cached'
[2020-05-22 19:46:07] INFO - prefect.TaskRunner | Task 'mapped_task': Starting task run...
[2020-05-22 19:46:07] INFO - prefect.TaskRunner | Task 'mapped_task[0]': Starting task run...
[2020-05-22 19:46:07] ERROR - prefect.TaskRunner | Unexpected error: KeyError('filename')
Traceback (most recent call last):
  File "miniconda3/envs/py37moc/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "miniconda3/envs/py37moc/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 651, in check_target
    if result.exists(target, **prefect.context):
  File "miniconda3/envs/py37moc/lib/python3.7/site-packages/prefect/engine/results/local_result.py", line 123, in exists
    return os.path.exists(os.path.join(self.dir, location.format(**kwargs)))
KeyError: 'filename'
[2020-05-22 19:46:07] INFO - prefect.TaskRunner | Task 'mapped_task[0]': finished task run for task with final state: 'Failed'

Here is minimal example to reproduce (I think)

from prefect import task, Flow
from prefect.engine.results import LocalResult

@task(target="a_specific_file.txt")
def return_list():
    return [1, 2, 3]
@task(target="{task_name}/{filename}/{map_index}.prefect")
def mapped_task(x):
    return x + 1
with Flow("blah", result=LocalResult(location="/home/ilivni/prefect_guide/results/{flow_name}")) as flow:
    mapped_task.map(return_list)


st = flow.run()
flow.visualize(flow_state=st)

Hey @itay livni - this is a good find; I don’t think we populate the

filename

attribute for local runs. I’ll use this thread to open an issue to resolve that and possibly clarify this situation (mapping + targets)

@Chris White But without {filename} and with {map_index} local runs work fine

That does make sense! Thanks for pointing it out to us.

FWIW: My preference would be to use {map_index} or {file_name} to differentiate

@task(
    target="{flow_name}/{task_name}/{map_index}",