prefect-community #prefect-community

prefect-community

Daniel Saxton

05/17/2022, 10:10 PM

hi all, what's the best way to have a flow that's only triggered via API call (i assume this is possible)? say we're using Prefect Cloud

David Beck

05/17/2022, 10:20 PM

Hi all! for some reason, I have a Task class that doesn't seem to be registered with my test flow to Prefect Cloud. Our set up is a CI/CD process which packages flows with KubernetesRun config. When registering, there are no obvious errors or logs indicating the necessary files for the Task are left out or anything of that sort. What can I do to troubleshoot this further?

davzucky

05/18/2022, 1:30 AM

I started to look deeper into Orion. One of the think I'm missing is the story about secret. Do you have any update about it? I can see that aws collection has support for read_secret. However this return a string which will be serialized in clear in the storage. Compare to prefect 1 which was using a custom SecretResult this is less secure.

✅ 1

Jan Bršťák

05/18/2022, 8:24 AM

Hello, we are running ECS Agent with

prefecthq/prefect:1.2.1-python3.9

image, and we are using Prefect Cloud. All seems to be fine with the Agent, we can see it in the UI, and also in Cloudwatch logs, whenever we trigger a flow run, it says

Deploying flow run…

and

Completed deployment of flow run

. But no Flows are getting started, it gets stuck on Submitted every time (it worked before, but we changed image for agent from

prefecthq/prefect:0.14.13-python3.8

and also roles in AWS which have the same permissions as original had). There are no errors, and nothing in logs. I’m not really sure what to do? Thanks for any help. Run config looks like this:

RUN_CONFIG = ECSRun(

_labels_=["prod"],

_task_role_arn_="arn:aws:iam::XXX:role/prefectTaskRole",

_execution_role_arn_="arn:aws:iam::XXX:role/prefectTaskExecutionRole",

_task_definition_arn_="prefect-task:4",

_run_task_kwargs_=dict(

_cluster_="XXX",

),

✅ 1

Florian Guily

05/18/2022, 10:16 AM

Hey, so far i developped flows locally and tested them locally. Now that they seems robust enough, it's time for production. This woud be on aws but i'm quite new about that and a bit lost among aws services. I red this article on how to deploy but i'm still confused, especially about where the kubernetes agent is running ? i suppose this is the only thing that needs to run all the time. I'm also a novice about k8s but if i understood correctly, this is the best option regarding scalability. Any suggestion on ressources to look at to do so ? thanks !

Daniel Sääf

05/18/2022, 12:32 PM

Hi! I finally had some time to dive into the world of prefect (partly motivated by the great interview in Talk Python With Me) I am trying to get prefect2.0 to run for a simple workflow - and everything was working really well when i followed the tutorial but i ran into an error as soon as i stepped out from the tutorial 🤦 (i realize that this might be something simple - but i’ve tried everything i could come up with and have been reading the docs a couple of times.. So excuse me if this is something obvious - i’m just stuck.. ) The problem occurs when i’m trying to create a deployment for my flow that runs well locally. My current setup is that i have the leo_flow.py and leo_deployment.py. (According to the tutorial) But i added a file test.py (in the same directory as leo_flow.py and leo_deployment.py) from which i import a function called hello_world in the leo_flow.py (from test import hello_world) As i said, i can run the flow as expected:

>>> result = leonardo_dicaprio("leo")

However, when i try to create the deployment i get a:

ImportError: cannot import name 'hello_world' from 'test' (/usr/local/Cellar/python@3.10/3.10.4/Frameworks/Python.framework/Versions/3.10/lib/python3.10/test/__init__.py)

So basically, my question is: how do i import functions from local python-files in a flow? (Let me know if this isn’t the right place for this type of questions!)

yair friedman

05/18/2022, 12:44 PM

Hi , first - thanks for the wonderful package! i have created a basic scikit pipeline flow , and i am using the DaskExecutor as the executor. i wonder if the scikit-learn algorithms are really using dask to run in a parallelised distributed manner or the wrapper task are distributed but the actual ML work in it is running locally…. is it the same as running daskML algorithms ?

🎉 1

✅ 1

Krzysztof Ciesielski

05/18/2022, 1:10 PM

Hello, I'm trying to wrap my head around the

max_retries

retry_delay

task parameters. I wanted to wrap some of my tasks with retries, and I thought it should be enough to add these parameters to the

@task

annotation. However, when a task raises an exception, it shows in the Prefect Cloud UI that it's state is "Retrying Task (after attempt 1 of 61)", but after a short while the entire flow gets cancelled. Example:

@task(log_stdout=True, max_retries=60, retry_delay=timedelta(seconds=10))
def perform_something(param1: Optional[str],
                              param2: any,
                              year_month_str: str,
                              environment_label: str) -> datetime:

Alvaro Durán Tovar

05/18/2022, 1:22 PM

Hi, Given a registered flow can I easily modify one of it's parameters via graphql? Ideally creating a new version of the flow

Ben Collier

05/18/2022, 1:56 PM

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:

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:

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

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:

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

Which is registered with:

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:

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

Registering

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?

Title

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?

for logs I meant the flow run logs we see in Prefect Cloud UI

Kevin Kho

05/18/2022, 10:44 PM

Log is stored in the database that comes with Prefect Cloud. Parameters of past flow runs can also be queried using the GraphQL API

Shaoyi Zhang

05/18/2022, 10:48 PM

Thank you. Does “the database that comes with Prefect Cloud” means Prefect team has access to the logs and past parameter values in some degrees? Can we trust that even Prefect SDEs/DEs cannot retrieve the logs or will you be able to do so?

Moreover, is there a way to route parameters and logs to our own database/storage systems when using Prefect Cloud?

Kevin Kho

05/18/2022, 10:54 PM

We could retrieve the logs (to some extent) to help look at what happened to customer Flows. If you want to prevent something from being logged, you could create a filter to prevent those from being sent.

Shaoyi Zhang

05/18/2022, 10:54 PM

Thank you!

Kevin Kho

05/18/2022, 10:55 PM

You can add your own handler to the logger and there is this setting you can turn off

:thank-you: 1

Parameters pass through the API to kick off a Flow run so those runs are kept

View count: 4