prefect-community #prefect-community

Hello all, thanks for building such an amazing product! I do have a situation that i could use some help with: i have a task that does something very small; is there way to define a FunctionTask on the fly and add it to the flow without having to build an actual function with the

@task

decorator? Here is an extremely simplified version of what i'm doing (See thread). Am I able to define the run_load_proc function as just a FUnctionTask within the flow without an explicit function? Also, not sure if defining the global SqlServerExecute is a total anti-pattern, let me know. Thanks for all of your help.

Haseeb Ahmad

03/07/2022, 2:35 PM

Hey guys I have a prefect job that is running on a schedule. What I want to basically achieve is that the new flow does not get run until the old one is finished. For instance consider a flow is scheduled every hour and if it takes more than an hour for it to finish then the new scheduled one should not start until the old one has finished running. I was looking at

Inspecting flow runs

from prefect.backend import FlowRunView

flow_run = FlowRunView.from_flow_run_id("4c0101af-c6bb-4b96-8661-63a5bbfb5596")

Can you please guide me on the best approach to achieve the above use case. Really appreciate your time and help.

Sarah Floris

03/07/2022, 2:42 PM

okay so I am trying to translate a prefect core flow to prefect cloud flow. I want to be able to make changes locally and have it update my cloud flow. Is this possible?

Kevin Kho

03/07/2022, 3:36 PM

Join our events channel for Prefect-related events!

Daniel Nilsen

03/07/2022, 4:26 PM

I am trying to register a flow to my helm deployed server with

flow.register(«myProject»)

. To connect to the server I change the

server.endpoint

in config.toml to the correct url. This works. But I don’t want to change the config manualy like this. Is there a way to temporarily change it when registering? I have tried

set_temporary_config({"server.endpoint": "123"})

but it does not seem to work 🤔

Zhibin Dai

03/07/2022, 4:34 PM

when using conditional logic in a flow, is it necessary to have two

with case

statements, or can i use just one that does something if a condition is true?

Jason Motley

03/07/2022, 4:59 PM

What tends to cause this error for a task in the middle of a flow, when upstream tasks have run fine?

: No heartbeat detected from the remote task;

Paul Butler

03/07/2022, 5:01 PM

Hi Prefect community!

👋 2

Paul Butler

03/07/2022, 5:33 PM

Copy code

Question for the Community - I am trying to use Prefect to schedule and monitor a Dbt project/pipeline.   I follow example and pass in dbt_kwargs for connecting to snowflake.  My dbt project runs fine with dbt run or dbt compile command, but the DbtShellTask fails - yet does not provide any reason for error.  Anywhere more detailed logging is recorded??  

Sample tasks like the hello one included here, run OK.   I'm using Studio Code to debug this code, but also try to run it in Python IDLE and get same error output     I'm running this:

from prefect import task, Flow, Parameter
from prefect.tasks.shell import ShellTask
from prefect.tasks.dbt import DbtShellTask

@task(log_stdout=True)
def say_hi(name):
    print("hello {}!".format(name))

with Flow(name="dbt_flow") as f:
    name = Parameter('name')
    say_hi(name)

    task = DbtShellTask(
        profile_name='default',
        environment='dev',
        dbt_kwargs={
            'type': 'snowflake',
            'threads': 4,
            'account': 'mysnowflake.account',
            'user': '<mailto:myemal@myco.com|myemal@myco.com>',
            'authenticator': 'externalbrowser',
            'role': 'ROLENAME',
            'database': 'DBNAME',
            'warehouse': 'ENGINEERING_XS',
            'schema': 'DV_PROTO'
        },
        overwrite_profiles=False,
        profiles_dir='C:\\Users\myDBTuser\.dbt'
    )(command='dbt compile')

out = f.run(name='Paul') 

[2022-03-07 17:18:07+0000] INFO - prefect.TaskRunner | Task 'DbtShellTask': Starting task run...
[2022-03-07 17:18:07+0000] ERROR - prefect.DbtShellTask | Command failed with exit code 1
[2022-03-07 17:18:07+0000] INFO - prefect.TaskRunner | FAIL signal raised: FAIL('Command failed with exit code 1')
[2022-03-07 17:18:07+0000] INFO - prefect.TaskRunner | Task 'DbtShellTask': Finished task run for task with final state: 'Failed'
[2022-03-07 17:18:07+0000] INFO - prefect.TaskRunner | Task 'say_hi': Starting task run...
[2022-03-07 17:18:07+0000] INFO - prefect.TaskRunner | hello Paul!
[2022-03-07 17:18:07+0000] INFO - prefect.TaskRunner | Task 'say_hi': Finished task run for task with final state: 'Success'

Adam Roderick

03/07/2022, 6:06 PM

Hi, after upgrading to prefect 1.0 we are seeing this warning during our deployments

...venv/lib/python3.8/site-packages/prefect/client/client.py:848: UserWarning: A flow with the same name is already contained in storage; if you changed your Flow since the last build, you might experience unexpected issues and should re-create your storage object.

I don't understand why we see that, because, we create a new Docker storage object on every deployment

Copy code

flow.storage = Docker(....)

Can anyone explain what this error means, or how to address it?

kevin

03/07/2022, 7:40 PM

Quick question about the documentation: https://docs.prefect.io/orchestration/ui/interactive-api.html On this page there’s an image that shows that you can query by

flow_run.duration

but I can’t seem to find that queryable value in the

flow_run

fields on the interactive API in Prefect Cloud. Could someone help me resolve this differential?

Pedro Machado

03/07/2022, 8:51 PM

Hi there. We are running 0.15.13 and would like to upgrade to 1.0. We are using Kubernetes with an in-cluster agent. Will the 0.15.13 agent be able to run 1.0 flows? Alternatively, if we upgrade the agent first to 1.0, can it run 0.15.13 flows?

Jean-Michel Provencher

03/07/2022, 9:20 PM

Hi, I’m using the docker image

prefecthq/prefect:1.0.0

and when scanning the docker image with Snyk I’m getting up to 114 security issues, as well as 8 critical vulnerabilities. Are you guys planning on fixing them in the base docker image?

Dylan

03/07/2022, 9:40 PM

I haven’t done much digging on the topic, but I’m sure I’m not the first to ask or think about a data transform registry. Basically there’d be some k:p pairs used to assemble a tree of business stakeholders and the upstream / downstream dependencies based on what’s in a properties file, or can be generated from the code similar to how reflection works. Any thoughts on this? Would be cool to see a table or visualization of it. Maybe some Prefect workflow tags can be used for this too?

Dylan

03/07/2022, 9:41 PM

That way when you wanna look up data pipelines or transforms by their business function, connector source, or their criticality, you could filter them etc.

Kelly Huang

03/07/2022, 11:27 PM

Hi, I'm trying to run flows on the cloud using GitHub storage and a local agent. But I'm getting this error:

Copy code

in get_flow
  from github import UnknownObjectException
ModuleNotFoundError: No module named 'github'

I'm reading differing things about whether or not github storage can access package dependencies? Correct me if I'm wrong, but shouldn't it definitely be able to access my Pipfile and therefore dependencies? Otherwise what would the point of github storage be?

discourse 1

Andrew Moist

03/08/2022, 9:37 AM

Hi everyone. Our company is looking to get some help from a data engineering consultancy in Europe, ideally with some experience with Prefect/DBT/Redshift. Any recommendations would be appreciated. We’re mainly based in the UK. Thanks.

👀 1

Tomer Cagan

03/08/2022, 9:38 AM

Is is possible to use introspection in the API? In hasura documentation (and SO) I see it should possible but for whatever introspection query I try to use, I am getting an error (details inside)

discourse 1

Vadym Dytyniak

03/08/2022, 9:51 AM

Hi. How the default

checkpointing

works in prefect 1.0.0? In documentation I see that it is enabled by default, but what is the default

Result

Bruno Murino

03/08/2022, 12:48 PM

Hi everyone — I’m trying to start a new flow run by using the StartFlowRun prefect task, like in the screen grab, but I’m getting an error:

TypeError: StartFlowRun.run() missing 1 required positional argument: 'self'

. I don’t know if I’m doing something wrong, but it’s a bit unclear what to do

Tom

03/08/2022, 1:06 PM

Hi all, we are evaluating Prefect and were wondering if it is a good idea to initiate flows within flows

Tom

03/08/2022, 1:57 PM

I have another question regarding checkpoints, is it correct that results are not persisted when you run flows through

flow.run()

? (https://stackoverflow.com/questions/66660927/prefect-workflow-how-to-persist-data-of-previous-every-schedule-run)

discourse 1

Thomas Opsomer

03/08/2022, 2:55 PM

Hello 🙂 We have issues with tasks that require a manual approval (with

manual_only

trigger), after being approved using the UI on prefect cloud, it takes a very long time before the flow is resumed (like several hours :/). It's not the first time it happens, but currently it's happening on every flows. Anyone having the same issue ? Is it an issue on prefect backend side or should we do something ?

Marius Haberstock

03/08/2022, 3:18 PM

Hi, I have a question about

apply_map

. Details following in the thread 🙂

discourse 1

Max Kolasinski

03/08/2022, 3:49 PM

I’m trying to write a flow that creates a number of sub-flows and running into some issues getting Prefect to register it- would appreciate some guidance (example in thread)

discourse 1

Roger Webb

03/08/2022, 3:52 PM

Hey All, Im wondering the best way to make a flow force a failure. For instance, if you have a task that fails, and a task that runs a trigger for "any failed" toward the end of your flow... the flow succeeds if the "onfailure" task succeeds. However I want the OnFailure task to do what it needs to .. then make the entire flow fail.

discourse 1

Adam Roderick

03/08/2022, 3:59 PM

Have you seen this issue when launching jobs from the ECS Agent? I'm troubleshooting and not sure where to start with it. botocore doesn't like something about the parameters. Prefect version is 1.0

Bruno Nunes

03/08/2022, 4:21 PM

Hello, Do you know when is the prefect 2.0 expected to be released for production use?

ben

03/08/2022, 5:14 PM

Does there happen to be any kind of white paper, or similar, available related to Prefect Cloud Secrets management? (I know that it uses Vault on the back-end, but our security folks tend to like more detail than that). Thanks!

Martha Edwards

03/08/2022, 6:01 PM

Hi! Is it possible in Prefect to create a sub-flow that works as follows: • task A runs and may fail. Retry A until success. • task B runs and may fail (or the task itself succeeds, but it returns a result indicating failure of the sub-flow). In that case, loop back to retry A. This seems like it could be possible to do using looping on a nested flow as described in this stackoverflow post.

discourse 1