Hello, what is the best way to use global variables, for example database engine objects, in different tasks?

Hi, all, can you help me, i need a sequential execution on map method (not parallel, but serial), or something similar to achieve sequential looping over a list of params. Here's my code:

from prefect import task, Flow, Parameter
from prefect.executors import LocalDaskExecutor, DaskExecutor
from prefect.tasks.prefect import StartFlowRun
import prefect

executor = LocalDaskExecutor(num_workers=1)


flow1 = StartFlowRun("flow1", project_name='test', wait=True)
flow2 = StartFlowRun("flow2", project_name='test', wait=True)

with Flow("example", executor=executor) as flow:
		
	table_dict_param_list = []
	with open('parameters.txt', 'r') as f:
		lines = f.readlines()
		for cnt, line in enumerate(lines):
				dict_param = {}
				dict_param['param1'] = cnt
				dict_param['param2'] = line
				table_dict_param_list.append(table_dict_param)
	
	flow1 = flow1()
	
	## i need this execution to be sequential like in a "for loop"	
	flow2.map(parameters=table_dict_param_list)

flow.register(project_name="test")

Design question: Why isn’t

idempotency_key=flow.serialized_hash()

the default? https://docs.prefect.io/orchestration/concepts/flows.html#core-client I can’t think of why someone would want the Flow version to change if the Flow definition hasn’t.

Hello all, I'm migrating to the newly released versioned Prefect Server helm chart and deploying it on GCP, but I seem to be running into a couple deployment issues.

prefect-server-towel

seems to have an

ErrImagePull

error, and

prefect-server-hasura

and

prefect-server-ui

can't pull their respective images from the registry. Here's my deployment command I entered:

helm repo add prefecthq <https://prefecthq.github.io/server/>
helm install ${NAME} prefecthq/prefect-server --values=values.yaml

Hi Experts, I created a flow of flows with startFlowRun. Dependencies have been set in the control flow. I have set wait=True in each of the startFlowRun(wait=True, *kwargs). Based on my understand the flow should keep going down to the dependent flows if the upstream flows has been executed successfully. I set the the test in a way that flow3 only depend on flow2 and flow2 only depend on flow1. I was think that flow2 will start execution when flow1 succeed; but it turns out that it did not get going until flow4 finish running. This is not what I was expecting. Can you see what I might have done wrong?

Hi all I'm trying to run a long running task that takes in a local DaskCluster but it seems like something on prefect arbitrarily kills off the worker... My flow uses the Dask ResourceManager looks something like

@task(checkpoint=False)
def long_running_dask_task(inputs: dict, client: Client) -> boolean:
  futures: List[Future] = []

  while(True): 
    next = get_next(inputs)
    if not next:
      break
    f = client.submit(func, next)
    futures.append(f)
    
  client.gather(futures)
  return True
  
with Flow('local-dask-flow') as flow:
  with DaskCluster(...) as client:
    long_running_dask_task(param_1)
  flow.executor = LocalExecutor()

https://docs.prefect.io/core/idioms/resource-manager.html#example-creating-a-temporary-dask-cluster

hi guys, how do i control the image name that is used for docker storage? e.g. if i enter a flow name with underscores, it converts it to dashes to get a repo name https://docs.prefect.io/orchestration/flow_config/storage.html#docker

Hello all, has anyone experienced issues with agent health? Long running agent processes have suddenly stopped polling Prefect Cloud

Anyone else had the problem of mapped tasks succeeding with every sub-task but staying in a

mapped

state?

the

RUN

tab with

Docker

doesn't seem to parse the JSON version of the Environment variables correctly seems to keep the wrapping quotes as part of the value

hi everybody, I am currently using

prefect cloud

, and my flow will invoke

aws lambda

and return me with the response. However, after registering my flow, whenever i try to run it, it throws

Unexpected error: TypeError("cannot serialize '_io.BufferedReader' object")

Is there a workaround on this?

@task(name="invoke_lambda")
def invoke_lambda(function_name, table_path, etl_target_date):
    lambda_client = boto3.client("lambda")
    response = lambda_client.invoke(
        FunctionName=function_name,
        Payload=json.dumps({"table_path": table_path, "etl_target_date": etl_target_date}),
    )
    return response


with Flow(
    "test-flow",
    executor=LocalExecutor(),
    run_config=LocalRun(),
    storage=S3(
        bucket="random-bucket",
    ),
) as flow:
    x = invoke_lambda("test", "a/b/c", "2021/04/02")


flow.register(project_name="xxx", labels=["dev"])

Hello all, I see that I can set up retry logic with

@task(max_retries=3)

but how can I set up the retry logic if I am not setting up tasks with the functional API? I do not see anything in the docs that explains. Thank you in advance!

Hi Folks, we are in the process of migrating to make use of a KubernetesRun config and a DaskExecutor, alongside our Kubernetes agent on EKS. We seem to be running into issues running our flows with a custom context from the prefect UI.

Hi all, we have a flow which is infinitely running because a mapped task is stuck on "Starting task run...". The log is showing just that, nothing more. The task itself is decorated with

@task(max_retries=6, retry_delay=timedelta(minutes=10), timeout=60)

, however it seems the task run is stuck before it can trigger the retry. May be there another reason this is happening other than the OOM Killer, as seen in a similar topic from 15.03?

Hi, we are using SQL Server Task - SqlServerExecute to execute a truncate table query (also merge query). We are encountered this error. The documentation suggests that the data field can be optional - https://docs.prefect.io/api/latest/tasks/sql_server.html#sqlserverexecute we are able to execute SqlServerExecuteMany and SqlServerFetch. Can someone help?

Unexpected error: TypeError('execute() takes no keyword arguments')
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 865, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 299, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/tasks.py", line 454, in method
    return run_method(self, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/tasks/sql_server/sql_server.py", line 90, in run
    executed = cursor.execute(query=query, vars=data)
TypeError: execute() takes no keyword arguments

Hi all. Can anyone explain how does garbage collection works? like, will the result of

produce

stay in memory until the flow finishes, or will it be removed as soon as

consume

finishes?

@task
def produce(url):
    return download_big_json(url)

@task
def consume(big_json):
    do_something(big_json)

with Flow('my_flow') as flow:
    urls = Parameter('urls')
    produced = produce.map(urls)
    consume.map(produced)

Hey Prefecters, how to I upgrade from prefect version 0.14.9 to 0.14.10?

hi everybody, just wondering if its possible to search `task_run_id, flow_id`from the prefect UI?

is there a way to initialise a custom result handler in a prefect agent run flow? I can, of course, initialise a result handler at build time but now I am all fancy with common containers for test and production I don't want the redis/azure result destination to be hard coded in the docker storage.

Hi there! Does anybody here work with Prefect together with Python notebooks and MLOps/Feature/Model stores?

curious if anyone has any thoughts on how to avoid this kind of looping behavior when the dask worker is repeatedly killed after running out of memory...is this what version locking is for? I can't remember exactly how that works and it doesn't seem to really be documented anywhere

Hey guys, can someone please point me in the direction of importing user-created functions from other python files? We have a large library of common functions that are used in almost every script, I'm developing a use-case for using Prefect over VisualCron but I'm running into some issues here

Any experience/advice creating flows dynamically in one script? I have several client configs in json files and one flow looping around those configs to execute the tasks according to each client. I was thinking of dynamically loop but have many flows created instead. any thoughts?

Hey team, is it possible to use a Deploy Key instead of a PAT for authentication when using GitHub Storage?

Hello everyone, I'm stuck in an error when running my flow with helper scripts from  

GitLab

  The UI tells me: 

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named 'mail_client'")

 . The main problem here is that the file mail_client.py is not present in the agent, and for me it is impracticable to send each auxiliary script to the agent (there is going to be a lot of flows) The structure of the repository:

gitlab-repository/
├── flow.py
└── mail_client.py

Inside my flow.py it imports the mail_client:

from mail_client import MailClient
...
...

The configuration for 

GitLab

 storage:

flow.storage = GitLab(
    repo="XXXXX",
    host="XXXXX",
    path="flow.py",
    secrets=["GITLAB_ACCESS_TOKEN"]
)

I need the agent to be able to pull the entire repository because there will be many processes being inserted in the prefect and there is no way to change the agent with each modification in a process. does anyone have any solution for this? Thanks

how are flow configurations supposed to be used? https://docs.prefect.io/orchestration/agents/docker.html#flow-configuration these labels and env variables are presumably specific to each agent that i start up, so i can control, say, staging and prod but this looks like something global that's checked in to the codebase that goes with the flow code definition

Hrm, so if I run the

DaskExecutor

in a Jupyrer notebook and I'm using threads (

cluster_kwargs={"processes": False}

) then I see the logs in the notebook. If I take that off though, then the logs disappear. How do I have the logs go to the notebook again?

Hey! I've task that is scheduled to run every 4 hours. However sometimes the associated agent isn't running for let's say a day. In that case the pending runs obviously pile up. I'd like to avoid that, especially because there is no point running that many runs the next day. I can imagine 3 ways to implement such behavour, unfortunately I didn't find anything in the docs so far: 1. set a limit of pending runs of a given fllow 2. timeout a pending run after some time 3. add a task to the flow that checks the scheduled time and skips the actual task if above some threshold I assume 1 and 2 are not available. 3 should be possible somehow. Any pointers on how to implement this?

I'm creating an integration with datadog and I need to add a custom handler to prefect's flows. I need to add a handler that gets all logs from the run and spits it out as json. I've seen a lot of examples like the one bellow, but that's not what I need. I don't want to add a handler to each logger.

logger = prefect.context.get('logger')
logger.addHandler(log_handler)

I need something like this

log_handler = logging.StreamHandler()
log_handler.setFormatter(DatadogFormatter())
get_logger().addHandler(log_handler)

I add the handler to the 'root' logger and everything is logged the way I need. I've tried the latter and it works fine with

flow.run()

, but when I register the flow I can't get it to work with k8s. More details in the thread