prefect-community #prefect-community

How would I execute a flow from Prefect Cloud/UI on a dask cluster in the absence of a dask agent? Seems as though dask assumes responsibility for what the agent normally does, but without an agent how do we indicate to prefect cloud where to find the dask cluster?

Matt Drago

12/03/2020, 5:03 AM

Hey Folks, I'm building a flow that will use the MySQLFetch task and I have noticed that I cannot use Secrets to populate the user and password fields. I noticed this thread from a few months ago, but I cannot find an open issue in github. I'd like to work on this issue, is it best for me to raise the issue or is there another way I should go about it?

Joël Luijmes

12/03/2020, 12:31 PM

Hey question! I’m runing Prefect in Kubernetes, and I began working on a flow which will fire up 10 Kubernetes jobs. However, currently it is spawning these jobs sequentially, while I want to make them start all immediately. In order to do so, I deployed dask to the cluster (with https://helm.dask.org), used the DaskExecutor and point it to the scheduler. However, the same is still happening 😞. Are there other configuration required in order for this to work? I tried running the flow locally with the DaskExeuctor, and then it does its magic in parallel 🙂 (Note: tested it with a different flow.) I’ll post the details in thread.

✅ 1

Dimitris Stafylarakis

12/03/2020, 1:22 PM

hey all! I have an issue with using Docker storage; I am trying to use a pre-built base image, however I get an error (

requests.exceptions.HTTPError: 404 Client Error: Not Found for url: <http+docker://localhost/v1.40/images/create?tag=latest&fromImage=>...

) and docker.errors.ImageNotFound when I try to register the flow. Others have mentioned similar errors in the past, but as far as I saw this was when trying to deploy the flow, not when building the image. Any clues/hints/tips would be appreciated! I'm on OSX using docker-desktop btw, just the default config. Thanks in advance 🙂

Payam Vaezi

12/03/2020, 3:05 PM

I'm trying testing prefect server and prefect agents within local kubernetes on my machine. I have all pods up and running and submitting job to kubernetes agent. However, I'm getting

Kubernetes Error: Back-off pulling image "4988c227fd72"

error when I want to pull an image I built on my local docker. I'm passing the image ID in the environment of graphql call

"metadata": {"image": "4988c227fd72"}

. Any idea why my local kuberenetes agent can't pull a local image?

Daniel Rothenberg

12/03/2020, 3:41 PM

Hi community - I have pair of tandem flows that are essentially a crude training/prediction workflow; one flow is intended to run once day, train an ML model, and archive its output to Google Cloud Storage, and the other flow is intended to run once an hour, grabbing the latest trained model it finds in GCS as well as the latest input data from the past hour, making a prediction, and shooting that off to a database. Prefect has worked amazing for this so far... it was super easy to integrate an agent into my team's k8s cluster, which has a lot of useful infrastructure including shared NFS mounts between pods and a dedicated

dask-gateway

to spin up distributed workflows. Major kudos to the team! I have a question though about scaling out this workflow. In my particular application, I basically need to apply this training/prediction workflow for O(10,000) different "assets". Put another way - my workflows each have a

Parameter

called "asset". I see two different ways of scaling my toy with one asset to the much larger set of a few thousand: 1. Use the

.map()

capabilities on my tasks and simply "map" a list of "assets" for the flow to run through; or 2. Convert my "asset"

Parameter

to be a list of all the assets, and do all the mapping inside each of my tasks in the workflow - by leaning on the fact that each task already has access to the dask cluster I'm running the workflow on I'm not sure which is the "recommended" approach here. The training/modeling flow generates, for each asset, artifacts on the order of 1MB or so that have to get passed around, but I'm worried about the performance of Prefect's scheduler if it needs to map to tasks over a list of several thousand items. On the other hand, things get... complicated (with my naive workflow for saving trained models and whatnot)... if I have to manually do all the book-keeping inside each task in the flow. Thoughts? Does anyone have experience scaling out ML training/prediction workflows on Prefect that can offer some insight?

Ben Fogelson

12/03/2020, 3:56 PM

Does prefect support task memoization through either results or targets? I.e. I want to cache task results based on their inputs, not based on a timestamp or a specific flow run.

👀 1

Chris Jordan

12/03/2020, 3:58 PM

Hey, I've got a question that I haven't been able to find a clean answer to in the docs. I'd like to have a flow evaluate the results of one of its tasks and then conditionally spawn a new version of itslef. Something like:

Copy code

@task(name="get next batch of records")
def get_batch(result=PrefectResult()):
    ...
    return len(records)

with Flow("import_flow") as flow:
    num_of_records = get_batch()
    if num_of_records.read() > 0: # this particular syntax doesn't do it, and I'm asking for how to read this here (if I should be)
        kickoff_task = StartFlowRun(project_name="imports", flow_name="import_flow") # StartFlowRun also doesn't seem to spawn a new task - is this the right way to call this?

Brian Mesick

12/03/2020, 5:05 PM

Hey folks. We’re running into a weird issue where running a Docker stored flow version 0.13.18 (also tried 0.13.13) against a 0.13.13 Kubernetes agent and getting the error:

Copy code

Usage: prefect execute [OPTIONS] COMMAND [ARGS]...
Try 'prefect execute -h' for help.

Error: No such command 'flow-run'.

Looking through the history of the Slack it looks like this was a known issue with 0.13.0 agents, but I’m not seeing any other possible explanations. Any hints?

Julio Venegas

12/03/2020, 5:35 PM

Hi everyone, what’s the better pattern to finish a looping Flow as successful that has been scheduled indefinitely. As of now, the main Task of the Flow raises

ENDRUN(Finished(message="all rows from file read"))

when it has finished reading all the lines in a big file (I’m not using the LOOP signal, the Flow is an exercise to properly use

Result

objects). Should I create a state handler and trigger a final reference task or is there something like

ENDRUN

but at the Flow level? Any other alternatives?

Richard Hughes

12/03/2020, 7:35 PM

Hi, I saw a flow that had 2 PID's in the logs, and for some reason the flow executed 2x instead of one time from a scheduled execution. This caused an issue in my environment. How can I narrow down how this happed? flow-run a24b31dc-16ea-45ce-9816-1eccb4cbb34a

Alex Joseph

12/03/2020, 8:06 PM

Hi, I've a mapped task creating n different subtasks. I want to persist+checkpoint the output such that if tasks completes, it doesn't need to be rerun (but the other subtasks are rerun) for the next run. I can implement this by using the

target

option, but I don't seem to be seeing any option to add the function arguments to template the output. For example, I'd like to have:

Copy code

@task(target="{single_pattern}_{config['a']}.txt", checkpoint=True, result=LocalResult(dir="cache/"), )
def run_spark_job(single_pattern, config):
   ...

I can template parameters, date etc, but I don't seem to have way to template the arguments themselves. What's the correct way to handle this situation? Thanks 🙂

Lee Mendelowitz

12/03/2020, 10:33 PM

Hey Community - are there any examples out there for how to use the AWSClientWaitTask? (https://docs.prefect.io/api/latest/tasks/aws.html#awsclientwait). Specifically I’m trying to wait on an AWS Batch job to finish. It’s not clear to me how to pass the job name or job id into the task. Thanks!

JD Margulici

12/04/2020, 1:14 AM

Hello, I'm a new user to Prefect, currently building a SaaS application that involves document digitization and a business intelligence layer. One question I have been asking myself and that seems to elicit a good amount of discussion is whether it is possible to compose flows. I have seen proposals for subflows, and there is also a PIN on combining tasks: https://docs.prefect.io/core/PINs/PIN-05-Combining-Tasks.html. The PIN concludes that the model won't change and that "we have decided [...] [to] write clear documentation for how to build individual "combined" tasks for individual use cases". Is that documentation available? With all the proposals floating around, it is hard to separate between what is implemented and recommended practice vs. proposals and speculations. So in short: what is the recommended way to group tasks into logical units of work and make the overall application modular with respect to these groups?

Andrew Hannigan

12/04/2020, 1:44 AM

With Environments being deprecated, what is the suggested replacement for something like DaskCloudProviderEnvironment?

Severin Ryberg [sevberg]

12/04/2020, 9:23 AM

Is it already possible for Prefect's Lambda-related tasks to take advantage of AWS Lambda's new support for container images? This is a game-changer, in my opinion, and I'm really eager to try it out! I imagine it could be something as "easy" as the DaskKubernetesExecutor, but with the serverless-goodness of Lambda 😄. If not already supported, is this something that's on the development radar?

Neeraj Sharma

12/04/2020, 11:59 AM

What is authentication mechanism on prefect?

David Kuda

12/04/2020, 4:03 PM

Hey everyone! I have been testing Prefect for two weeks now. We soon want to test it on some real projects! 🙂 I am trying to accomplish some things, but I don’t succeed yet … There is for example this issue which prevents me from advancing really: How do I change the port for database? This is how I do it now: I run

prefect server start --postgres-port=5435 --ui-port=8081

-> that works. And now I want to change it so the

config.toml

handles it. So in my home directory/.prefect (~/.prefect) I have the file `config.toml`with following content:

Copy code

toml
[server]
	[database]
		host = "localhost"
		port = 5435
		host_port = 5435
		url = "<https://localhost:5435>"
		connection_url = "<postgresql://prefect:test-password@localhost:5435/prefect_server>"
		name = "prefect_server"
		username = "prefect" 
		password= "test-password"
		volume_path = "/Users/david/.prefect/pg_data"

I have tried many things with this config.toml file … but nothing has worked. When I try

prefect server start

, I receive an error:

ERROR: for t_postgres_1  Cannot start service postgres: Ports are not available: listen tcp 0.0.0.0:5432: bind: address already in use

. What’s going wrong? When I write something wrong, say I use a colon instead of an equal sign,

prefect server start

returns yet another error, which is good. I have examined the possible config options with `prefect config`; I did not see any other setting for database / postgres. I am following the docs very precisely, I even did the YouTube tutorials which were hosted by Laura, which were all great. Best Regards from Berlin!

Hui Zheng

12/04/2020, 6:38 PM

Hello, @Kyle Moon-Wright We ran into a strange issue with Flow to prefect-cloud deployment. The context 1. We use

flow.storage = Docker()

for build our flow deployment. 2. we then use

flow.storage.build(push=False)

to build the docker container locally and test it run locally. 3. lastly, we use

flow.register()

to deploy the flow to prefect-cloud All 3 steps works fine on my local machine. I could deploy the flow to prefect-cloud successfully. My colleagues wants to do the same on their local machine, however, when two my colleagues did it, they ran into a healthcheck issue at

step 3

(See issue detail in the thread). we are using the same code base and using the same build-context environment. We have setup another docker container in which we build and deploy prefect flow, so that it ensures that we all have the same run-time and libraries when doing build-and-deploy. This is a high priority issue for us, because currently only I could do the flow deployment. It put high risk on our production downtime if my colleagues could not path and deploy the prefect flow when there is emergency incident situation. cc: @jars @Julie Sturgeon prefect version:

Copy code

prefecthq/prefect:0.13.15-python3.8

DJ Erraballi

12/04/2020, 6:54 PM

https://docs.prefect.io/orchestration/concepts/flows.html#toggling-heartbeats we are running into a task that is long runing ~ 1hr that is getting killed by the zombie killer

DJ Erraballi

12/04/2020, 6:56 PM

Is our best option to disable heartbeat? by calling mutation on the current flow id?

DJ Erraballi

12/04/2020, 6:56 PM

is there an easy way to figure out why heartbeats would stop getting sent?

DJ Erraballi

12/04/2020, 6:59 PM

Copy code

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

Daniel Nussbaum

12/04/2020, 7:18 PM

Apologies if this is a basic question -- we are considering using Prefect Cloud for our ETL tasks. I cannot find specifically what data is sent from Prefect Server to Prefect Cloud -- is there a document outlining the data being sent and security standards / policies more generally? I'd appreciate anyone who can point me in the right direction here!

Alex Joseph

12/04/2020, 7:41 PM

While trying to use

S3List

with

prefix

and

delimiter

(trying to list "folders" in an S3prefix, say) I get an empty array:

Copy code

S3List(config['input_bucket_name']).run(prefix=test, delimiter="/", max_items=10)

If I run it manually using boto, I'm able to get results:

Copy code

import boto3

    client = boto3.client('s3')
    paginator = client.get_paginator('list_objects')
    result = paginator.paginate(Bucket=config['input_bucket_name'], Delimiter='/', Prefix=test)
    for prefix in result.search('CommonPrefixes'):
        print(prefix.get('Prefix'))

Is this expected behavior?

Hui Zheng

12/04/2020, 7:50 PM

Hello, I am implementing Timeout for one of my flow. I have a few questions. 1. how to get the flow to enter

Timeout

state when timeout happens to a task. https://docs.prefect.io/api/latest/engine/state.html#timedout It seems there is only signals.FAIL instead of signals.TIMEOUT. 2. the fow DAG: A > B > C > D > E. Where E is always running with

trigger=triggers.always_run

. when timeout happens to task A, B, C, D, the task raises an exception

TimeoutError

, which is not a

FAIL

signal. and it causes the flow to terminate without running E properly. To overcome that, Currently I have to implement the exception handler in every task, it’s very tedious. Any better way to do it?

Copy code

Try:
            except TimeoutError:
                raise signals.FAIL("Task Timeout.")

👀 1

Mark McDonald

12/04/2020, 8:01 PM

Happy Friday Prefect community! I just wanted to share this delicious flow run we recently had. I hope you're all loving Prefect as much as I am!

😂 2

🍗 7

😄 7

Hui Zheng

12/04/2020, 8:44 PM

Hello, I ran into another issue with timeout setting. I would like to set a default timeout for all the tasks. I shall be able to do it through environment variable

PREFECT__TASKS__DEFAULTS__TIMEOUT

according to this doc

Copy code

<https://docs.prefect.io/core/concepts/configuration.html#environment-variables>

However, when I do it properly, I got an error.

TypeError: Only integer timeouts (representing seconds) are supported.

even when I assigns the integer value. (See thread for details)