Channels
pacc-may-31-2023
prefect-ai
pacc-clearcover-june-12-2023
marvin-in-the-wild
data-ecosystem
geo-israel
pacc-june-14-2023
geo-japan
prefect-cloud
ppcc-may-16-2023
prefect-azure
prefect-docker
prefect-recipes
gratitude
geo-nyc
geo-bay-area
geo-boston
geo-london
geo-dc
geo-chicago
geo-berlin
geo-texas
geo-seattle
geo-colorado
prefect-community
data-tricks-and-tips
prefect-aws
prefect-gcp
introductions
find-a-prefect-job
prefect-dbt
random
events
ask-marvin
show-us-what-you-got
prefect-getting-started
prefect-integrations
prefect-contributors
best-practices-coordination-plane
announcements
prefect-server
prefect-ui
prefect-kubernetes
Title
j
John Ramirez
03/11/2020, 7:40 PMHey - does anyone have a code sample for Control Flow task such as
switchifelsez
Zachary Hughes
03/11/2020, 7:50 PMHi John, is something like this helpful?
https://docs.prefect.io/api/latest/tasks/control_flow.html#functions
j
John Ramirez
03/11/2020, 7:51 PMNo I was having this conversation in a different thread
Laura Lorenz (she/her)
2 days ago
Is it possible to implement your situation such that the dataframe is returned from the upstream task already split up into two, like {'a': df1, 'b': df2}, and then the downstream tasks index that result like task1(result['a'] and task2(result['b'] ? Basically doing the splitting itself in the upstream task
Laura Lorenz (she/her)
2 days ago
the switch task operates on a dictionary, but also it is for only doing one task tree of a set of mutually exclusive task trees, but it sounds like you want to run the task tree for group A AND group B, not just one or the otherI understand what Laura is saying but not familiar enough with prefect to implement it
I have split results into two separate pandas data frames and want to send the results to different tasks
daily_pull = df[df['adjustment_factor'] == 1.0]
full_refresh = df[df['adjustment_factor'] == 2.0]
return {
'daily': daily_pull[COLUMN_MAPPING['quote_cols']],
'full_refresh': full_refresh[COLUMN_MAPPING['quote_cols']]
}z
Zachary Hughes
03/11/2020, 8:00 PMOkay, gotcha. What I think she's proposing is something along the lines of:
@task
def generate_dataframe():
return {"portion_a": 123, "portion_b": 456}
@task
def process_dataframe_portion_a():
# do something
return ...
@task
def process_dataframe_portion_b():
# do something
return ...
with Flow("switch-flow") as flow:
dataframe = generate_dataframe()
portion_a_results = process_dataframe_portion_a(dataframe["portion_a"])
portion_b_results = process_dataframe_portion_b(dataframe["portion_b"])j
John Ramirez
03/11/2020, 8:07 PMok that is what i though but I was getting errors cause I was passing a mapped object
thank you for your help
👍 1
z
Zachary Hughes
03/11/2020, 8:38 PMSure thing!