If tasks are mapped in the complex way like https://docs.prefect.io/core/concepts/mapping.html#complex-mapped-pipelines, does something else need to be done to get them to cache?

Hi there. I have a module that users

loguru

https://github.com/Delgan/loguru It's used inside of a task. When I run the flow locally, I see the loguru output but when it runs remotely (ECS), I don't see the logs in Prefect Cloud. How could I get it to write to the Prefect logs? Also, how should I set up my ECS flow to get it to write the container logs to cloudwatch? Thanks!

Hi, I’m new to Prefect. I’m trying to get a pandas dataframe out of a mapped task. Here

query_list

is a

list of str

, each of which is a valid query. Let’s say there’s 3 queries and each query gives me back 10 rows. This task works fine.

data = execute_test_query.map(query=query_list)

Now I want to transform a concatenated dataframe in its entirety.

@task
def n_rows(df):
    rows = df.shape[0]
    print(f"There are {rows} rows!")
    return rows

I was expecting

data_2 = n_rows.map(flatten(data))

to give

"There are 30 rows!"

, but I get key errors. Any idea what I need to do to flatten

data

?

Does anyone have a dummy task available that would check whether a task above has run?

Does Prefect support Vyper?

Can I expose flows through API using prefect cloud? A use case is that some ETLs are running in SSIS sever which is windows environment, and some programs running in other dedicated applications servers. I'm imaging that I need to install prefect agent on each server, and a dedicated flow to run programs installed on the server and then use one flow to orchestrate them together. each dedicated flow works as a task in this "parent" flow. and in each task, I could call API to trigger these flows in various servers. Does that make sense? Any better solutions? I'm newbie here.

@Chris White have you seen this "fact check" of your "Why Not Aurflow?" article? https://www.linkedin.com/pulse/airflow-prefect-can-we-fact-check-prefects-claims-muthuvel Would you happen to have another article that addresses the objections listed? Thanks!

Hi. I have a flow that does the following: • query snowflake for a list of URLs (there may up to 100k URLs) • create a list of lists (create chunks of 120 URLs in each sublist) • a mapped task processes each chuck/batch of URLs and calls an API for each URL. An

S3Result

with a

target

is configured for this task so the output of the API lands in S3 in json format. • A pipe in Snowflake will detect and load these files My question: is relying on the Prefect results an acceptable way to save data to S3 or should I have an explicit task that will handle writing to S3? In this approach (S3Results), will the data stay in memory until the end of the process or will the memory be released when the data is written to S3?

Hi All, I have what I hope is a quick question. We are running prefect server and seem to have an issue with the Zombie Killer for a long running process. I want to set the heartbeat_mode to "thread". Looking here: https://docs.prefect.io/orchestration/concepts/services.html#heartbeat-configuration I see the following environment variable. Is this appropriate when running prefect server? Or just for Cloud?

PREFECT__CLOUD__HEARTBEAT_MODE

Or is there an equivalent env for server, like this:

PREFECT__SERVER__HEARTBEAT_MODE

Thanks, Matt

Just throwing this out there, if any artist on here wants to design and sell stickers based on Flow task run names, I’d buy a bunch off you.

I would like to ask if anyone been thru this?

https://prefect-community.slack.com/archives/CL09KU1K7/p1638014722400800 any ideas?

Hi

I am wondering if someone can help me. I was interested in running some distributed model training on kubernetes using Horovod. I was wondering how I would go about incorporating that into a prefect flow. I am assuming I would use something like RunNamespacedJob (i.e., https://docs.prefect.io/api/latest/tasks/kubernetes.html#runnamespacedjob) with the kubernetes manifest for an

MPIJob

(from the https://github.com/kubeflow/mpi-operator library) given in the body of the task. Is that roughly correct?

Would be interested to know if anyone has tried something like this before?

Hi all, I might be missing it but, is there a way to add overwrite=True to the Prefect Azure BlobStorageUpload method?

Is there a way to get the logs to show the line in my code that caused an issue? In several cases I get Prefect-specific errors and the stack trace is through the Prefect file chain, but doesn't indicate which line in my code I could tweak/remove to avoid an issue. There a way to do this? Also, the logs to the screen seem to exclude more complex outputs at times, presumably due to formatting, where's the default save location of them for review after the fact without these constraints?

Does prefect default to a certain time zone? Local flow runs show my local time but will that hold in production? This would be for datetime functions

I am not able to access my prefect cloud (was working fine just 3 hours ago)

Hi, If I kill a agent in docker container, is there a way to enable graceful shutdown? I mean if I have some long lasting job running, I’d like to give it a timeout window instead of killing it immediately.

Hi, does anyone have recommendations or best practices that they follow around dependencies? My specific example has to do with extraction/load vs transformation steps. It is convenient to put it all in one Flow so that the transformations start after the extraction/loads are complete. But our Flows have become quite monolithic, and we want to be able to test various sections (i.e. extract/load) separately from other steps (i.e. transformations). Any recommendations here?

Hi, I am trying to execute flow run though graphql query from google cloud functions, how can I set up client so that the cloud functions know what is my client. How can I reach my prefect cloud externally from anywhere?

from prefect.client.client import Client
client = Client()
client.graphql(query)

Hi everyone, as part of our partnership program, we are exploring what official integrations to add to our task library in 2022. Feel free to chime in what partnerships/integrations you want to see in this thread.

Hi! is it possible to pass source of flow that stores in local or docker storage on one machine to agent on the other machine?

Hello! I am currently using an ECS service to run a prefect agent. When defining the service, I set a security group shared with an RDS instance. However, my flows launched from that agent are unable to connect to the RDS. Is there a way to define a security group for the task that the agent executes?

I’m looks at the Prefect Cloud plans page and I’m a little confused by the page honestly. 1. The front

Standard

card shows the feature `Role-based permissioning`; however, below in features when

Standard

is selected it has

RBAC

and

Custom Permissions

greyed out which would make me think it doesn’t have

role-based permissioning

. a. Can I get clarification on what

Role-based permissioning

actually is comparative to the features list and which plan would be needed for them? 2. From what I see, the features have either a checkmark, an open circle, a dotted circle, a greyed out filled circle, and a black full circle without an explanation of the differences, this makes it hard to understand some the differences without hovering over every. a. This is just feedback from a user perspective that it might be hard to quickly figure out the features. Thanks!

I am working on a POC to use Prefect to help orchestrate an ELT process. I was able to get a flow and tasks to run locally and show up in the local server ui. Then i setup a cloud prefect account because i wanted to test task concurrency and got it to kick of my flow from my local machine. My next step is to see what it takes to productionize the flow in Azure, so i have a few questions. 1. I know there is a azure marketplace item to Create Prefect Agent. Is that the best way to get that part setup? Is Kubernetes the way to go over just a VM with Docker? I tried getting the kubernetes version to install but it failed on the last step with "The received access token is not valid: at least one of the claims 'puid' or 'altsecid' or 'oid' should be present. If you are accessing as application please make sure service principal is properly created in the tenant.". no idea what that means and when i tried to run the install again, it gave me "A custom role with the same name already exists in this directory. Use a different name" 2. Does anyone know of a complete walkthough of setting up an agent in Azure (VM with docker or kubernetes). i did find something, but they dont show anything on setting up the kubernetes service etc. https://infinitelambda.com/post/prefect-workflow-automation-azure-devops-aks/

Assuming i’m running prefect agent on EC2 (for our PoC), is there any clear advantage to running a docker agent vs. a local agent? also, is it possible to use

rootless mode

for the docker (to increase security), or will it hinder Prefect in anyway? https://docs.docker.com/engine/security/rootless/

Hi, a basic question – what’s the recommended way to loop parameters if .map() doesn’t work with function of multiple outputs?  I have a pipeline like this

def pipe(dt):
    #get intermediary output
	output1, output2, output3 = step1(dt)
    #write to database
	step2(output1, output2, output3)

is it like this or is there a better way?

with Flow('multi-output func') as flow:
    dt = Parameter("urls", default=['dt1', 'dt2', 'dt3'])
    for dt in dt_range:
        pipe(dt)

thank you