Braun Reyes
12/13/2019, 3:41 PM2019-12-13 1:40am prefect.CloudFlowRunner INFO Beginning Flow run for 'dbt_run'
2019-12-13 1:40am prefect.CloudFlowRunner INFO Starting flow run.
2019-12-13 1:40am prefect.CloudFlowRunner ERROR Unexpected error: TypeError("start() missing 1 required positional argument: 'self'")
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
new_state = method(self, state, *args, **kwargs)
File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 400, in get_flow_run_state
with executor.start():
File "/usr/local/lib/python3.7/contextlib.py", line 239, in helper
return _GeneratorContextManager(func, args, kwds)
File "/usr/local/lib/python3.7/contextlib.py", line 82, in __init__
self.gen = func(*args, **kwds)
TypeError: start() missing 1 required positional argument: 'self'
This was a fargate task started by the fargatetaskenvironmentjosh
12/13/2019, 3:46 PMyour_executor.start()
instead of your_executor().start()
. As in the executor isn’t getting initializedBraun Reyes
12/13/2019, 3:55 PM# Load serialized flow from file and run it with a DaskExecutor
flow = storage.get_flow(flow_location)
with set_temporary_config({"engine.executor.default_class": self.executor}):
executor = get_default_executor_class()
executor = executor(**self.executor_kwargs)
runner_cls = get_default_flow_runner_class()
runner_cls(flow=flow).run(executor=executor)
fargatetask:
# Load serialized flow from file and run it with the executor
with open(
prefect.context.get(
"flow_file_path", "/root/.prefect/flow_env.prefect"
),
"rb",
) as f:
flow = cloudpickle.load(f)
runner_cls = get_default_flow_runner_class()
executor_cls = get_default_executor_class()
runner_cls(flow=flow).run(executor=executor_cls)
executor = executor(**self.executor_kwargs)
josh
12/13/2019, 4:03 PMexecutor=executor_cls()
Braun Reyes
12/13/2019, 4:04 PMexecutor_cls = get_default_executor_class()()
josh
12/13/2019, 4:04 PMBraun Reyes
12/13/2019, 4:04 PMjosh
12/13/2019, 4:06 PMBraun Reyes
12/13/2019, 5:13 PMenable_task_revisions
for the fargate agent and fargate task environment that will leverage task definition revisions instead of creating a new task definition family every timejosh
12/13/2019, 5:16 PMBraun Reyes
12/13/2019, 5:16 PMjosh
12/13/2019, 5:19 PM