Hi 🙂 I have an issue to use _*GCSResult, t*_he results are not stored (but there isn't any error raised). First, I tried to use environment variable :

from dotenv import load_dotenv
load_dotenv()

@task()
def add(x, y=1):
    return x + y

with Flow("Test GCSResult", result=GCSResult(bucket='thread-prefect-flows')) as GCSResult_flow:
    first_result = add(1, y=2)
    second_result = add(x=first_result, y=100)

Without success, however :

from dotenv import load_dotenv
load_dotenv()

def implicit():
    from google.cloud import storage

    # If you don't specify credentials when constructing the client, the
    # client library will look for credentials in the environment.
    storage_client = storage.Client()

    # Make an authenticated API request
    buckets = list(storage_client.list_buckets())
    print(buckets)

implicit()

works well, then I'm not sure to understand : "_Make sure your Prefect installation can authenticate to Google's Cloud API_" (https://docs.prefect.io/core/advanced_tutorials/using-results.html#example-running-a-flow-with-gcsresult) Then, I tried to use Local Secret (https://docs.prefect.io/core/advanced_tutorials/using-results.html#example-running-a-flow-with-gcsresult)

[context.secrets]
GOOGLE_APPLICATION_CREDENTIALS="auth/token.json"
GCP_CREDENTIALS="auth/token.json"

(don't know what to use between GCP_CREDENTIALS (prefect.utiliities.gcp) or GOOGLE_APPLICATION_CREDENTIALS (in the documentation of GCP), anyway I tried both and it didn't work neither. What did I do wrong? Please help 🙏

Hi, I am getting the following error on prefect cloud, No heartbeat detected from the remote task; marking the run as failed. I tried adding [cloud] heartbeat_mode = "thread" in my backend.toml file and this line, flow.run_config = UniversalRun(env={"PREFECT__CLOUD__HEARTBEAT_MODE": "thread"}) in my code but I'm still getting the same error. Any help to what I am missing please?

Hello - has anyone successfully been able to pass parameter names into a flow name? I am trying to schedule a flow of flows by doing the following within a flow, but it always creates a flow titled

My Flow - today - today

rather than creating the date range

start_date = Parameter("start_date", default="today")
end_date = Parameter("end_date", default="today")

# this function handles parameter typing and converts today to YYYY-MM-DD
start_date_str = get_date_param_as_string.run(start_date)
end_date_str = get_date_param_as_string.run(end_date)

create_flow_run(
        run_name=f"My Flow - {start_date_str.run()}-{end_date_str.run()}",
        flow_name="My Flow Name",
        project_name=prefect_project,
        parameters={"start_date": start_date, "end_date": end_date, "file_type": "file_type"},
    )

What's the best way to incrementally load tables (e.x. a loop) but then also ensure that if a connection is lost, the task retry will re-establish the connection AND not create duplicates on previously loaded rows?

Hi,How can we skip some tasks in a flow

Im using helm chart in k8s for running prefect server. In it's value.yaml, I've passed env (which gets populated from secrets). Problem is, context secrets gets populated only if I manually edit the agent deployment and set the --env argument in "agent start" command. just setting the env doesnt populate the context.secret in tasks. Any help?

Hi everyone! Is there a way to set a time limit for a flow run or a specific task? We’ve updated our python dependencies and some packages don’t seem to play nicely with each other anymore, one of the external apis we use somehow got stuck in an infinite loop, which made one of our flows run for 16hrs+ and probably longer if I hadn’t noticed and manually canceled it. 😓

Hey, i’m wondering for

prefect agent kubernetes install

what are the default mem and cpu limits? I have my tasks running on kubernetes spun up on EKS, and setup the agent using that command. However, the task is a download FTP file and it just stalls. No progress for hours. I’m wondering if this could be a possible mem and cpu limit, and if I should basically run the install command again - and will that overwrite my old manifest?

Hi, quick question about this workflow, from the diagram it appears flow B and C are running 'in-parallel' but when I run this flow B and C are actually one after other with C running after they both complete. The flow of flows works great but does not appear to run the flow in true parallel... https://docs.prefect.io/core/idioms/flow-to-flow.html#scheduling-a-flow-of-flows

Is it expected that the logs contain

Starting task run...

for tasks that are skipped because an upstream task fails? (Example code in thread.)

When a parameter is changed thru the UI under settings for a flow, does this automatically propagate to future runs? I usually don’t see the change reflected in a pending flow run unless I flick the schedule off and on - I only do that to be pedantic but was wondering if it is neccesary

hey so im working with some legacy code that has working directory requirements (ie needs next to or one level up from a number of dependent files). When i run a flow locally in a script using flow.run() is there a way to have it run from a specific directory irrespective of where the python session is initialized from?

Hello, Is it possible to define dependencies between a flow scheduled to run on an agent on premise and an azure ML pipeline scheduled on the azureML that runs on a cluster managed by azure? I have some flows that perform some ETL that should be done before my AzureML pipeline runs. Sometimes the ETLs runs longer than it should and since the data is refreshed, my AzureML pipeline runs on stale data. Appreciate any help

Hey all 🙂 I’m setting up Prefect with GE and can’t find the right way to implement the logic I have in mind. If the GE Validation fails (some data is wrong) I would like to be notified via Slack (Prefect Automation) as sorts of a warning, but the flow should carry on. What would be the correct way to implement that? 🧐

Hi, I’m using prefect cloud and an AKS (Azure Kubernetes) cluster running a vanilla k8s agent (from the prefect agent kubernetes install command). The agent connects to prefect Cloud, but when I run a flow I get an error saying

'Connection aborted.', OSError(0, 'Error')

. I have added the RBAC stuff to the deployment, and opened outbound traffic to api.prefect.io on port 443, but I have not been able to get anything to run on the AKS agent. Any ideas?

Hello, We are facing an annoying issue while using prefect with prefect server. Here is the precise stack: • prefect 0.15.12 (but issue was also often observed with previous version) • prefect server 2022.01.12 • executor DaskExecutor with cluster_class being KubeCluster • using kubernetes runner • and azure as storage What we observe is that sometime our temporary dask cluster which is started by the prefect run is not tear down at the end of the run. It's kinda annoying because if we don't pay attention it means we keep using the VM resources for nothing. Has anyone faced a similar issue?

Hi - looking to call a flow within another flow and return the results of the child flow to the parent flow. This works well using the Local agent and Local storage, but returns the following error when called with the Kubernetes agent and GitHub storage:

ValueError: The task result has no `location` so the result cannot be loaded. This often means that your task result has not been configured or has been configured incorrectly.

I’m not explicitly providing where or how results of the child flow should be stored, and I believe that is the root cause of the error thrown. Is that correct? If so, what is the preferred method for storing results? e.g. is PrefectResult preferred over S3Result or is it a matter of opinion?

Hi all ! New to Prefect here 🤟 I'm working on a flow that would send data retrieved using MySQLFetch() to an S3 bucket. What would be the best way to transform the tuple that returned from the MySQLFetch() so that it can be written into an S3 in the appropriate format ? Currently I'm getting this error if I'm sending the results directly to the S3Upload() function :

stream = io.BytesIO(data.encode())
AttributeError: 'tuple' object has no attribute 'encode'

Hi all, How to use secret defined in prefect cloud in Azure storage? I have added "azureflowsecret" as secret in prefect cloud. This is my code: Storage=Azure("edwflows",connection_string_secret="azureflowsecret") but when I tried to register it to prefect cloud, it failed with the error: ValueError: Local Secret "azureflowsecret" was not found.

Join us for the Prefect livestream at 3 pm today! Looking forward to seeing you guys. We will parallelize a machine learning model grid search over Dask with Prefect.

Say I’ve got two or more lists I’d like to map a task over. If I provide two mappable arguments to

Task.map()

it iterates through both of them together like

zip()

. Is there a way to loop over every combination instead, like

itertools.product()

?

Is there a prefect recommended way of checking length of tuples received from database fetches (PostgresFetch, SqlServerFetch)? I don't think you can do get length on a task. This is just to set a case statement depending on length of tuple.

What means are available for tracking task runs across different versions of a flow? If I have a flow with one task, and I reregister the flow, the task’s ID will change.

Hi everyone, I’m running into this error:

Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'cfmeta\'",)\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n  - python: (flow built with \'3.6.15\', currently running with \'3.6.8\')\nThis also may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.',)

I see that it is common and I read this thread. However, our flow needs to use specifically 3.6.15 because some of our dependencies require it and I do not see an agent that matches this version. I saw that script based storage was suggested, but we generate our flows dynamically so that is not an option. Any suggestions on how to resolve this? Thanks in advance!

Hi all, In our prefect deployment, we see task logs like.. "Task finished with final status Pending". Does Lazarus process handle this? I have seen this stuck for hours.

Hi everyone. I have a process that submits multiple jobs to an external service, and I have to wait for all of them to be finished before proceeding. I've found how to wait in a task for some external process https://github.com/PrefectHQ/prefect/discussions/3944 to be finished but no idea how to wait for multiple ones to be finished. Thanks

Hello everybody, I want to this https://github.com/PrefectHQ/prefect/issues/5297. But i didn't see any docs about it . Can i do that? I want to create dynamic workflow with input.

Hi everyone! I've setup a prefect server on a Google compute instance, but I still have problems grasping some concepts: I have the server and local agent running on the GCE instance and the UI is working. I can register and run a flow that resides on the GCE machine without errors. Now I want to setup GitLab as code repository. My initial intuition was: I can probably setup a connection to my Gitlab repo in a config file and then all flows will be synced from there. When I commit something, it will be updated in prefect as well. This is not the case as it seems. I'm trying to wrap my head around how to use the Storage options, but I really don't understand it. On my server, do I have to create python files for each of my flows that reside in Gitlab? What sense would the repository make then? Thanks for your help! I'm really lost here...

Hi I am facing an issue I have created an extraction flow and run it locally , its working fine, this program contains one task and inside this task its calling another function from different python module for ex import abc @task def fun() abc.fun1() assume like above , its working in local but when i register the flow in cloud its failed saying abc not found how can we resolve this should I write all the logic inside task function itself ??

Hi! In Prefect cloud flow name should be unique across the team or projects?