Actually, I've tried that above, but it does not work

I'm having trouble understanding how Contexts work. This code always seems to return None

prefect.context.get('flow_run_name')

How do I get the flow run name (like zealous-dormouse) from inside the flow?

can the Prefect

SlackTask()

be configured to send messages to different channels depending on what flow it is being run in?

Hi all, Can i prevent run two instance of a scheduled flow?. I dont found any configuration to do that. Thanks for your help

👋 hello from Chicago! When configuring a flow using run configs, is it possible to provide extra data (think a small dictionary whose values are strings or ints) that will show up in

flow.serialize()

? Will share more details in thread, thanks!

👋 hello all! I’ve been playing around with the great expectations validation task using the documentation here: https://docs.prefect.io/api/latest/tasks/great_expectations.html#rungreatexpectationsvalidation. I don’t have

validation_operators

in my

great_expectations.yml

file so I call the task as follows:

validation_task = RunGreatExpectationsValidation()
    validation_results = validation_task(
        batch_kwargs=validation.get_batch_kwargs(records, 'pandas_s3'), expectation_suite_name=GE_EXPECTATION_SUITE_NAME, validation_operator=None
    )

However when I execute my flow I get the following error:

Unexpected error: DataContextError('No validation operator `None` was found in your project. Please verify this in your great_expectations.yml')
Traceback (most recent call last):
  File "/Users/soren.daugaard/Code/data-pipelines/.direnv/python-3.8.6/lib/python3.8/site-packages/great_expectations/data_context/data_context.py", line 1272, in run_validation_operator
    validation_operator = self.validation_operators[validation_operator_name]
KeyError: None

According to the documentation that parameter should be optional, but I am not able to figure out how to set it to nothing. I tried

None

and the empty string but with similar errors. Any pointers would be much appreciated.

Hi ! I am trying to deploy Prefect Server and UI on Amazon ECS (Fargate). But I have a problem… Deploying seems to succeed. However, when I opened the Prefect UI screen, the page did not seem to display correctly. When I look at the console of Google Chrome, I get the following error.

Uncaught (in promise) Error: passed invalid or empty tenant object
    at l.setDefaultTenant (index.js:51)
    at vuex.esm.js:840
    at vuex.esm.js:462
    at Array.forEach (<anonymous>)
    at vuex.esm.js:461
    at l._withCommit (vuex.esm.js:620)
    at l.commit (vuex.esm.js:460)
    at l.commit (vuex.esm.js:405)
    at tenantNavGuard.js:12
    at u (runtime.js:63)

Does anyone have any idea how to solve this problem ? By the way, the architecture on AWS is as follows.

client - ALB - ECS (Fargate)

hello, hello

started playing with Prefect a bit on the weekend, i have some questions about the "docker-compose" setup, e.g., how I would role my own. e.g., I don't have python on the host and would like to run it in containers a 100%. I also tried the suggested docker image, but that drops me into a shell. the docs are a bit opaque here. would anyone have some pointers? 🙂

Hello team, is it possible to use custom adjustment implementation for shcedule?

hii, does anybody now how to declare "from datetime import datetime" as python dependency in docker storage?

flow.storage = Docker(registry_url="<http://gcr.io/keller-steering-enabling/flows|gcr.io/keller-steering-enabling/flows>", image_name="billwerk-contracts",
                      python_dependencies=["pandas", "oauthlib ", "requests", "requests_oauthlib", "datetime", "oauth2client", "snowflake", "datetime"], secrets=["GCP_CREDENTIALS"])

above doesn't properly work thanks!

Is there a way to authenticate the client other than using prefect auth login? I am trying to use prefect inside a CI/CD pipeline and it would be ideal to set an environment variable instead.

Hi, I tried to install prefect on my local machine to test it. However I am stuck after starting it via docker-compose. I created the docker-compose.yml file via prefect server. However I can not connect to the server with the prefect agent local start command. This exits with a 'message': 'No tenant found.'. Also the UI on localhost is mostly empty. What am I missing? Logs from docker-compose up seem fine.

Hello, need some help with mapped/ upstream tasks. I'd like to run a mapped task, that will only start upon the success of another task but I am doing sth obviously wrong in the following simple example ( probably thinking of the parameter upstream_tasks wrong). Thanks in advance for your help

Question, if I have multiple agents with the same labels, they seem to be running the tasks in a flow once each...that seems wrong

Hey there, I had 2 flows that were running for a few days without realizing they were using up all the concurrency slots I had available in Cloud. I am trying to cancel all the “late” runs and essentially reset my scheduler… I have tried to do so via the interface, but it does not seem to clear the late runs out.. further, as I try to cancel the 2 flows that have been running, I am stuck in a “Cancelling” state. Is there any way someone can help me reset the current state of my Cloud? I can certainly kill the k8 pod that is running, but would rather it happen via UI. Thanks!

What happens when you run

prefect register flow --file ... --project ...

? Is the python code in the file executed? Similarly, if I am using Prefect cloud, and a k8s agent that spawns jobs; what env vars are available when: 1. The job is starts the python file (thereby declaring the flow / DAG) 2. The flow is run

Hi just moving this over from the wrong channel: I recently finished porting one of our pipelines over to prefect and have been eager to move out of local and over to an EKS/ECR configuration. I saw some guides by Anna and Jimmy that were really helpful so far, but I've hit multiple collisions and was seeking some help to work through some of them. Specifically two blockages I've encountered involve using our own custom base image, which is hosted on ECR, which is actually pulling code from CodeArtifact; that approach is hitting healthcheck issues when building. Another issue I'm having is being unable to create new EKS clusters through eksctl (where CloudFormation spits out max # of elastic IPs reached) , but I'm assuming that's more of a provisioning/permissioning issue and for that I'm just wondering if anyone else has gotten stuck at these points as well. I can however deploy an EKS Agent and view it in kubectl get pods, and publish a flow on Prefect Cloud regardless of these blockers, so I feel like I'm tangled up in the last few steps in getting this to work. (Thanks Dylan for pointing me to this channel!)

hi folks, how to specify a secret with key, value pairs and to read it with PrefectSecret or Secret().get()? Thanks!

Can I supply secrets from env vars from the k8s agent's deployment?

What call translates a Task into an ordinary value? E.g. what translates a

Parameter

to a value in a

def

?

Hi all, question about logging… 🧵

?

I'm trying to configure the final bits of my install on Kubernetes. I'm looking at https://docs.prefect.io/api/latest/cli/agent.html#kubernetes-install — vs the

start

command that also takes

--job-template <path>

— how do I configure my own job template when using

prefect agent kubernetes install

?

Hi, I'm trying to find out what is the recommended way to bulk load data from csv into the database (eg postgres) regularly in prefect. I understand I can do "insert into table ... " and use postgres.PostgresExecuteMany task, but I'm concerned that for very large files it might be slow. Is there a way to execute pg copy (or COPY), or "insert into" is how I should do it in prefect?

@Maria You could use a shell task to execute psql COPY, etc

That all relies on the dependencies being there for you to use it though.

I am using BatchSubmit and I need to pass parameters to the job so I am passing

n_estimators = Parameter('n_estimators', default=500)
BatchSubmit()(..., batch_kwargs={
'parameters': {'blah', n_estimators}})

The issue is that batch requires the values for parameters be strings but I want the input Parameters to be integers. Is there a way to convert a Parameter from one type to another? Should I have a task that takes the parameter and returns the string version?

Hello Prefect team. I wonder how to properly raise a FAIL signal from within a task without being retried. Retries is mostly to handle intermittent failures. When we set

retries=3

for a task_A, we want to retry task_A for up to 3 times when it fails due to some network connection issue or intermittent failures. Meanwhile, however, we have the logic in the

task_A

that explicitly and intentionally raise

FAIL_A

signal for downstream tasks. That

FAIL_A

is not meant to be re-tried by

task_A

. When

task_A

encounters this intentional

FAIL_A

, it shall skip retries and return it directly. How could I do that?

I’m trying to use a parameter as a condition in a flow. What I’m finding is that if the condition is False, then every task after it is skipped.

with Flow('MyFlow') as flow:
    do_thing = Parameter('do_think', default=False, required=False)

    data = get_data()

    with case(do_thing, True):
        data = do_thing_task(data)

    data = do_another_thing(data) # This gets skipped when do_thing = False