What should be done to register a flow to prefect ...
# prefect-uid
David Yang
12/09/2021, 8:02 PMWhat should be done to register a flow to prefect cloud? I got error "Malformed response received from Cloud - please ensure that you are authenticated".
k
Kevin Kho
12/09/2021, 8:02 PM
Hi @David Yang, it sounds like you need to make an API key and authenticate with
prefect auth login --key API_KEYd
David Yang
12/09/2021, 8:10 PMNice. It works. Thanks a lot!
👍 1
David Yang
12/09/2021, 8:18 PMBTW, how to run tasks in parallel? I would like these two "EL data source" run in parallel and then run "load_snowflake" task. although I set the dependencies, It seems that it runs extract_DS1 first and then extract_DS2, then load_snowflake.
@task
def extract_DS1():
pass
@task
def extract_DS2():
pass
@task
def load_snowflake(Source1,Source2):
pass
@task
def notify(message):
pass
# Configure extra environment variables for this flow,
# and set a custom working directory
with Flow(
"DavidEDWFlow"
) as flow:
flow.run_config = LocalRun(
env={"DEV": "1"},
working_dir="C:/Development/prefect-poc"
)
rs = extract_DS1()
ebs = extract_DS2()
loadtwo = load_snowflake(rs,ebs)
notification = notify("done")
# flow.set_dependencies(
# task=loadtwo,
# upstream_tasks=[rs,ebs])
flow.set_dependencies(
task=notification,
upstream_tasks=[loadtwo])
flow.register(project_name="EDW")
k
Kevin Kho
12/09/2021, 8:20 PM
The default executor is a Sequential Executor. To get parallelism, you need to use the
LocalDaskExecutor Copy code
flow.executor = LocalDaskExecutor()d
David Yang
12/09/2021, 8:49 PMawesome. Thanks a lot. It runs in parallel.
k
Kevin Kho
12/09/2021, 8:50 PM
Of course!
5 Views