@Anna Geller @Kevin Kho (or anyone else) - Just curious, has there been anyone making a Pandas task library? I seem to be making my own as I go along, but just making sure I'm not doing it the hard way

I have a task which I am attempting to parallelize. The function I am mapping returns a small dataframe. I need to append all of these small dataframes into a final result which i will send to the next task. So far, I am attempting to append to a global DF within the flow steps (no luck):

finalDF = finalDF.append(foo.map(aList))

TypeError: cannot concatenate object of type '<class 'prefect.tasks.core.function.FunctionTask'>'; only Series and DataFrame objs are valid

Am I on the right track here, or do I need a completely new approach?

I started using prefect cloud to manage my workflows. I like the UI but I would like to be able to perform prefect cloud tasks with python code. Specifically I am attempting to do the following three things: 1. Start a flow, which I registered on prefect cloud with Python code 2. Fetch the parameters used to run a flow from prefect cloud with Python code/GraphCL 3. Change the parameters and then rerun the flow The documentation on how to achieve these tasks is a little sparse, but I think it is possible. How can I go about achieving these goals?

Sorry, wrong slack lol

Hi everyone, I have a question about how to correctly search for flows/tasks in a given state using the API. What I’d like to do is search for all flow runs that have finished. If I use

flow_run(where: {state: {_eq: "Finished"}})

that returns nothing, I think because it is using string matching. Is the set of strings that can appear in

state

listed somewhere? Thank you!

Airbyte Task Operator Naming I am attempting to set up a flow which triggers a number of Airbyte pipelines using the pre-made Prefect Airbyte task. The run is working as expected using these docs as guide(https://docs.airbyte.com/operator-guides/using-prefect-task). However, all the tasks are called AirbyteConnectionTask which makes it harder to troubleshoot errors etc. This is the code:

from prefect import Flow, task
from prefect.tasks.airbyte.airbyte import AirbyteConnectionTask
from datetime import timedelta
from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock
from prefect.executors import LocalDaskExecutor


schedule = Schedule(clocks=[CronClock("0 0 4 * * *")])


pipeline1 = AirbyteConnectionTask(
    airbyte_server_host="localhost",
    airbyte_server_port=8000,
    airbyte_api_version="v1",
    connection_id="xx-xx-xx")

pipeline2 = AirbyteConnectionTask(
    airbyte_server_host="localhost",
    airbyte_server_port=8000,
    airbyte_api_version="v1",
    connection_id="xx-xx-xx")

pipeline3 = AirbyteConnectionTask(
    airbyte_server_host="localhost",
    airbyte_server_port=8000,
    airbyte_api_version="v1",
    connection_id="xx-xx-xx")

pipeline4 = AirbyteConnectionTask(
    airbyte_server_host="localhost",
    airbyte_server_port=8000,
    airbyte_api_version="v1",
    connection_id="xx-xx-xx")



with Flow("dwh Sync", schedule, executor=LocalDaskExecutor()) as flow:
    flow.add_task(pipeline1)
    flow.add_task(pipeline2)
    flow.add_task(pipeline3)
    flow.add_task(pipeline4)

# Register the flow under the "airbyte" project
flow.register(project_name="airbyte")

I have already tried to use the @ task operator along with a function, but the task then finishes in a second without actually triggering the pipeline.

I have saved a flow to a docker registry and now would like to load and register with my prefect server. How can I load from the deployed image?

Is there any functionality for Prefect to notify me if an agent stops querying? I have notifications right now set up from tasks if a task fails, but if the agent never kicks off the flow because it stopped querying, I’ll never know.

I am having issue installing prefect. It throwing an error of “could not find a version that satisfies the requirement dask>=2021.06.0(from prefect) can anyone help me on this ?

Hi all, what's the best practice around using RDS postgres database authentication for Prefect Agents on AWS? Do folks use IAM database authentication or Postgres Native Authentication?

Hello, I am trying to run a prefect flow triggered by when there is a change in a github repository that is NOT the repo that the prefect flow lives in. TLDR: pull request on repository A triggers prefect flow living in Repository B. This is the closest I've come to figuring this out: https://docs.prefect.io/core/pins/pin-08-listener-flows.html Has anyone done this before or have instructions?

Hi, I was wondering whether Prefect Cloud allows one to export all existing data and them import to an on-premise Prefect Server. This is desirable if one starts with Prefect Cloud but later find on-premise setup is a better choice for whatever reason.

Getting a bizarre error when creating a project with the CLI. The referenced project (

veg_management

) was recently deleted with

prefect delete project veg_management

because I had some old flows in there I wanted to clean-up. But now when I attempt to recreate it I get the errors below. Note that if I change the name of the project (e.g.

veg_management_

) then it works.

Hello! We were getting an error during a github action that deploys our pipeline:

prefect.exceptions.AuthorizationError: Malformed response received from Cloud - please ensure that you are authenticated. See `prefect auth login --help`.

We changed from using an auth token to api key as a result of the error and are our flows are hanging. Are there changes that need to be made to our agents or to using service accounts? Thanks for the help.

Hello! How do I access the string value in a Parameter object? I'm passing a string Parameter to a flow throw the CLI as

prefect run --watch --id flowId --param strparam="paramValue"

. Then in the flow.py file, I am initializing outside the flow

strparam = Parameter("strparam", default "hello")

but I'm unable to access

"paramValue"

. Prefect only sees this

<Parameter: strparam>

Hello :) is there any working example which demonstrate the usecase of packaging flows in Python packages and then importing them e.g. as a subflow or just import the flow and run it from an other script? Thank you for your support!

After upgrading to v1, I am seeing an error when my ECS Agent attempts to start a new task. Any idea why?

Hi, I'm new to prefect and I am facing an issue. I have a flow A which takes a country_code as a paremeter. I would like to know how to map the flow A to a list of country_code ? Currently, I was thinking triggering the flow A by mapping the flow A to each country_code specified in a flow B (as shown in the images) Thanks by advance

Referencing Local Files in Code I am attempting to reference some key files that are stored locally on the VM where the flows were registered from, but prefect cant seem to find the files in the local file system. I have tried to change the storage to local with no luck. Sorry if I have missed something very basic here! It runs fine on my local machine with similar file references(have adapted the code to fit with the VMs file locations before registering the flow).

@task
def run():
    script....

key_location = "~/prefect/Prefect/keys/key.json"

with Flow("pipeline", executor=LocalDaskExecutor(), run_config=LocalRun(labels=["local"]), storage=Local()) as flow:
    run()

I am getting error: FileNotFoundError: [Errno 2] No such file or directory: Any ideas?

Getting Error when trying to pass data from task What is the proper way to do this ? See code in Thread

Hi, is there a way this

--param a=2

can accept a parameter name with a space in it? I only find this from the doc, where in my case the parameter is called “config list”:

$ prefect run --id "d7bfb996-b8fe-4055-8d43-2c9f82a1e3c7" --param a=2

I'm trying to test

DaskExecutor

, would

DockerRun

be a good method to see if the dockerfile works or not?

Hi - I’m having trouble implementing a “flow of flows” pattern on ECS. My flows, individually are running fine - but when using the pattern all my

wait_for_flow_run

fails. Not sure why that is - but the error is:

"An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Role is not valid"

Something extra (or missing) required for the ECS task definition/role maybe?

Any advice for handling floats values as Parameters? Will selecting integer in this menu coerce my values?

What difference, if any, is there between the labels set during

flow.register

vs. on the

run_config

object e.g.

ECSRun

? If not, is there a preferred place to set it?

hey everyone, how can I get the flow run id of the current flow that is running in flow definition?

I'm trying to use

DockerRun

with

LocalStorage

but I have a

config.py

file that is not packaged in my package, but it is beside the

flow.py

however seems that

flow.py

was not able to access this. The working directory in docker is set to the same place where both

config.py

and

flow.py

. Is there a setting I can configure?

Can i use docker-storage for flow with ubtuntu 20.04 as a base image not prefect in Dockerfile?

Hi, I have noticed, that the execution time for a flow which is run through the server is much than just running the flow locally without the monitoring from the server. I know that specific metadata will be exchanged between the server and the execution environement, but is there a way to optimize it?

Machine Learning Model Written in R? Hi Guys, I have tried to search around without any luck. Is it theoretically possible to run a machine learning model written in R through a R docker image with the docker agent? Or is a Vertex custom model using the Vertex agent the only viable option here?