Hi folks, we are getting failures for all of our flow runs after updating to prefect

v0.14.15

basically the failure happens after the flow has completed running at the state handler level - please see the traceback in the thread

Hi. I am having some trouble making

prefect register --module

work. My project looks like this:

flows
  + __init__.py
  + my_flow.py
  + shared_tasks
    + __init__.py
    + util.py

When I run

prefect register --project 'Prefect Testing' -m '<http://flows.my|flows.my>_flow'

, I get

No module named 'flows'

. What am I missing?

Hello. I'm still test driving Prefect (v0.14.15), but my flow stucks and I get zombie processes. Any ideas? Here's the code without imports:

SQL_DIR = Path('sql')

@task
def list_query_names():
    return [f.name for f in SQL_DIR.glob('*.sql')]

@task(log_stdout=True, timeout=15, task_run_name='{name}-{date:%F_%T}', checkpoint=False)
def exec_query(name: str):
    sql = Path(SQL_DIR / name).read_text()
    print('Query name: %s' % name)
    engine = sqla.create_engine(DSN)
    rs = engine.execute(sql)
    return dict(keys=rs.keys(), rows=rs.fetchall())

@task
def save_results(rs, name):
    with (OUT_DIR / name).with_suffix('.txt').open('w') as f:
        csv_writer = csv.writer(f, delimiter="\t")
        csv_writer.writerow(rs['keys'])
        csv_writer.writerows(rs['rows'])

with Flow("Queries") as flow:
    query_names = list_query_names()
    results = exec_query.map(query_names)
    save_results.map(results, query_names)
    
flow.executor = LocalDaskExecutor(num_workers=2, schedule='processes')
flow.run()

To all the adventurous apple M1 users out there (and in general everyone working with different OS):

How do you build your prefect flows to ensure that the docker images run on different/the desired architectures/OS?

Is there any way to have one LocalAgent running multiple flows at the same time?

Hi - I am having an issue where a user has issues viewing/ joining the group via the web interface. Any advice is welcome.

How can mapped tasks be checkpointed/cached? A default Flow level result with

location="{flow_run_id}/{task_run_id}.prefect_result"

works for the unmapped tasks, but I don't see saved results for the mapped tasks (even though each one has a unique

task_run_id

).

Hey all! Looks like Vincent might be having the same issue as me, but I am unable to add users with the correct role via the web interface to our tenants. Is there anyone from Prefect who could help me out with this? I see the error: Looking for role-based access controls? This feature is only available on Enterprise plans; check out our pricing page for more details. I can confirm we are on an enterprise plan. We did recently create 5 new tenants under our business license, but this issue is affecting the previous tenant with RBAC.

Hey, Does anyone have any good tips for hunting down why a Flow does not serialise?

how do usually manage staging and production versions of a flow? i.e. for a single flow, deploy the latest version to staging, run it on staging, if it looks good, then push that version to production is there anything better than just copypasta the flow definition code, just with a different flow name, like a

data-ingestion-flow-staging

and

data-ingestion-flow-production

?

Hi everyone, we know we can set default parameter to schedule. But is it possible to set default FlowRun name with some prefix for a schedule? Example: in schedule A, the flow run will be named as schedule_a_run_2020-01-01, and schedule B will be schedule_b_run_2020-01-01 Thanks in advance!

Hey! There's a problem. Just as an example, say there's a website that has a list of tv shows, a a page for each tv show has links to information about every episode. I want to scrape all this info and save everything into a table (url, show_name, episode_name, episode_description)

Hi! Sorry if this is the wrong place to ask this question 🙇 Does someone know if there are any plans to support Google Cloud Run as an agent? It's the Google equivalent of Fargate. I mainly use Google infrastructure and I'm thinking of automating some tasks through Prefect, so trying to figure out what options there are 😊 Thanks!

Following up a question that was successfully answered yesterday. When you run an agent (in my case local agent) + Cloud, what flows does it listen to? Any flows within the account (deduced from the token give)?

Hello. I have a little question: I found an example: https://github.com/PrefectHQ/prefect/blob/master/examples/old/task_library/mysql/mysql_flow.py how I can make a host as a parameter?

What is Prefect-ish way to end a flow run. Suppose one of the early tasks in a flow does not have the required data to continue the rest of the flow, how does one gracefully end a flow?

Hello, Is there a way to export Prefect flow definition to some config format like YAML, and then re-build the flow based on the same imported config?

Is there a reason that flow.result and flow.executor are not serialized for flow.register, unlike flow.run_config and flow.storage?

Hello, I am looking to use Prefect to help scale some large batch processes with ECS (Fargate) and Dask and am wondering if anyone has any insight on doing this or could point me to some resources.  Additionally, I am having some difficulty figuring out why my submitted Flow’s ECS Task immediately becomes INVALID when provisioning resources.

Did anyone try to build a multi stage Docker image for use with prefect? I'm using python:3.8-buster as the base for the builder and python:3.8-slim-buster for the main file, but I get errors like this when trying to do a KubernetesRun:

Pod prefect-job-6ab817cf-d6446 failed.
	Container 'flow' state: terminated
		Exit Code:: 1
		Reason: Error

DockerRun doesn't even continue after pulling the image.

Anyone know what it means when you attempt to register a flow with

prefect register

and you don't get any error/warning messages but only:

================= 0 registered, 1 skipped =================

Is there a clear documented example where a Prefect Flow is defined in the entrypoint script in a Docker image and that Flow can be invoked/executed from the Prefect UI? All of this running locally.

Hi everyone. I'm trying to implement dynamic secrets usage using

PrefectSecret

Task I want to pass the secret key via

Parameter

. But it says that PrefectSecret argument should be statically defined so impossible to use parameter as secret key. Is that any workaround to solve that? Thanks in advance!

Hi! How I can add the tasks in runtime? For example: I am waiting for Parameter where should be count how many same tasks I want to create.

Hi all, we are currently facing a problem where a flow run in Prefect Cloud takes significantly more time (factor 20/25) than the same flow run in Prefect Server. The flow retrieves data from a mysql database and inserts the data into a sql server database. When running the flow in Prefect Server, with Docker storage and a Docker agent I get around 1000 rows/sec insertions into the sql server db. While in Prefect Cloud, with Docker storage and a Kubernetes agent I get around 40 rows/sec insertions. Any ideas on the cause of this performance issue?

Hello guys! Sorry for a stupid question, but maybe someone knows if it is possible to pass

host

Parameter as

MySQLExecute

argument? Pseudo code:

from prefect.tasks.mysql.mysql import MySQLExecute
from prefect import Flow, Parameter

example_query = MySQLExecute(
    name="example_query",
    db_name="db name",
    user="some user",
    password="123456",
    port=1234,
    query="select * from example_table;"
)


with Flow("example") as f:
   db_host = Parameter("db_host", default="host_address")
   
   mysql_execute = example_query(host=db_host)

when I execute this pseaudo code, I get this error:

TypeError: __init__() missing 1 required positional argument: 'host'

According to documentation: https://docs.prefect.io/api/latest/tasks/mysql.html#mysqlexecute ,

host

argument is not optional, so by default it is required to be defined in a

MySQLExcute

function. Maybe there is a proper way to pass

host

value to the

MySQLExecute

function?

Any tips on how to further investigate root cause for:

Unexpected error: ValueError('Could not infer an active Flow context.')

What does it mean when on invoking a system call from within a task, a <Parameter> is used(?) instead of the expected string?

--destination_bucket_name <Parameter: destination_bucket_name>

when was expecting:

--destination_bucket_name myorg-S3-bucket

Hey, we've recently started exploring Prefect and we'd like to schedule Spark jobs on Cloud Dataproc - I've noticed the excellent integration with Databricks but didn't see anything re. running jobs on other Spark platforms/distributions. Am I missing something? Has anyone implemented this already? Thanks a lot for your help!