Hey all, I'm running into an issue with prefect cloud submitting jobs but not doing anything after that. The issue first presented with a new flow I was writing. I restarted my agent and I upgraded from 14.6 to 14.12. This only had the effect of spreading the issue to the rest of my flows. I was wondering if you could offer me some guidance in debugging this issue.

Hi all, i have this problem. When i use datetime or pendulum library, and then register a flow, this will always take register datetime , not the current time of the running flow. My code. Any suggest?

import prefect 
import pendulum

cron_now = pendulum.now()
str_date = cron_now.strftime('%Y%m%d_%H%M%S')
custom_schedule = CronSchedule("0 9 * * 0", start_date=cron_now)


def slack(text):
	data = '{"channel":"XXX","text":"%s: %s"}' % (str_date, text)

@task
def task():
	## do something
	##call_slack
	slack("Hello")
	
with Flow("fact_czenk", schedule=custom_schedule) as flow:
	task = task()
	
flow.register()

Hey team! Just wondering, does prefect have prometheus integration inbuilt already? We’re looking for some method by which to set up alerting if a task fails and our other service utilise prometheus for this. If not, what other methods have people use / are recommended to keep track of failures? Can you set up slack messages on any task failure?

hello, prefect team. I see there are the two methods available for flow registration. Are both equally recommended? 1. prefect cli [1] 2. functional API

flow.register

context: we want to create a dynamically defined flow (e.g.

Flow(result=S3Result(location=f'bucket/{<http://args.my|args.my>_result_key}', ...)

for flow reuse. To parse user-pass args, we instantiate the Flow behind

__main__

. Now we noticed we can’t use prefect cli to register because it can’t pass extra user args. Alternatively, we are considering

flow.register

and wonder if there would be any risks we should be aware of. [1]: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/cli/register.py#L70

Hi everyone, I upgraded my Prefect server from

0.14.6

to

0.14.12

and since then I see even when idle, CPU hovers between 10%-50%.

docker stats --all --format "table {{.ID}}\t{{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}"

shows the following:

CONTAINER ID        NAME                CPU %               MEM USAGE / LIMIT
c6eda45c6f59        tmp_ui_1            0.00%               4.953MiB / 7.432GiB
2466edd82357        tmp_towel_1         0.00%               52.57MiB / 7.432GiB
e9d0643d134e        tmp_apollo_1        2.69%               62.68MiB / 7.432GiB
ad41dcf8646a        tmp_graphql_1       3.24%               69.5MiB / 7.432GiB
a8f65bcb2efa        tmp_hasura_1        4.62%               153.8MiB / 7.432GiB
bc9742482bd8        tmp_postgres_1      4.64%               27.86MiB / 7.432GiB

Hi all, I'm looking for some help with StartFlowRun. Executed the following running on my local machine:

prefect backend server
prefect server start
prefect create project etl-project
prefect agent local start --label etl-label
prefect register flow --file simple_flow.py --name a-simple-etl-flow -l etl-label -p etl-project

In the UI, I can click on QUICK RUN and observe the flow execute. Next , in another simple script named start_flow_run.py:

from prefect.tasks.prefect.flow_run import StartFlowRun

kickoff_task = StartFlowRun(
    project_name='etl-project',
    flow_name='a-simple-etl-flow'
)

which I execute like this:

python start_flow_run.py

But nothing happens. The agent doesn't detect it; no activity in the UI. I was expecting my registered flow named 'a-simple-etl-flow' to execute. Please advise. Thanks.

Is it possible to fetch the entire log from a certain flow rown as text or json without interacting with the ui? Maybe by using curl on a prefect-server UI link? EDIT: easy. using graphql 🙂

Hi all, I just got started on prefect (I mostly used AF before for orchestration) and it's great ❤️ Two quick questions: • I was able to manually write Python code that runs my great expectations checkpoints - the GE built in task does not work with my LegacyCheckpoint, and I wasn't able to make even a simple flow run (below). Did anybody get the GE integration work? Is that worth the trouble, when full "verbose" GE code to run validation is 4 lines of Python? • I re-build the DBT cloud client from AF to work as a Prefect task (two tasks really, scheduling the job, and polling DB cloud for the JOB status) - as I prefer DB cloud to DB shell locally installed. Is there any interest from the community in this sort of integration? If yes, happy to chat / share the simple POC I built. GE CODE THAT DOES NOT WORK

from prefect import task, Flow, Parameter
from prefect.tasks.great_expectations import RunGreatExpectationsValidation

# Define checkpoint task
validation_task = RunGreatExpectationsValidation()

with Flow("ge_test") as flow:
    validation_task(checkpoint_name='gitter_checkpoint')

flow.run()

Hey team! Im trying to use the

on_failure

callback of a flow to send a message to a slack channel that ideally looks something like “OH MAN THE FLOW FAILED - Click here to see the flow” with a proper link. Does the flow object contain any information I can use to construct a useful URL, specifically the flow_id?

I am trying to raise a RETRY signal inside a mapped Task. The follow up task is using

flatten

to gather all of the results back up. When

RETRY

is raised I am getting the following:

TypeError: object of type 'RETRY' has no len()

Does anyone have an idea?

Hi folks, the way it currently stands, one has to re-register a flow to modify the schedule of a flow - is that correct ?

Hi Team, I am currently facing an issue where my flow references a module within the same repository but is unable to load the module at execution time. I am using Prefect Cloud, GH Storage, and a Kubernetes Agent. Seems it could potentially be a pathing issue as I must set

PYTHONPATH

in my local environment, but not having much luck setting a similar ENV variable on the agent or job_spec. Any suggestions or documentation that might help get over this hump? Repository Structure:

/src
--/flows
----/flow.py
--/tasks
----/task.py

Error Message:

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named 'src.tasks'")

Hey team, is there a way to tell prefect “Retry all failed flows in the last day/hour/etc”?

Hi Team, I'm running into ECS permission issues while trying to deploy a flow-- Though I'm specifying ecs cluster on agent start, I'm getting a permission error on resource *. This user only has permissions to the ECS cluster listed. Any thoughts? Thanks.

prefect agent ecs start --cluster my-cluster-arn --launch-type EC2

botocore.errorfactory.AccessDeniedException: An error occurred (AccessDeniedException) when calling the RegisterTaskDefinition operation: User: arn:aws:iam::**********:user/********* is not authorized to perform: ecs:RegisterTaskDefinition on resource: *

How would I share a variable between two tasks? I don't want to return it in the result object because the data grain is off, but I need it for an evaluation.

Hello community! Is there a log that tracks when flow schedules were turned turned on or off on Prefect Cloud? Our flow runs seem to magically turn off sometimes, and we’d like to be able to track why that happened. Additionally, we’ve noticed that the schedule slider can be a little finnicky, sometimes showing that a schedule is on when there are no upcoming scheduled runs (or vice-versa), or take a long time to enable/disable.

Hey everyone! Friendly reminder that @Laura Lorenz is about to get started live streaming about DevOps for Prefect Server — come learn about the Server helm chart and all the info you ever wanted about deploying and maintaining Server! Link here:

When using ECSRun how can I configure it to not assign a public ip to tasks it launches?

Not sure where to post/report this, but we pretty often have problems accessing the docs:

Hey team, what is the best practise for accessing binary files from a flow? We're processing a fair amount of weather data, stored in a binary format (netcdf). We're running the kubernetes agent, on Azure, with volumes mounted on the pods running our flows with a custom yaml file. It does works but is slightly brittle

Hi all 👋 I'm running prefect server on a single node private server and have a LocalAgent process running on the same machine and querying the API for flow runs. The use case is ETL. I just noticed that the same flow, when launched from the command line with flow.run(), takes 30s to execute, and when launched from the UI with "quick run", takes 5 minutes. The first task of the flow returns a list of 200 files, and the subsequent tasks are mapped to handle these files one by one. Any idea why this 10x difference in execution time occurs ? Many thanks!

Hi, I've noticed that the scheduler service sometimes stops scheduling new runs. Nothing is logged by the towel container, the last message is a standard "Sleeping for 300.0 seconds". Restarting the services resolves the issue. I have a slight suspicion that this is occurring when one flow run overruns past the start time of the next run. But I see no relevant GH issues in either the

prefect

or

server

projects, so wondered if maybe I'm just misunderstanding something about how it's supposed to work. Is anyone aware of this already?

Hey folks, is it possible to dynamically add tasks to a flow while the flow is running? I'm reading through the flow_runner.py code and it seems like if it's possible, it's...involved.

I have a flow where the vizualization is adding a bunch of List->List->Dict boxes which makes it hard to read. How can I remove those?

standardizer_task = submit_standardizer_job(files, bucket, 'mcule', version)
standardizer_wait_task = AWSClientWait(client='batch', waiter_name='JobComplete', max_retries=2, retry_delay=datetime.timedelta(minutes=1))(waiter_kwargs={'jobs': [standardizer_task]})

Hey! I'm looking to schedule a flow to happen daily, but I don't want the server to be running locally. What is the best way/best agent or executor to use to achieve this?

Just want to clarify something about Cloud concurrency: even if you have multiple agents, the ability to run tasks is still hard limited by the concurrency limit correct?

Hello everyone, We're currently evaluating Prefect as our workflow scheduling solution. So far we love the simple API and architecture, especially with Prefect Cloud, which makes it very easy to get going. We do have a question however: What is the best practice for project structure, packaging and CI/CD with Prefect Cloud and Kubernetes Agent? We have multiple jobs across multiple github repos. These are generally various modules in various packages. These packages are mostly added to custom docker images tailored for other systems where we deploy. They also have complex dependencies from other public and private PyPi repos. Thanks, Alex P.S. Moved further details in the thread

Hi Team, I'm trying to access result of a failed task with the following code.

cluster = KubeCluster.from_yaml(dask_worker_spec_file_path)
    cluster.adapt(minimum=1, maximum=10)
    executor = DaskExecutor(cluster.scheduler_address)
    state = flow.run(executor=executor)

but when I try to access the state I'm getting this.

Python 3.8.6 (default, Dec 11 2020, 14:38:29)
Type 'copyright', 'credits' or 'license' for more information
IPython 7.21.0 -- An enhanced Interactive Python. Type '?' for help.

In [1]: state
Out[1]: <Failed: "Unexpected error: TypeError('Could not serialize object of type Failed.\nTraceback (most recent call last):\n  File "/usr/local/lib/python3.8/site-packages/distributed/protocol/pickle.py", line 49, in dumps\n    result = pickle.dumps(x, **dump_kwargs)\nTypeError: cannot pickle \'_thread.RLock\' object\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File "/usr/local/lib/python3.8/site-packages/distributed/protocol/serialize.py", line 307, in serialize\n    header, frames = dumps(x, context=context) if wants_context else dumps(x)\n  File "/usr/local/lib/python3.8/site-packages/distributed/protocol/serialize.py", line 58, in pickle_dumps\n    frames[0] = pickle.dumps(\n  File "/usr/local/lib/python3.8/site-packages/distributed/protocol/pickle.py", line 60, in dumps\n    result = cloudpickle.dumps(x, **dump_kwargs)\n  File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps\n    cp.dump(obj)\n  File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump\n    return Pickler.dump(self, obj)\nTypeError: cannot pickle \'_thread.RLock\' object\n')">

any idea what I'm doing wrong?

Hi, Iam new to prefect, Trying to run snowflake query with prefect, PFB

import prefect
from prefect.tasks.snowflake.snowflake import SnowflakeQuery
from prefect import task, Flow

query = """
    SHOW DATABASES;
"""

snowflake_def = SnowflakeQuery(
    account="account",
    user="user",
    password="****",
    database="***",
    warehouse="****",
    role="***",
    query=query
)

with Flow("hello-snowflake") as flow:
    snowflake_def()

flow.register(project_name="tutorial")
flow.run()

when i trigger quick run from UI, Iam facing below issue

Unexpected error: TypeError("cannot pickle '_thread.lock' object")
Traceback (most recent call last):
  File "/opt/prefect_env/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/opt/prefect_env/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 "/opt/prefect_env/lib/python3.8/site-packages/prefect/engine/results/local_result.py", line 116, in write
    value = self.serializer.serialize(new.value)
  File "/opt/prefect_env/lib/python3.8/site-packages/prefect/engine/serializers.py", line 73, in serialize
    return cloudpickle.dumps(value)
  File "/opt/prefect_env/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 72, in dumps
    cp.dump(obj)
  File "/opt/prefect_env/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 540, in dump
    return Pickler.dump(self, obj)
TypeError: cannot pickle '_thread.lock' object

I made Checkpoint as FALSE