Jon
11/01/2022, 9:53 PMTraceback (most recent call last):
File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/runner.py", line 48, in inner
new_state = method(self, state, *args, **kwargs)
File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/flow_runner.py", line 569, in get_flow_run_state
executors.prepare_upstream_states_for_mapping(
File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/utilities/executors.py", line 682, in prepare_upstream_states_for_mapping
value = upstream_state.result[i]
KeyError: 0
└── 17:51:25 | ERROR | Unexpected error occured in FlowRunner: KeyError(0)
any help what this error means?with prefect.Flow(
name=flow_config.scheduler_flow_name,
run_config=run_configs.UniversalRun(
labels=flow_config.labels,
),
terminal_state_handler=workflow_utils.workflow_terminal_state_handler,
schedule=schedule,
) as scheduler_flow:
workflow_id = register_workflow.register_workflow(
flow_config.scheduler_flow_name,
)
# fetch all consumer configs for this provider/resource_type
consumer_configs = extract_utils.fetch_consumer_configs_for_provider_resource(
flow_config.provider_code,
flow_config.resource_type,
)
# for each config, start a child flow that runs the ETL process
# <https://docs-v1.prefect.io/core/idioms/flow-to-flow.html>
mapped_flow_ids = create_flow_run.map(
parameters=dict(
consumer_code=consumer_configs,
provider_code=prefect.unmapped(flow_config.provider_code),
resource_type=prefect.unmapped(flow_config.resource_type),
file_type=prefect.unmapped(flow_config.file_type),
flow_labels=prefect.unmapped(flow_config.labels),
),
flow_name=prefect.unmapped(
EtlActionEnum.PIPELINE.value,
),
project_name=prefect.unmapped(
flow_config.global_config.project_name,
),
labels=prefect.unmapped(flow_config.labels),
)
# wait for the mapped Flows to complete before exiting
wait_for_flow_run.map(
mapped_flow_ids,
raise_final_state=prefect.unmapped(True),
)
Mason Menges
11/01/2022, 10:51 PMJon
11/01/2022, 10:52 PM