prefect-community #prefect-community

matta

07/31/2021, 2:58 AM

Hrm, trying to make a Prod vs Test project, with separate CI/CD pipelines. Ideally, people won't have to think about anything but making a feature branch - but I'm using Git storage, so for now I do have to tell it where the feature branch is. It's not THAT big of a deal, but is there any way to set it up so that it'll "know" to tell the storage to look in the feature branch that it's on? Thanks!

Michael Warnock

07/31/2021, 3:45 PM

I'm trying to switch to S3 storage (from Docker, which was working). The flow shows up in cloud, but my local agent doesn't log anything about it. If I run a docker agent, it sees it, and complains about the storage type. I've reread all the docs on storage and agent types, and I don't know what I'm doing wrong.

Irvin Tang

08/02/2021, 12:24 AM

hi, i have a flow that interacts with a library that my team built for storing/indexing data. i added some additional logs in that library and in my

pyproject.toml

have a git dependency pointing to that test branch. after building the image with this new version of the library, registering the flow with that image, running the flow doesn’t display any of the additional logs i have another project where i’m doing the same thing. the flows in that project do log the additional messages. i’m not sure if this might be the way i configured prefect/how i’m registering the flows. has anyone ever encountered this problem or have encountered something similar? also please let me know if i was unclear about anything. thank you!

Reece Hart

08/02/2021, 1:03 AM

Hi Community. I need some advice about the kinds of workflows that Prefect is good for or, conversely, not so good for. I'm considering use Prefect to coordinate the execution of genomics data processing. The pipeline consists of ~50 steps, each of which ranges from seconds to 2 hours. The steps are all command line tools written in a mix of compiled C and C++, Java, Python, shell scripts, and even some Perl for bad measure. The pipeline starts with 100-150GB of data, processes that into a series of similarly large files, then analyzes those data to result in files roughly 100MB or smaller. Processing takes 250-500 CPU-hours (e.g., 5-10 hours with 48 threads on a EC2 c5ad.12xlarge). The current pipeline is coordinated by a Makefile, which works surprisingly well. My primary concern (and question for the community) is whether prefect is really suited to the tasks at hand. For example, one of the first steps looks roughly like this:

Copy code

seqtk (args) | fastp (args) | sentieon (args) | samtools (args) >out.tmp

In our pipeline, most steps are wrapped in a script, which is what the Makefile calls. This step starts with apx 100GB of data and dumps a 150GB of data. Given the data volume, I would be reluctant to write intermediate files in lieu of the pipes. Given the nature of workflow -- all command-line tools with file-based data -- I think adopting Prefect would amount to making most of our scripts into Prefect's ShellTasks. I wonder whether this is really worth the effort. The main drivers for choosing a workflow tool are to help with pipeline versioning, schedule and track jobs, to help orchestrate infrastructure scale up/down. Thanks for any guidance.

Omar Sultan

08/02/2021, 8:23 AM

Hi Everyone, We are using Prefect's imperative API , and wanted to use the GitHub storage option. All the tutorials refer to the use of the functional APIs when registering and storing script based Flows. If anyone can point me in the right direction this will be really appreciated

Samuel Tober

08/02/2021, 9:33 AM

Hi everyone! I am using .map() to run a task with a list as parameter, but I do not get a desired behaviour. For example, if I specify a parameter as:

markets = Parameter('markets', default=['stockholm', 'oslo'])

And then run a function:

Copy code

total_df = load_data.map(
    city=markets,
    date_from=date_from
)

where I pass a single value parameter, date_from, and my list parameter markets. Everything runs without error, however, the function is only run for the first value in the markets list. What I want is to run for each value in markets, using the same value of date_from for each element in markets. How can I achieve this?

Jai Deo

08/02/2021, 9:51 AM

Hi, I am trying to set up a flow with Azure storage and AKS for the agent and cloud server. The flow registers but when I try to quick run it I get the error 'Failed to load and execute Flow's environment: AttributeError("'NoneType' object has no attribute 'rstrip'") - not sure what I am doing wrong

aman gupta

08/02/2021, 12:07 PM

Getting this error while deploying the flow on PrefectCloud while locally executing it is working fine. We are trying to save the Results to GCS_RESULTS in buckets. Can anyone please assist with this? Unexpected error: PicklingError('Pickling client objects is explicitly not supported.\nClients have non-trivial state that is local and unpickleable.') 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 900, in get_task_run_state result = self.result.write(value, **formatting_kwargs) File "/usr/local/lib/python3.8/site-packages/prefect/engine/results/gcs_result.py", line 75, in write binary_data = new.serializer.serialize(new.value) File "/usr/local/lib/python3.8/site-packages/prefect/engine/serializers.py", line 73, in serialize return cloudpickle.dumps(value) File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps cp.dump(obj) File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump return Pickler.dump(self, obj) File "/usr/local/lib/python3.8/site-packages/google/cloud/client.py", line 166, in getstate raise PicklingError( _pickle.PicklingError: Pickling client objects is explicitly not supported. Clients have non-trivial state that is local and unpickleable.

Robert Hales

08/02/2021, 1:59 PM

Hi again, is it possible to set

max_retries

on a task at flow run time?

Bouke Krom

08/02/2021, 3:18 PM

Hey all, I'm trying to include my

Artifact

(just a link in our case) in a custom Slack notification. The State Handler gets a

Flow

and

State

. I'm having trouble finding the

Artifact

somewhere in the

Flow

objects. I guess I should try and find a

flow_run

object of some sort?

Mehdi Nazari

08/02/2021, 5:12 PM

Hey All, Can prefect’s secret retrieval tasks be done in a flow definition? I have a flow during which I’d want to retrieve a secret from Prefect Cloud and pass the value to a resident function (not decorated with task). I keep getting a

Task<name>

instance instead.

Miguel Angel

08/02/2021, 6:23 PM

I was wondering if is any pattern to expose modules to be containerized in a python way. The project I'm working on has the following structure:

Copy code

.my-package
├── __init__.py
├── _config.yml
├── flows
    ├── __init__.py
    ├── flow1.py
    ├── flow2.py
    └── flow3.py
└── utils.py

So I can expose my flows via storage object like this:

Copy code

from prefect.storage import Module

storage = Module("my-package.flows")

Each of the flows that belong to

my-package

have the following structure:

Copy code

from prefect import Parameter, task
from prefect.utilities import logging as logging

def core_function(**args)-> prefect.flow:
   # process flow
   return flow

flow = core_function()

Seem legit to me, since I haven't spotted any downside, do you have any recommended pattern or advise?

Krapi Shah

08/02/2021, 7:54 PM

Hi all, Is there any automation to send alerts in case of late flow runs? I have currently set an alert for job that does not start for 100 seconds, but this only triggers after it has been scheduled, In case the flow schedule gets delayed due to any reason like agent being down, there is no alert. Any help appreciated! Thanks

Kyle McChesney

08/02/2021, 8:01 PM

Hello, Does anyone have a bit more in-depth description of how results work? Specifically S3Results? What should I expect to find in those files for a given task. Example

Copy code

@task(result=S3Result('bucket', location='example.out')
def example():
  return [1, 2, 3]

Is it just a pickle file that when loaded, it recreated the list of

[1, 2, 3]

How does it work for more complicated returns, for example a task that returns a tuple or a pandas DataFrame?

Madison Schott

08/02/2021, 8:05 PM

Hi all, does anyone know how using a Fivetran task changes the Fivetran sync schedule of your connectors? Not sure if it's a coincidence but I just noticed that the tables I was testing using Prefect haven't been synced on my Fivetran UI in 6 days. Normally I have the schedule set to sync every 24 hours.

Philip MacMenamin

08/02/2021, 8:45 PM

Hey - I just ran a deploy of current, looks like 4200 is running, but 8080 is not.

Harry Baker

08/02/2021, 9:36 PM

Is it appropriate to re-use a task across multiple different flows, similar to how you would import a python function? I am trying to organize my code base by DRYing it out, and I find myself reusing the same code defining tasks to instantiate and return a google sheets API client. Would I be able to define a task in a file, that is then imported in multiple other files to be used across multiple flows? Or does it not work like that

Mehdi Nazari

08/02/2021, 9:53 PM

Hi again all, Best Practices question; what are practical ways to FAIL a flow when a single task fails? also, good ways to send out notification when that failure happens?

Leon Kozlowski

08/02/2021, 10:18 PM

Hi all, have there been any thoughts to add a choice concept to

Parameter

? Something like:

Copy code

choice_param = Parameter(
    name="choice_param",
    choices=["some", "choice", "params"],
    default="some"
)

Sumit Kumar Rai

08/03/2021, 4:02 AM

Are there any online resources that will help me create development environment for Prefect projects?

Scarlett King

08/03/2021, 9:56 AM

Hey, I’m trying to pass a date parameter to a unmapped variable. I’m having trouble reading the parameter value. Can someone help me out please?

Copy code

snapshot_date = Parameter(‘snapshot_date’, default=dt.datetime.now().strftime(‘%Y-%m-%d’))
run = apply_map(full_flow, params=params, snapshot_date=unmapped(snapshot_date))

And inside full_flow

Copy code

def full_flow(params, snapshot_date):
	# ..

Copy code

snapshot_date = dt.datetime.strptime(snapshot_date, ‘%Y-%m-%d’)
	print(f’{snapshot_date:%Y%m%d}’)
	# ..

It keeps giving me error because the Parameter object is passed instead of a string. How can I access the parameter value only?

Samuel Kohlleffel

08/03/2021, 2:48 PM

Hi, I'm trying to build out a registration process that uses

flow.serialized_hash()

to only register the modified flows. However,

flow.serialized_hash()

is not returning a consistent hash value for flows that have not been modified. Why would this be the case? For context, I'm testing by registering the flows locally with

flow.register()

Kyle McChesney

08/03/2021, 4:05 PM

Has anyone worked extensively with ShellTask? Specifically I am looking to use

.map

and include the mapped value into either the command or the helper_script. For example:

Copy code

@task
def files():
    return [
        '/opt/file1.txt',
        '/opt/file2.txt',
        '/opt/file3.txt',
        '/opt/file4.txt',
    ]

rm_task = ShellTask(
    command='rm $file',
)

with Flow('shell') as flow:
    files_to_delete = files()
    rm_task.map(files_to_delete, helper_script='file="{mapped_value}"')

Mehdi Nazari

08/03/2021, 4:41 PM

Hello All, Is there a way to define a placeholder on a defined Parameter from the Cloud UI interface? I’d want prefect to fill the place holder at the run time basically!

Miguel Angel

08/03/2021, 4:45 PM

Have anyone worked with a local environment for testing in a, lets say, docker-compose fashion? Suppose that you have an aws ECS cluster to execute your flows and Prefect Cloud as backend.

Philip MacMenamin

08/03/2021, 4:51 PM

I'm getting "Skipped (metadata unchanged)" upon registering a flow that I've updated (as in, the source code has changed, and is being ignored.

Constantino Schillebeeckx

08/03/2021, 5:22 PM

hi gang! what's the best practice way of registering flows that aren't contained to just one

.py

file? we've split out commonly shared functionality between flows into an e.g.

utils.py

file which is referenced in the flow. given all the storage documentation this design doesn't seem to fit into the intended use of storage.

David Elliott

08/03/2021, 6:32 PM

Hey all, I've been looking at upgrading my prefect version and also my distrbuted, dask & dask-kubernetes versions for our production pipeline, and just wanted to clarify some change in behaviour that I've noticed? • Previously when I ran a flow, the k8s agent would create a job which was in effect the dask scheduler, creating and retiring pods as it needed to. In my case that

prefect-job-xxxxx

would create 4 ephemeral dask workers (named something like

dask-root-xxxx

) • Now the behaviour I'm seeing is: ◦ K8s agent creates the

prefect-job-xxx

◦ In the

prefect-job

logs, it gives me _prefect.DaskExecutor | Creating a new Dask cluster with

__main__.make_cluster

.Creating scheduler pod on cluster. This may take some time._ ◦ there are then 5x

dask-root-xxx pods

created, where 1 of them is a dask scheduler - ie the scheduler no-longer sits within the
prefect-job-xx
? Just wanted to check if this was expected/intended behaviour - I couldn't see any reference to it in the prefect release notes • In addition, (and this is more a side note that I think the prefect k8s rbac needs updating) - I've had to add 2x more rulesets to my k8s RBAC to make it work - see these docs for what's now required. Here is specifically what's changed vs the prefect docs Thanks!

Billy McMonagle

08/03/2021, 8:24 PM

Hi all, I'd like to add a "create project if not exists" script into my build pipeline (for git branch based work). I know I can do

prefect create project "My Project"

... is it safe to run this command multiple times?

Jeff Baatz

08/03/2021, 9:00 PM

For some reason on Cloud, I'm noticing that most of the UI elements aren't loading properly. Is anyone else experiencing this? I've tried it across a few different browsers. It's unfortunately bad enough that I can't really use the UI at all since almost none of the text loads. I tried clearing caches and all that good stuff already. EDIT: Oh wow, you know what... Dark mode was checked, but it doesn't seem to be working well. I forgot that toggle existed.

🙌 1