Channels
announcements
ask-marvin
best-practices-coordination-plane
data-ecosystem
data-tricks-and-tips
events
find-a-prefect-job
geo-bay-area
geo-berlin
geo-boston
geo-chicago
geo-colorado
geo-dc
geo-israel
geo-japan
geo-london
geo-nyc
geo-seattle
geo-texas
gratitude
introductions
marvin-in-the-wild
pacc-clearcover-june-12-2023
pacc-may-31-2023
ppcc-may-16-2023
prefect-ai
prefect-aws
prefect-azure
prefect-cloud
prefect-community
prefect-contributors
prefect-dbt
prefect-docker
prefect-gcp
prefect-getting-started
prefect-integrations
prefect-kubernetes
prefect-recipes
prefect-server
prefect-ui
random
show-us-what-you-got
Title
w
Wei Mei
04/01/2022, 3:30 PMHi! In my flow I call my tasks like this. I am experiencing a problem where the snowflake_load is running before the upload_to_s3 task.
today, dir = get_data(endpoints=api_endpoints)
upload_to_s3(today, dir, endpoints=api_endpoints)
snowflake_load(today, schema="statistics", endpoints=api_endpoints)k
Kevin Kho
04/01/2022, 3:32 PMHey @Wei Mei, you can find the syntax to explicitly define upstream tasks here
:gratitude-thank-you: 1
w
Wei Mei
04/01/2022, 3:36 PMthis looks to have solved it!
today, dir = get_data(endpoints=api_endpoints)
upload = upload_to_s3(today, dir, endpoints=api_endpoints)
snowflake_load(today, schema="statistics", endpoints=api_endpoints).set_upstream(upload)👍 1
Hi kevin, I found the schematic page and it looks cool; it shows that my flow is definitely not what I thought it would be.
the endpoints=api_endpoints is a list of three strings that I use in loops. I think I should be using the map feature but I still don’t fully understand it yet.
In the tasks I see three tasks Get Data (FunctionTask), Get Data [0] Getitem, and Get Data [1] Getitem.
in the logs it looks the Getitems didnt do anything, the FunctionTask printed out the 3 lines of log for the 3 items in my list.
k
Kevin Kho
04/02/2022, 7:19 PMCan I see the code?
w
Wei Mei
04/02/2022, 7:21 PMsure!
# returned as pandas df's
for endpoint in endpoints:
file = f"{endpoint}-{today}.csv"
df = get_viewer(seg, endpoint)
if not df.empty:
<http://logger.info|logger.info>(f"{endpoint} has data")
df.to_csv(f"{dir}/{file}", index=False)
<http://logger.info|logger.info>(f"{file} created.")
else:
<http://logger.info|logger.info>(f"{endpoint} is empty: {today}")k
Kevin Kho
04/02/2022, 7:25 PMis this inside the
Floww
Wei Mei
04/02/2022, 7:27 PMthis is a task get_data()
this is whats in my flow:
api_endpoints = [ "site1", "site2", "site3" ]
today, dir = get_data(endpoints=api_endpoints)
upload = upload_to_s3(today, dir, endpoints=api_endpoints)
snowflake_load(today, schema="statistics", endpoints=api_endpoints).set_upstream(upload)k
Kevin Kho
04/02/2022, 7:31 PMI think this works though right? Or is there something wrong? I think the GetItem is a task Prefect is adding right?
w
Wei Mei
04/02/2022, 7:33 PMyes, it working so far.
was curious what the two Getitem tasks were.
Thanks for your time Kevin. My POC is running now.
thank you and everyone else for the hand holding.
k
Kevin Kho
04/02/2022, 7:46 PMPrefect adds intermediate tasks to put stuff in list or unpack them sometimes. I don’t know exactly where in this case but I think it’s related to the api-endpoints being a list
w
Wei Mei
04/02/2022, 7:50 PMgotcha. I was worried based on the schematic that the snowflake_load task might run before the upload_to_s3 because of the way it is drawn, but the .set_upstream() should take care of that so I wont ask about the magic behind this one 😄.
k
Kevin Kho
04/02/2022, 7:52 PMMapping perfectly fits your use case though so you don’t need to loop inside the task btw, you should try it out
w
Wei Mei
04/02/2022, 7:54 PMi definitely will!
Have a great weekend!