I used task looping in a really long flow that failed 6 hours in due to an obscure error. I was disappointed to see that when I restarted it from the Prefect Cloud UI, it restarted from the beginning rather than from the latest state. Is there a way to preserve "task_loop_result" context and use it in the task when manually restarting a failed flow?

Hello all. I tried using

create_flow_run.map()

with

wait_for_flow_run()

and

get_task_run_result()

but I'm getting some errors. I'm not sure if I'm doing this correctly

We are running into an issue with flow runs hanging after a task fails -- a task directly downstream from the failure ends up stuck in a Pending state even though all its upstream tasks are complete. We're using an

ECSRun

with a

LocalDaskExecutor

and I'm unable to find any way to debug or diagnose why the task isn't ever executed.

Continuing to be confused by parameters here…is there any easy way to dynamically use parameters as strings for various task inputs. IE I have a

param=Parameter("myvalue", default="myvalue")

. I want to us it in a variety of tasks say…

execute = execute_file(f"sql/{param}.sql")
upload = upload_file(f"{param}.csv")

Is there a way to do this without constructing a task for each input?

Passing Pandas DF to task if i have

task_1

return a df and I pass the df as an input to

task_2

, does the df physically go to the Prefect server ? the issue I have is that it looks like it is getting stuck when trying to return the df and pass it to the next task. is there a better way to pass df from one task to the next ?

@task()
def task_1():
    df = ....
    return df

@task()
def task_2(df):
    df2 = df ....
    return df2

def main():
    with Flow("voc_sentiment") as flow:
        df = task_1()
        df2 = task_2(df)

according to the docs it's possible to define individual environment variables at runtime for a given agent by using the

--env

flag, but is there also something like an

--env-file

flag if you have several environment variables defined in a file (kind of like you have with docker-compose)? https://docs.prefect.io/orchestration/getting-started/flow-configs.html#configure-environment-variables

Heya! Trying to make a flow that does some GraphQL stuff - specifically, turning off schedules (via

set_schedule_inactive

). It lets me read (I grab all the flows from a given project), but the

mutate

gives me this error (the query works when I go to the Interactive tab on the GUI btw)

AuthorizationError([{'path': ['flow'], 'message': 'AuthenticationError: Forbidden', 'extensions': {'code': 'UNAUTHENTICATED'}}])

Hey Good Morning everyone, Have anyone tried to use an NVIDIA-DOCKER as a prefect agent to run some flows. I have got the setup ready and running, but when the task which needs to use the GPU's start executing, I see it doesn't use the GPU, but instead it runs on CPU. I can see the below logs

Hello community 🙂 We experienced a strange behaviour of the "restart" button on prefect cloud: When going to a task run and restarting the task, it correctly reschedules the task, but it doesn't reschedule the downstream tasks (in the UI they stay green instead of light blue)

hey guys would anyone have a simple example of using a GQL query to trigger a flow run I can look at?

Hello ! We have some memory intensive flows. These flows sometimes are triggered at the same time and executed through a LocalExecutor. The LocalExecutor will immediately launch all flow runs and cause memory errors. I wonder if there is a way to make LocalExecutor or any executor to execute one flow run at a time (sequentially) rather than launch all of flow runs at the same time?

I’ve built a flow that continues to be stuck in the ‘Submitted’ state. Following this documentation, I was able to narrow down that the cause is with my docker image. I have multiple docker images in my repo that works fine, but using this specific docker image in my ECSRun causes my flow to be stuck in a submitted state. Could anyone let me know what might be wrong with this docker image?

FROM ubuntu:18.04
COPY requirements.txt /requirements.txt
COPY google_secret.json $HOME/.config/gspread_pandas/google_secret.json
ENV PATH="/root/miniconda3/bin:$PATH"
ARG PATH="/root/miniconda3/bin:$PATH"
RUN apt-get update

RUN apt-get install -y wget && rm -rf /var/lib/apt/lists/*

RUN wget \
    <https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh> \
    && mkdir /root/.conda \
    && bash Miniconda3-latest-Linux-x86_64.sh -b \
    && rm -f Miniconda3-latest-Linux-x86_64.sh 
RUN conda install pip
RUN pip install -r requirements.txt
RUN conda install -c conda-forge theano-pymc -y

I have a noob question, are you able to add a

@task

decorator on a function imported from a library (as opposed to one I defined)?

Im trying to return three pandas dataframes in a task (aka function) and it gives me this

TypeError: Task is not iterable. If your task returns multiple results, pass `nout` to the task decorator/constructor, or provide a `Tuple` return-type annotation to your task.

Can anyone point me to an example where I can make it iterable?

My flow failed and when I checked the error is the exception I inserted to check for completeness of data being extracted. I Actually did it because in the past without this check incomplete data got extracted. Could it be the way I raised the ValueError exception? Is there's a way in prefect to do exception that would suit my case? and secondly is there a way I can link this exception to a retry mechanism? That is if this occurred the flow could be seen as failed and therefore retry

small nitpick here, but when logging into Prefect Cloud the suspicious login alerts show up as coming from Okta instead of Prefect, does it make more sense to brand these as Prefect emails just to minimize confusion (and not reveal what Prefect is using on the backend)?

Can I have a complex schedule that runs the same flow at the same time with different parameters?

is it bad practice to define a parameter in a module scope rather than in the flow context?

input_a = Parameter('a')

@task 
def abc(b, c):
  return a + b + c

with Flow("abc_flow") as flow:
  # in context
  result = abc()

Hi Team - Trying to use a secret I created in the Cloud UI with Flow(name='flow_example') as flow: my_secret = PrefectSecret("my_password") data = needs_credentials(snow_pw=my_secret) Error that no local variable found. Need to set to reference cloud API

Hello Prefect community! I'm new here, exploring Prefect's features for the first time. I have a question. Can Prefect tasks be long-running listener tasks, sort of like Sensors in Airflow? Any recommendations or best practices to accomplish this?

Hi everyone, join the #events channel to keep up with Prefect events (or show/advertise your own Prefect-related event)

Kicking off the #events channel with my latest Meetup presentation!

Hi folks -- I'm exploring Prefect Orion as an option to replace Airflow. (I understand it's an evolving technical preview; this is for a longer-term R&D project.) First off, let me complement the team on the documentation. While maybe everything isn't documented yet, the documentation is excellently written. One question I had that was not immediately clear from the docs (and I've tried some searching, but haven't turned up an answer) is whether Deployments can be invoked with arbitrary parameters? It looked from the docs like I would need to create a different Deployment for each parameter? Maybe stepping back a bit, what I want to do is use an API to trigger "on-demand" pipeline work for a given input (e.g. building map tiles for a geo region). If Deployments aren't the right way to think about this problem, would it be better to just instantiate a Flow with a parameter [via API]? (I understand this to be possible based on some docs I was reading last night.) Ultimately, we would want to kick off these ETL or longer-running jobs and be able to get information about where in the system (which task in a flow) a specific job is. It feels like Prefect is a good fit for this workflow. Thanks in advance!

Hello Prefect community, I upgraded from 0.14.22 to 0.15.13 and containers are no longer launching. I'm deployed on ECS and I can see a ConnectTimeOutError that is preventing the tasks from coming up. This is the error in question:

requests.exceptions.ConnectionError: HTTPConnectionPool(host='127.0.0.1', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f1911c36d90>: Failed to establish a new connection: [Errno 111] Connection refused'))

If I look at the task itself, I can see that the environment variable for PREFECT__CLOUD__API is set to http://127.0.0.1:4200. So this seems like the problem. The host it's trying to connect to is clearly wrong (since the server itself is running on an EC2 instance). So I've adjusted my ~/.prefect/config.toml to look like this:

host_ip = "my.ip.goes.here"
host_port = "4200"
host = "http://${server.host_ip}"
port = "4200"
endpoint = "${server.host}:${server.port}"
  [server.ui]
  apollo_url = "<http://my.ip.goes.here:4200/graphql>"
[cloud]
api = "${${backend}.endpoint}"
endpoint = "<https://api.prefect.io>"
graphql = "${cloud.api}/graphql"

No luck. So I added the PREFECT__CLOUD__API definition to my environment variables in the container definition. Still no luck. However, when I look at the task definition, I can see the correct (or at least intended) PREFECT__CLOUD__API environment variable there. But the variable in the task is still set to http://127.0.0.1:4200, and the problem persists! I am pretty stuck on this, and hoping that someone here has a line of sight to the solution. (This all worked without much configuration previously ... which now seems weird.) All help appreciated!

I am testing Prefect deployed into a kind kubernetes cluster using the helm chart. I'd like to mount a folder on my local filesystem to a running job. I've configured this in the job template using:

apiVersion: batch/v1
kind: Job
metadata:
  name: prefect-job
  labels: {}
spec:
  template:
    metadata:
      labels: {}
    spec:
      containers:
        - name: flow
          image: prefecthq/prefect:latest
          imagePullPolicy: IfNotPresent
          ...
          volumeMounts:
            - name: filesystem-dir
              mountPath: /job-files
      volumes:
        - name: filesystem-dir
          hostPath:
            # directory location on host
            path: /path/to/my/dir
            # this field is optional
            type: Directory

I'm unable to mount the path which I believe is a function of the volume not being available on the agent. Is there a straightforward way for me to mount a volume to the agent in the helm chart deployment?

Hello Prefect Community! A question about mapped tasks and concurrency - we are not sure what is supposed to happen.

Do mapped tasks run concurrently? Or does concurrency only happen when a dask executor is used?

Thanks!

Hey everyone, I have a pretty complex flow with multiple tasks that reads data from multiple sources, then extract the relevant parts from it and saves it to google storage. I have 2 questions: 1. This flow was originally written outside of Prefect and used pool.map in order to extract data from the objects I have (html pages) - is there any knows issue regards using pool.map inside a Prefect task context? 2. In another task that isn’t using pool.map Im running this (valid) pandas row -

comparable_items_df['tag_rank'] = comparable_items_df.groupby(['id', 'tag_name']).cumcount()

And I get the following error (when running outside of a task context it works)

TypeError: unhashable type: 'ResultSet'

For now I’m running this locally on my mac but it will run eventually using Vertex AI. I’ll appreciate any help since I’m currently stuck. Thanks.

Hi Guys - I have a few questions on prefect cloud - the questions are after viewing the architecture picture on this https://docs.prefect.io/orchestration/ under section architecture overview authentication question - can prefect use auth0 for authentication of users ( real people not service accounts/users ) - any reference diagrams will be helpful network and traffic filtering related questions on prefect cloud hosted on GCP - 1) how does prefect cloud connect to gcp, looking at the picture on https://docs.prefect.io/orchestration/ under architecture section the agent 1,2 on gcp and connection to cloud api - is it over internet? any references on prefect cloud documentation stating this 2) on the same connection question, is traffic filtering possible using GCPs private access point option https://cloud.google.com/vpc/docs/private-service-connect, if yes, please any references of this prefect cloud documentation? This is just for my knowledge, and this is more of a documentation question than actual architecture involved here :-) as similar hosted solutions like snowflake and elasticsearch, these two hosted solutions do have all the documentation of above 🙂 thank you in advance.

Hey everyone, I have got the docker agent working with GPU based on the instructions from @Kevin Kho. But how do I run a flow locally in the agent and not by spinning up a new docker agent? The problem with this is that the new docker agent being created by prefect when triggered by the flow is actually not having runtime as Nvidia and doesn't see the Nvidia drivers and devices which makes it run in CPU. So I would like to know if it is possible to run the flow in the docker agent itself with starting a new Docker agent for the Docker run? Thanks, Sen

Hi! I don't know if its a known issue but I experienced some strange behavior with the date selector when using the DateTimeParameter, selecting any date between april and october froze the UI for one minute then it selected some later date (25 october for 2020, 31 october for 2021 for example, for the ones I tested) Seems like a timezone conflict, Prefect Cloud detected my time zone as Paris, I changed to local it's working for now 🙂