Dear all, while the flow was running perfectly before and still runs locally, I suddenly get an error that seems to be related to pickle when running it on cloud:

TypeError: can't pickle _thread.lock objects

See full error message below. Any ideas, what could be the reason?

Is there a way to use a simple function in a flow, for example when I am going to call the function many times. I can easily make it work by making my function a 'task', but I can't see why it should have all the scheduling overhead:

@task
def get_host_from_kv(opts):
    hid = Schema({
        'host': str,
        Optional('identity', default=0): int
    }).validate(opts.kvopts)
    return HostParam(identity=hid['identity'], host=hid['host'])


with Flow("Query a single host") as flow:
    opts = Parameter('opts')

    result = winrm_powershell(host=get_host_from_kv(opts),
                              command=load_script(opts, 'win_ps_version.ps1'),
                              opts=opts)

Hi Prefect Community, I have prefect server running with GraphQL. I also have an agent registered. How do I deploy my script as a flow to this server? I do have an

Agent ID

from the server. A little confused around this. TIA!

Hi 😛refect: team, a question about current best practices running Dask on Fargate…

I’m not getting a response on GH, so I bring it here again to attention: https://github.com/PrefectHQ/prefect/issues/3603 TLDR: parsing secrets with newlines in value of json (as env var), requires escaping. Is this the intended behavior or can I adjust it such it works as expected.

👋 I'm running into issues setting up Fargate tasks. The agent successfully launches the ECS task, but the task fails at the validation stage after it loads the flow from S3. It thinks the task and container definitions are different, but neither has changed since I registered the flow and the agent scheduled it. I wonder if it's because the flow is using S3 storage? I didn't use Docker because I couldn't figure out how to use it with ECR and provide credentials. Thoughts?

2020-11-06T07:56:19.166-05:00

Copy
	containerDefinition.flow-container.command -> Given: ['/bin/sh', '-c', "python -c 'import prefect; prefect.environments.execution.load_and_run_flow()'"], Expected: ['/bin/sh', '-c', 'prefect execute flow-run']
containerDefinition.flow-container.command -> Given: ['/bin/sh', '-c', "python -c 'import prefect; prefect.environments.execution.load_and_run_flow()'"], Expected: ['/bin/sh', '-c', 'prefect execute flow-run']

Hi, I'm experimenting with "Non-Docker Storage for Containerized Environments" using AWS S3, but I haven't found a way to provide a list of external packages on which the flow is dependent (an equivalent for python_dependencies option available in Docker storage). Is there a way to achieve this?

Dear prefect people, we get plenty of

No heartbeat detected from the remote task; marking the run as failed.

errors. 💔 (e.g. https://cloud.prefect.io/accure/task-run/8e44bd11-e7c3-4982-bcf2-7711bc1ca4a9?logId=83475348-2531-4efb-9663-33748515908e) This is particularly unexpected, as most of those tasks seem to have already almost finished... 🤔 How do you approach these errors and debug them? 🙂

Hi everyone, I need to build a service which runs a couple of the same workflows for about 10 000 b2b clients on a daily basis. It will take and process data from their APIs on schedule according to the clients' off-hours. So I want to write that workflows with a CLIENT_ID parameter and get the same control and functionality of them (e.g. UI, parallelism, logging) as if there were personal workflows for every client. It seems there is no such possibility in Airflow. Is there one in Prefect?

Hi everyone! I’m exploring ideas around dynamically configuring a Fargate instance/cluster based on a flow parameter / input. In short, I have an hourly job that migrates data for a group of clients based on their timezone. For some hours of the day, there will be more clients and thus more work. Ideally, I want my flow to scale resources based on the number of clients queried for a timezone. Does anyone have ideas for a way to do this? Since flow environments are set at registration time, I’m struggling to find a feasible approach. I’ve also started to explore the new

ECR Task

and

run_config

recently added, which sound like a potential solution. Thanks for the input!

Hello! I am running into an interesting problem. I am using a parameter called "lifecycle" in my flow (Dev and Prod) which determines which secret the flow should be using. The secrets are set using the following code:

lifecycle = Parameter("lifecycle")
    true_branch = PrefectSecret("SCM_AUTOMATION_PROD")
    false_branch = PrefectSecret("SCM_AUTOMATION_DEV")
    ifelse(tasks.get_lifecycle(lifecycle), true_branch, false_branch)

    secrets = merge(true_branch, false_branch)

When I run this flow locally, setting the lifecycle parameter set to DEV it works as expected. However, on the cloud, it is only using the PROD secret. Any guidance on why this might be happening? I am positive the cloud secrets are all configured correctly and the issue seems to be with the IFELSE, MERGE Thanks in advance!

Is there some sort of standard practice with event driven prefect flows? Say I have some sort of data-cleaning-flow in prefect that takes in a data file, cleans it, and stores the new file in cloud storage. I have two other flows, A and B, that I want to have triggered by the completion of the data-cleaning-flow. I want them to more-or-less subscribe to the data-cleaning-flow. Is there a prefect-esque way of doing this, or is my only option to create some external machinery that detects the successful completion of the data-cleaning-flow and keeps track of all the flows that I want to have triggerd by that flow.

I could combine the data-cleaning-flow logic into flow A and flow B, and not have this problem, but then I would be repeating work

Another way of phrasing it: Suppose I have an E-T-L flow. And one day I decide I want to have another L be triggered from the data produced by the T. Maybe one day I have five Ls that need to be triggered from the data produced by the T. I don't want to start adding more *L*s to the same flow because that would get messy after awhile. Also, maybe I want a flow that just does the T-L and not the E, because I already have the data output from E from a previous flow and E takes awhile and I don't want to do it again.

Have a simialar question to the above ^^ whats the easiest way to trigger downstream flows from a completed flow? Use the prefet API to directly trigger?

Also just some questions on “all_finished” behaviour, if a -> b -> c -> d and d is triggered on all “all_finished” if a fails and so b and c are skipped, will d still run (looking for a finally block).

Is there a way to quickly start running an existing flow starting with a task in the middle of the dependency graph? (treating like all its dependencies have already been met)

Suppose I have a class called

ReadTable

which extends

Task.

This class just SELECTs data from a big SQL table. I want to perform this SELECT in parallel with say n sub-Tasks. So the

ReadTable(n)

object should spawn n sub-Tasks when it runs. How would this work within Prefect's OOP paradigm, where

ReadTable

inherits

Task

?

Hi! Quick question, we had flow concurrency for a specific label set to 5, we scheduled 20 flows and the first 5 ran for 12 hours (and are still running). We updated flow concurrency for that label to 20 but the remaining 15 flows have remained in the scheduled state. We've confirmed that we can run more than 5 flows concurrently by triggering a brand new flow which gets submitted and run immediately, but we can get those original 15 to submit. Any suggestions?

Hello Prefect Community, for a reason I don't understand my file can't recognize the prefect '_from_ prefect import task, Flow'. although it works, how can I fix that?

Does all tables in DB created by prefect have proper FK relations so if i remove let say a flow via graphql, all connected flow runs and logs will be removed too?

Hi, some of my tasks import modules from a distant path using

sys.path.append()

. This seems to work fine when I run the flow locally with flow.run(), but not when running it from the Prefect Cloud/Server. (The error says the module can’t be found) Does it sound right? If so, why would it be the case?

Hi Everyone, I'm using the imperative API, and was wondering if there was any tutorials or samples for the use of Parameters with imperative API

Hello! Would it be possible to add to the Prefect CLI and/or the Core client an option to create projects only if they don't already exist? My use case is that I intend to create projects and register flows when an ECS task starts up, and then the ECS task will go on to serve as a Prefect agent. The new

idempotency_key

option and

serialized_hash()

method for flow registration makes this really nice, but I'll still get an error for the project creation when the ECS task restarts. I could ignore this error, but then that could be problem if somehow the error is not due to the project already existing, but a genuine failure to create a new project.

Hello guys, I am seeing the following error when trying to build a flow in an azure devops pipeline:

WARNING: Retrying (Retry(total=0, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NewConnectionError('<pip._vendor.urllib3.connection.HTTPSConnection object at 0x7f74c98d0710>: Failed to establish a new connection: [Errno -3] Temporary failure in name resolution',)': /simple/cryptography/
ERROR: Could not find a version that satisfies the requirement cryptography (from versions: none)
ERROR: No matching distribution found for cryptography

Removing intermediate container f73d8367b237
The command '/bin/sh -c pip install requests cryptography snowflake-connector-python wheel' returned a non-zero code: 1
Traceback (most recent call last):
  File "./flows/frieze-viewing-room-flow/flow.py", line 113, in <module>
    flow.storage.build()
  File "/usr/local/lib/python3.6/dist-packages/prefect/environments/storage/docker.py", line 351, in build
    self._build_image(push=push)
  File "/usr/local/lib/python3.6/dist-packages/prefect/environments/storage/docker.py", line 418, in _build_image
    "Your docker image failed to build!  Your flow might have "
ValueError: Your docker image failed to build!  Your flow might have failed one of its deployment health checks - please ensure that all necessary files and dependencies have been included.
##[error]Bash exited with code '1'.

any ideas on why this might be happening? it looks like it cannot connect to pypi for some odd reason. ideas/thoughts welcome. thanks!

Hi! I'm running prefect server and my flows keep getting stuck on

submitted

. I originally started with a kubernetes agent and thought it was a configuration issue between prefect and kubernetes, but I get the same behavior with a docker agent. The agents pull the correct docker image and submit it, but the flows don't enter a running state. I'd be grateful for any ideas or pointers! Cheers

Hey all. I'm trying to deploy all our flows automatically from Google Cloud Build or GitHub actions in our Prod environment. We use Docker storage for our flows. Has anyone had any luck in doing this? I can foresee a few issues: 1. We will need to run Docker within the Cloud Build container as the

flow.register()

method uses Docker to build the image. I'm not sure if this is possible. 2. We will need to create firewall rules that allow Google Cloud Build to communicate with Prefect server running as a Compute VM. I'd be interested how people are automatically deploying flows from any CI/CD platform.

Trying to wrap my head around the imperative api, I feel like I want a little more fine grained control on the graph than what the functional API provides, but its really hard for me to understand how i can pass data between tasks. Ex.

Task A: 
  run(self, x, y):

Task B: 
  run(self, a_result, x, y):


with Flow() as flow: 
    x_parameter = Parameter("x", required=True)
    y_parameter = Parameter("y",  required=True)
    # question is how do setup the parameter pasing from teh parameters -> A -> B
    # my attempt but still dont know how to keyword_tasks in this scenario to pass along results from A
    flow.add_task(x_parameter) 
    flow.add_task(y_parameter)
    a = A() 
    flow.set_dependencies(a , upstream_tasks=[x_parameter, y_parameter], keyword_tasks=??? )
    flow.set_dependencies(B(), upstream_tasks=[a, x_parameter, y_parameter], keyword_tasks=????)

also is there a common pattern for chaining tasks post mapping? I.e. if i map a list of ints to a task that multiples them by 2, and then want each of those to pass their value down to a parametrized StartFlowRun task (baiscally a task that will do some comptutation over a map of values, and then trigger n downstream flows for each value