Hey everyone - @Laura Lorenz (she/her) is about to spearhead an integration with Great Expectations and we’d value any input you care to share. Please forgive the cross-post, but we’ve had so many people join the main channel in the last few weeks I wanted to make sure everyone knew what was happening in #prefect-contributors! https://prefect-community.slack.com/archives/C0106HZ1CMS/p1588271140092800

Morning all! I’m new to Prefect, looking to deploy the server, or whole stack, in a container onto FarGate - has anyone tried this? Can’t find much on it..

Hi everyone! I have just created my first Prefect flow and deployed it to my manually set up Dask cluster via the Prefect Cloud. Everything went smoothly and I have to say this by far the best data pipeline tool I have used. However, I noticed that when I run the Flow from the cloud instead of just a script (

flow.run

) the memory consumption of my Dask Workers gradually grows until they run out of memory. When I run the same flow from a script everything finishes correctly and there are no memory issues what so ever. I am using the latest

prefect

and

dask

packages. Has anyone experienced something similar? Is there anything that could be done about it? Thanks a lot.

Hi all, we are waiting with much anticipation for https://github.com/PrefectHQ/prefect/issues/2041 ! It will improve performance for us a lot. Having said that I'm wondering if there is a pattern / feature that will come along with this to allow yielding results to the DFE? We have a task that is recursive in nature (walking a tree) that we implement using the LOOP pattern - wonder if there is something coming that will allow one loop to spawn multiple loops concurrently (I guess this is a separate issue), and then start processing the result of each loop in DFE way?

Hello everyone 👋, I’m having a conceptual issue I’m hoping to get some clarity on. Say for example, you have a number of tasks you want to execute in Redshift. I’m using psycopg2 to establish a connection and create a cursor. You cannot pass the cursor object from task to task because it is not serializable. How do you execute multiple tasks within a transaction block if you need a new connection per task? Am I thinking about this wrong? Is psycopg not the recommended connection method? Thanks!

hey everyone - is it possible to run a workflow in a docker image outside of the cloud

hi! I'm not actually a prefect user, i'm just a person willing to help and contribute in open source project. I'm not very experienced as developer, but I'm quick-learner and have enough spare time during self-isolation to grow my skills by ton of practice. I'm from Kyiv, Ukraine, would love to connect. I've already made my first PR and going to make more 🐛

hey everyone - separate question: I have a workflow and in a very rare instance need to run 3000 runs in parallel. Is there an easy way to point the worflows at a different env at at the time of execution

Coming from Airflow, I've become reliant on the detailed statistics and analytics for my jobs. Is this on the roadmap for Prefect?

how does everyone test the results of a task?

with Flow("test_target_transformer") as f:
        te = fit_target_transformer(problem, target, train_data)
        df = target_encoder_transform(te, imputed_categorical_df)
 assert df.isna().sum() == 0

I'm trying to test a @task that returns a dataframe, but assert statement errors on

AttributeError: 'FunctionTask' object has no attribute 'isna'

. How do I return this as a DataFrame and not a task?

Hi! First time trying this out. I got the server running locally. Can someone point me to instructions on how to connect to github? I would like to set it up so the first step is the pull the most recent code and executes that. I'm coming from Airflow so might be missing something so any help is appreciated!

Can anyone tell me what is going wrong on my local?

Just a quick one to see if

ifelse

has been integrated within the imperative api yet?

Hello everyone, thanks for this amazing product, I am just getting started with it. I was wondering the following though. I want to execute different flows using a local dask cluster. But as the documentation aptly describes, running

flow.run()

leads to a blocking process until the flow has finished. Is there a way of submitting jobs to the dask cluster in an almost fire and forget scenario? Especially since my outputs are written to file or DB I really don’t need the return/result object. I basically want to dynamically create workflows submit them to the dask cluster and then have them finish based on available resources. Could be that I am overlooking something super simple or using prefect for something it is not build for 🤷‍♂️ but would be nice to know 🙂

Hello there 👋, I ain't able to see logs of my tasks on Prefect Cloud for some days now. Does something has changed? The only logs I'm able to see are those from the Agent or the CloudFlowRunner (see attached image n°1) One strange thing is when a task failed, I can see the

Last State Message

in the UI but not in the logs tab (see attached image n°2 and n°3) For info, I'm using RemoteEnvironment with LocalDaskExecutor on Prefect 0.10.5

Hopefully someone can give me a light thump on the head to show me the errors of my thinking or teach me the magic Prefect incantation to simplify how I'm trying to solve the data science workflow. I have a pipeline consisting of Query, Prepare, Train, and Evaluate tasks. Each of them is time consuming and depends on the output from the previous task. I can turn on Checkpointing, Caching with a LocalResultHandler. The developer process we used previously kept a dependency tree off what python file and source file(s) were used to generate the next file. If you deleted a data file or updated a source file (think Task) and ran the pipeline it would figure out what task needed to be run. Let say you updated the code in the Model Task then the cached output of Prepare would be used with the new model task code to generate a model that was evaluated by the Evaluate task. I think of it is as doing a reverse look up on the DAG to find the earliest upstream task that needs to be run. All downstream of any task that is run are also run and their cached results aren't used. As I'm currently using Prefect if the Model Result is missing the Model Task will be rerun, but the Evaluate Task will find existing cached inputs and blindly use them. 1. Is there some slick way to handle this pattern in Prefect that I just haven't found yet? 2. Currently I have a CacheValidatorFactory that builds the cache function with a list of source files and destination files. The validator checks to see if any source files are newer than any destination files or if any destination files are missing to determine if the cache is valid or not. If a source file is newer or a destination file is missing then the cache is invalid and the task runs. This may be related to https://github.com/PrefectHQ/prefect/issues/2104 I've also noticed that if the output of a task is cached and used that it still reads all the cached inputs. 1. Is there a way to have a different cache_validator for inputs and results? 2. If the results cache will be used can we avoid loading the cached inputs? 3. What happens if the Prepare runs, Model doesn't and Evaluate should? Not sure my example makes total sense, but how do we trace back through the DAG to see which upstream results have changed?

Hi Prefect Team, I would love your thoughts on if this is the Prefect way on doing this. I have a scheduled flow where we need a

run_date

that is yesterday UTC time. I would love to use the prefect context to do this and this worked for me, but is there a better way?

Has anyone gotten flows executing in a Dask Gateway remote environment? based on the docs it looks like i'll be writing my own, but wanted to double check there isn't a way to do it currently without writing a new environment

I'm getting a request error when I try to

prefect auth login -t my_generated_token

requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <http://localhost:4200/graphql/alpha>

has anyone else ran into this? The server start up brought up all services just fine.

Hey! We want to use Prefect with k8s in a way that would automatically connect to k8s API and will create Pods according to our real-time requirements. can this be done with Prefect?

New to prefect : Do we need to install/setup anything locally if we use prefect cloud ?

Hi everyone. I am evaluating Prefect and would like to convert a workflow that is currently running in Airflow. This workflow calls a reporting API that returns a list of URLs pointing to data extracts (multiple compressed csv files for each report). The input parameters of the API include the start and end dates, a "report id" that tells the API the set of data we want, and a parameter that controls the interval of the report (daily, weekly, or monthly). The next step is to download each of those URLs and store them in s3 using a predefined prefix naming convention. Each report can have different a frequency and data readiness delay. For example, some weekly reports are ready on Wednesdays others are ready on Saturdays. Some daily reports are ready the next day, others the following Saturday. In principle, there is a single flow that 1) waits for a number of days or hours 2) queries the API to see if a given report is ready and 3) downloads multiple files to s3. In Airflow, I have multiple DAGs (one for each report/frequency combination). Each DAG pulls a report based on the inputs and

execution_date

. I am trying to figure out how I could implement this in Prefect. It seems to me that I could have a single parametrized flow but I am not sure how the parameters would be provided and the flow scheduled. Thanks!

I've just converted an event-based ETL job to Prefect. ATM it just kicks off the job and exits immediately. I'd love to see PIN 14 implemented so Prefect can observe the events and mark the job done when the appropriate event is received.

Hey team - is there a way to find out via the graphql api about which upstream task is linked to a downstream task run ? Specifically I’m interested in the input parameters to an upstream task

Hi. What's the recommended approach to implement something like an Airflow sensor? For example, one that will wait until a certain time or one that will poll an API or run a database query until it returns a certain value? Would a simple long running task with a loop and a delay work?

Hi! I am new both to Airflow and Prefect I have a scenario that when a file comes into a certain folder on a machine, I will have to run a python program and parse and send the contents of that file to a DB I was able to do it with using directory watchers in python (by using inotify or watchdog), I might have close to 1000 files coming to that directory anytime, so my file watchers would be always running But since discovering about the prefect project, I would want to incorporate prefect in the whole picture I was wondering if there is something of a directory watch trigger (which would trigger a data flow as soon as a file comes into that watch directory) available in prefect? similar to PIN 8

When i set up my flow with storage and an environment and then call

flow.run()

locally, it automatically defaults to the local executor instead of using the environment that i set for it, but when i register it and a kube runner picks it up it runs it with the correct environment. does anyone know why?

Hey community! Is there a way to run a flow inside a different flow? I mean suppose I have an existing flow and want to trigger it as part of a different flow, without specifying the flow again. Is there a way to do that? e.g. a Task that wraps an existing flow? Preferably a flow that is registered to the orchestration server?

Hi - I am encountering a botocore

NoCredentialsError('Unable to locate credentials')

error using Docker Storage (local image), prefect cloud and the

s3ResultHandler

When the flow is run locally. There are no credential issues initializing the handler

s3_handler = S3ResultHandler(bucket='some-bucket')

To resolve the issue I created a Prefect Secret called

"AWS_CREDENTIALS"

and tried

s = Secret("AWS_CREDENTIALS")
aws_scrts = s.get()
s3_handler = S3ResultHandler(bucket='some-bucket', aws_credentials_secret=aws_scrts)

What is the best way to resolve aws credential error? Thanks

Hi, I am refactoring my flow parameters and I encountered a slight problem when I compose two flows with a parameter with the same name:

ValueError: A task with the slug "limit" already exists in this flow.

I could come up with a plumbing hack like this:

flow = Flow(name)
        local_flow = build_local_flow()     # a function that returns a flow
        quetzal_flow = build_quetzal_flow() # idem

        # plumbing: both local and quetzal flows have a limit parameter
        limit_parameter = local_flow.get_tasks(name='limit')[0]
        other_parameter = quetzal_flow.get_tasks(name='limit')[0]
        for edge in quetzal_flow.edges:
            if edge.upstream_task == other_parameter:
                edge.upstream_task = limit_parameter
        quetzal_flow.tasks = set(t for t in quetzal_flow.tasks if t != other_parameter)
        flow.update(local_flow)
        flow.update(quetzal_flow)
        ...

which works, but it seems very hackish and far from elegant. Is there a cleaner alternative to this? (other than renaming the parameter, of course)