prefect-community #prefect-community

Hi there community - we’ve got a flow run which failed due to some sort of connection time out. I’d like to restart it at the task it was at when it failed. What’s the best way of doing that? It had been retrying a particular task.

Jake

05/18/2022, 2:37 PM

Hello! We have a flow that is highly resource intensive. We already specify custom resource requests but that alone is not enough. When the agent schedules it; we would like it to make use of a priority class to ensure that it has access to all the resources it needs. How can this be done? I see in the docs that there is a way to provide a custom job template:

Copy code

flow.run_config = KubernetesRun(job_template_path="<s3://bucket/path/to/spec.yaml>")

But I’m not sure how to write that template / how this template is used (are there any docs on this?). Thanks!

iñigo

05/18/2022, 2:42 PM

Hi, Is there an easy way to move from a local agent with python scripts to an docker agent?

Jessica Smith

05/18/2022, 2:42 PM

Is there any difference between using LocalDask vs Dask without providing a cluster? Do they do the same thing under the hood?

Tim Enders

05/18/2022, 2:53 PM

How do I read the output file of storing my Flow Results in GCS with

GSCResult

Alexandru Anghel

05/18/2022, 4:34 PM

Hello, I am new to Prefect and i've started evaluating the Orion version for including it into a project i'm working on. What i am trying to do is create a basic Bigquery table and insert some dummy values in it (the tutorial from here: https://prefecthq.github.io/prefect-gcp/bigquery/ I can create the deployment but when trying to run it i get this error:

Copy code

RuntimeError: Tasks cannot be called from within tasks. Did you mean to call this task in a flow?

I'm not sure what i'm missing here as i run the tasks inside the flow. Could you please assist me? Thanks! Awesome project, by the way!

Xavier Witdouck

05/18/2022, 4:35 PM

Hi all, I’v created a couple of secrets on prefect cloud and I am trying to test a flow locally that leverages these secrets, but when I run I get the follows error

Xavier Witdouck

05/18/2022, 4:35 PM

Copy code

raceback (most recent call last):
  File "/Users/witdxav/opt/anaconda3/envs/v0.9.216/lib/python3.10/site-packages/prefect/engine/task_runner.py", line 880, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/Users/witdxav/opt/anaconda3/envs/v0.9.216/lib/python3.10/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/Users/witdxav/d3x/projects/d3x-quanthub-sdk/src/d3x/quanthub/prefect/prefect_tasks.py", line 9, in docker_run
    barra_license = Secret("BARRA_LICENSE").get()
  File "/Users/witdxav/opt/anaconda3/envs/v0.9.216/lib/python3.10/site-packages/prefect/client/secrets.py", line 170, in get
    raise ValueError(
ValueError: Local Secret "BARRA_LICENSE" was not found.

Xavier Witdouck

05/18/2022, 4:35 PM

It seems that it is looking for a local secret

Xavier Witdouck

05/18/2022, 4:36 PM

I am logged into the cloud backend

Xavier Witdouck

05/18/2022, 4:36 PM

How do I convince my local flow to access secrets from the cloud?

Jacqueline Riley Garrahan

05/18/2022, 5:51 PM

Hi, I'm wondering if someone can point me to documentation on how I can compose a flow-of-flows and map upstream flow results to parameters in downstream flows.

Mike Vanbuskirk

05/18/2022, 5:55 PM

is anyone successfully getting ECS-based agent flows working?

Mike Vanbuskirk

05/18/2022, 5:55 PM

I’m banging my head against a wall

John O'Farrell

05/18/2022, 6:14 PM

Hello, I was trying to test getting a local flow up and running. Was able to successfully register and run it using the prefect cloud ui, but I tried switching to using the

prefect.Client

api to run the task and it threw an error

Failed to load and execute flow run: TypeError("'ABCMeta' object is not subscriptable")

. Now any subsequent attempt to run the flow using the prefect ui throws the same error, even though the code hasn't changed.

Jessica Smith

05/18/2022, 6:25 PM

I'm getting a weird error when trying to run a flow using the DaskExecutor:

AttributeError: 'Edge' object has no attribute 'upstream_task'

I'm getting this when dask tries to use pickle to load the first of my mapped tasks. The first two tasks of the flow are not mapped, and they don't have issues. Any thoughts?

Darin Douglass

05/18/2022, 6:56 PM

is there a way to have

create_flow_run

not base the new flow’s name off of the current flow without passing

run_name

Jessica Smith

05/18/2022, 7:30 PM

It seems that the UI will not let you do a quick run on a Flow that has a Parameter that is required but has a default value of None. Is this expected behavior?

William Jamir

05/18/2022, 9:04 PM

I have a file that defines the flow instance as a global variable, which is later used for registering the flow. Something like this:

Copy code

with Flow(...) as flow:
   x = Parameter(...)
   ...

Which is registered with:

Copy code

prefect register -p flows/my_file.py --project "MyProject"

But now I dont want to have this flow as a global variable, because my tests need to do some settings before and I dont want to trigger any of the code within this context manager. I can easily move this to a function, and make it return the flow instance, but I dont know how to deal with the registration by command line. I mean, its possible to still use the register by command line, or do I need to make it now programmatically? Basically, I’m looking for a solution like this:

Copy code

def main():
   with Flow(...) as flow:
      x = Parameter(...)
      ...
   return flow

Registering

Copy code

prefect register -p flows/my_file.py:main --project "MyProject"

Danny Vilela

05/18/2022, 9:30 PM

Is it possible to set a task to only write – not read – a result on Prefect Core? I was thinking of sub-classing a custom result class I wrote with flags

enable_read

and

enable_write

and just check those within the

read

and

write

methods, but maybe there’s something easier?

Shaoyi Zhang

05/18/2022, 10:02 PM

Hi! When using Prefect Cloud, where is the log of flow runs stored? What about the parameters in past flow runs?

Ken Nguyen

05/18/2022, 10:31 PM

I have a task that I think returns only 1 result, but I keep getting the below error. Is there a reason why this task thinks it’s returning multiple results? (code in thread)

Copy code

TypeError: Task is not iterable. If your task returns multiple results, pass `nout` to the task decorator/constructor, or provide a `Tuple` return-type annotation to your task.

Pasha Kravtsov

05/18/2022, 10:43 PM

I have a Prefect flow that fails when it is unpickled, the error message is related to me decrypting variables in pydantic's Settings().. is there any good way to move forward and debug pickling issues like this?

paimoe

05/19/2022, 1:24 AM

running on ECS (trying to), how can I set the equivalent of

prefect backend server

in ENV variables? since it's complaining about no API key

Eddie Atkinson

05/19/2022, 3:43 AM

Brain isn’t quite working today. I have a flow Parameter called

cluster

which determines whether a cluster is used which is accessed from a callback to

DaskCluster

. However, it’s not actually used in the flow which is causing flow run failures locally because it’s an ‘unexpected parameter’. I am 90% sure I’ve seen a way before where I can say to Prefect to relax and that it’s fine (technical terminology), but can’t remember exactly how

✅ 1

Raviraja Ganta

05/19/2022, 5:45 AM

Hey, I am getting this error when trying to register a flow. I am using protobuf for parsing input and output.

Copy code

TypeError: cannot pickle 'google.protobuf.pyext._message.MessageDescriptor' object

✅ 1

Guillaume Latour

05/19/2022, 8:20 AM

Hello everyone, I am using a dask cluster and I've run into this issue:

Copy code

distributed.protocol.pickle - INFO - Failed to serialize <Success: "Task run succeeded.">. Exception: cannot pickle 'lxml.etree.XMLSchema' object

which leads to

Copy code

distributed.worker - ERROR - failed during get data with <ip> -> <ip>

which at some point close the connexion

Copy code

distributed.comm.core.CommClosedError: in <TCP (closed)  local=tcp://<ip> remote=tcp://<ip>>: Stream is closed

this is managed by prefect by some retries (depending on configuration) and finally the agent raises an error and the flow is marked as failling

Copy code

distributed.scheduler.KilledWorker: ('concatenate_df_based_on_time-b91c06dc30f54c5084e9f5fe8b6b32a5', <WorkerState 'tcp://<ip>', status: closed, memory: 0, processing: 1>)

Do you have an idea on how to prevent this kind of error?

Jan Domanski

05/19/2022, 9:11 AM

Is the prefect2 setup for AWS ECS kind of the same as it was for prefect1? I followed this tutorial once and it was super-smooth for prefect1 https://towardsdatascience.com/serverless-data-pipelines-made-easy-with-prefect-and-aws-ecs-fargate-7e25bacb450c

✅ 1

Tom Manterfield

05/19/2022, 10:50 AM

Hey everyone! Can anyone tell me if the Orion API and/or Scheduler and/or UI support running multiple replicas? Is anyone running them in an HA config already?

✅ 1