prefect-community #prefect-community

Peter Roelants

06/09/2021, 7:14 PM

Hi Prefect, I started seeing the following typing errors when checking my code with `mypy`:

Copy code

error: Signature of "run" incompatible with supertype "Task"

For example for following Task:

Copy code

class RestTaskTest(prefect.Task):
    def run(self, string_val: str, int_val: int) -> TaskTestResult:
        return TaskTestResult(string_val=string_val, int_val=int_val)

This makes sense since the definition of

Task.run

def run(self) -> None:

. Is there any way to make mypy work together with overriding of Task.run in custom tasks? Or is the only option to ignore these with

# type: ignore

Zach Schumacher

06/09/2021, 7:40 PM

I assume the answer is yes, but just want sanity check. are datetimes on the context all UTC? https://docs.prefect.io/api/latest/utilities/context.html#context

Jonathan Chu

06/09/2021, 7:57 PM

what's the difference between

Flow Concurrency

and

Task Concurrency

? i have a flow running right now, and i can see it shows as

1 running flow (100%)

but then

0 running tasks (0%)

Leon Kozlowski

06/09/2021, 9:20 PM

Hi all, I’m in the process of migrating a mono-repo of scheduled tasks over to prefect. Does anyone have any deployment patterns they utilize for registration via CLI while maintaining each ‘service’ as an independent docker image with a Dockerfile in each service subdirectory? (Example Below)

Felipe Saldana

06/09/2021, 10:14 PM

I am just looking at the examples and I dont see many that have a main function. Is there a reason why they are not used?

if __name__ == "__main__":

Brett Naul

06/09/2021, 10:29 PM

we've recently been seeing StartFlowRun create multiple runs with the same

idempotency_key

<24 hours apart (say 30 mins). is the idempotency key queryable from graphql so we could try to debug what's going on?

Ben Collier

06/10/2021, 5:17 AM

Hi all - I’ve been trying to use

Copy code

client.get_flow_run_info(flow_run_id)

with a flow run id in the format “c2686800-3c0f-4b67-83b2-f85fa6fd6773” I’m getting the following:

Copy code

'message': 'invalid input syntax for type uuid: "c2686800-3c0f-4b67-83b2-f85fa6"'

I assume I’m doing something fairly stupid - could someone explain why the UUID is in the wrong format?

Talha

06/10/2021, 8:27 AM

Hi, I am trying to use Git repository for storage and run my flows but the flow is not working giving error that the unable to find directory. Can you help me here. Is there any working use case example which I look at which uses git Storage ? Thanks !

Noah Holm

06/10/2021, 9:22 AM

Based on this page, it looks like you can use environment variables to set values for agent options. Is there any documentation on which environment variables map to which agent options? The only page I can find is here which doesn’t mention the environment variable option. So far we’ve been creating our own image with a custom entrypoint for the agent to set all the option but it would be super neat if we could avoid that.

Thomas Nyegaard-Signori

06/10/2021, 11:05 AM

Hi all, I am still dipping my toes in the Prefect-ocean and I have some questions regarding flows that consist of tasks that all run some custom Docker image on a Kubernetes (AKS) cluster. Currently I have set up an in-cluster Kubernetes agent with the

prefect agent kubernetes install -rbac ...

command, so the RBAC is functioning on the agent. When starting a very simple flow that consists of a single

RunNamespacedJob

task running the custom Docker image the job pod starting the flow runs into RBAC issues, but the

RunNamespacedJob

task pod runs fine. My question is, how to handle job pods that are going to spawn several jobs on Kubernetes and the issues that arise with the RBAC on these pods. Am I thinking about this incorrectly? The error, for reference:

HTTP response headers: HTTPHeaderDict({'Audit-Id': '67b79e3e-ab13-45ee-8ad5-2ae1769c6a7f', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Content-Type-Options': 'nosniff', 'Date': 'Thu, 10 Jun 2021 09:21:02 GMT', 'Content-Length': '372'})

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"jobs.batch \"cmems-historical\" is forbidden: User \"system:serviceaccount:prefect-zone:default\" cannot get resource \"jobs/status\" in API group \"batch\" in the namespace \"prefect-zone\"","reason":"Forbidden","details":{"name":"cmems-historical","group":"batch","kind":"jobs"},"code":403}

Dmytro Kostochko

06/10/2021, 1:34 PM

Hi all. We think about using Prefect for ETL for our project and I try to implement some POC. Basic idea is load documents from Elasticsearch(

detect_new_documents

), load data that should be extracted(

get_skill_extractor

). Also this data should be periodically reloaded from database if new version was added. Save results(

save_skills

). I use such flow:

Copy code

with Flow("Analyze new documents") as flow:
    documents = detect_new_documents()
    extracted_skills = extract_skills.map(documents, unmapped(get_skill_extractor()))
    save_skills.map(documents, extracted_skills)

And looks like I need some stateful worker to keep instance of skill_extractor in memory and share for all tasks, because now it is passed with tasks to Dask workers, and serializing/deserializing takes more time then processing. Maybe someone can give some advice how this can be implemented with Prefect?

Devin McCabe

06/10/2021, 1:50 PM

Prefect repo maintainers: Would someone mind taking another look at this closed issue and my comment at the bottom? I think there's a pretty serious (but subtle) bug regarding what happens when prefect modules are imported. Namely, merely importing a class or method from a Prefect module does things before you've even called it. https://github.com/PrefectHQ/prefect/issues/4331 Would be happy to write up a new issue if that would be helpful.

Christian Eik

06/10/2021, 2:29 PM

hey, is there a way to prevent prefect from writing what my tasks return to the results folder? I have checkpointing turned off, and config values in the task below return False.

Copy code

@task(state_handlers=[post_task_fail])
def add_headers(headers: list, data: tuple) -> List:
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(prefect.config.flows.checkpointing)
    <http://logger.info|logger.info>(prefect.config.tasks.checkpointing)

    return headers + list(data)

*edit: but -> and config values ...

John Ramirez

06/10/2021, 7:47 PM

Hey everyone! Can the Progres Tasks support connection to an SSH Forward Tunneler object. Also can I have an example of how to use the Postgres task

Ben Muller

06/10/2021, 8:50 PM

Hi All, I am trying to launch my

prefect agent

in an ECS Fargate cluster. I am feeding in all my secrets via the cdk namely, RUNNER_TOKEN_FARGATE_AGENT
,
AWS_SECRET_ACCESS_KEY
and
AWS_ACCESS_KEY_ID
I am getting some type of auth error when I run my image command

Copy code

CMD           [ "prefect", "agent", "ecs", "start", "--token", "$RUNNER_TOKEN_FARGATE_AGENT", \
                    "--task-role-arn=arn:aws:iam::********:role/ECSTaskS3Role", \
                    "--log-level", "INFO", "--label", "fargate-dev", "--label", "s3-flow-storage", \
                    "--name", "fargate-demo" ]

The trace in 🧵 `````` Any ideas would be great! I assumed I dont have to register the agent with Prefect cloud as I did this with Ec2 and it didnt have to be done. Thanks in advance for any help on this...

Michael Law

06/10/2021, 8:55 PM

Hey folks, I was wondering if anyone had some docs or advice around the best way setup running agents on an AKS cluster? I am wondering, do I need to package my jobs up to ACR or some other container registers? I’m struggling to identify from the docs how the DockerStorage and the KubernetesRun config is used. I would find a diagram useful here to show the relationship between all of these things. I don’t suppose anyone can ELI5 it for me or had a diagram at hand which details the architecture of it all? Thanks again.

Tim Enders

06/10/2021, 9:52 PM

Is anybody using the Great Expectations tasks, and do they work with the newer v3 Great Expectations API?

joshua mclellan

06/10/2021, 10:09 PM

hey so im working on my first real flow as part of a POC for my org and im just trying to wrap my head around what the tradeoffs are in regards to a flow of flows or one big flow. We have a lot of already made tasks and im not sure if making too many flows will cause issues in the long run or of having one huge flow will have its own problems. Any thoughts on what the tradeoffs are and how one should go about balancing against the two?

Robert Bastian

06/10/2021, 10:13 PM

Hello Prefect! Is there a way to limit users access to specific projects in Prefect Cloud?

Ben Muller

06/11/2021, 4:10 AM

Hi Prefect devs, me again. I am oh so close to finish my deployment system for prefect for my company but am running into an issue. I have my agent deployed in a cluster and running successfully but then I am trying to execute a flow with a menial logging task to make sure everything works. It is running in the same cluster as the agent. When the task definition spins up the task, it is hit with an error:

Copy code

ResourceInitializationError: unable to pull secrets or registry auth: execution resource retrieval failed: unable to retrieve ecr registry auth: service call has been retried 1 time(s): AccessDeniedException: User: arn:aws:sts::*****:assumed-rol...

Both my execution role and task role have admin access ( while i am debugging ) so i dont think it is a role based issue. Wondering if anyone has come across something like this before and can help out. Oh and this is running on ECS and fargate, I have attached my flow code in the 🧵

Shivam Shrey

06/11/2021, 7:09 AM

Hi I have the following pub-sub use case to solve with Prefect flows : • Task A triggers a service via HTTP. Upon successful invocation, the service returns a MQ topic which needs to be subscribed to. • Task B needs to subscribe to this topic (via MQ). The service (it is a long running process) asynchronously publishes on this topic which the Prefect task waits for and then perform subsequent tasks. What are the best ways to do Task B in Prefect?

Emma Rizzi

06/11/2021, 12:59 PM

Hello, I'm trying to use Prefect Server deployed on a AWS instance with a ECS agent running on a ECS cluster. I created a small hello world task to print 'hello' in the logger, which runs correctly when I create a Local agent on the same instance where the server is deployed, but I can't manage to get the cluster working... Heres what happens : • I created a ECS Agent running on my instance with Prefect Server, and configured with my cluster arn • i register my flow and quick start on the UI, on AWS console I can see the task begin created, with status going from Pending to Running to Stopped. Although the task definition contains the tag [INACTIVE] which seems weird 🤔 • On prefect UI the task is stuck at the "Submitted" state and nothing is displayed in the logs I tried to follow this and this and added task_role_arn and execution_role_arn with lots of permissions but it doesnt seem to be the problem. I also tried to add GCS as storage option, no success. I have no idea of what's causing this issue so here I am hoping someone knows 🙂

Pedro Machado

06/11/2021, 1:36 PM

Hi everyone. I am trying to use

PrefectResult

in a flow that has a resource manager. The idea is to easily allow flow restarts without having to set up a different result backend. However, I am running into an issue with the resource manager tasks. The

setup

method of the resource manager returns an object that is not json serializable which causes the task to fail. Do you have any suggestions to workaround this? Is the only solution to use a different result backend for this task? Thanks!

Marie

06/11/2021, 2:04 PM

Hi everyone, I'm trying to bind the outputs of a task that returns 2 results to 2 different tasks. It would be easy with the functional API but I have to use the imperative API for an unrelated reason. I have been trying to use

bind

a few different ways without being able to separate the outputs of the first task (see simple example below). Does anyone know how to do this? Thanks!

Copy code

flow = prefect.Flow("example-flow")

flow.add_task(task_a)  # Task A returns 2 outputs
flow.add_task(task_b)  # Task B uses the first output of task A
flow.add_task(task_c)  # Task C uses the first output of task A

task_b.bind(job=task_a, flow=flow)
task_c.bind(report=task_a, flow=flow)

Talha

06/11/2021, 2:13 PM

Hello, I have using Git storage for my flow and I have successfully set up the Git repository. Now I am running my local agent. When I run the flow using my Local agent I get the following error: "Failed to load and execute Flow's environment: NotADirectoryError(20, 'The directory name is invalid')". But if I run a local agent through a docker container using docker-compose file. It works fine. Can someone please help me what can be possible problem. I am lost here. I am trying to find anything in documentation that can be helpful, but haven't had much of success

Nathan Atkins

06/11/2021, 4:13 PM

Is there a way to connect to the Dask UI when running with LocalDaskExecutor? Do I need to run a local cluster and DaskExecutor to see the UI? I was expecting to see it at localhost:8787/status. I have Bokeh installed.

Wai Kiat Tan

06/11/2021, 4:51 PM

i tried to login to prefect cloud but it keeps directing me to https://cloud.prefect.io/welcome/name-team. is it not accessible at the moment?

Joe

06/11/2021, 7:34 PM

Hey there! Total noob question, but I'm having trouble finding what I'm looking for elsewhere. Hypothetical: I have five servers running Prefect agents (A, B, C, D, E) all connected to one Prefect Server (F). I write five different Python scripts to run on the different agent servers. Do those scripts need to get loaded onto the five different agent servers (A-E,) or do they all get loaded to the Prefect Server (F) which will then magically run each script on whatever agent machine I specify? Basically, will I have to create a deployment pipeline that drops scripts in one location, or five locations?

Lucas Fobian

06/11/2021, 8:06 PM

Hey guys. New to prefect, but already love it. I was wondering if there is a way to debug/view the intermediate result of a task during the flow.run() call? Some task in my pipelines messes up my final output and I would like to be able to quickly tab over the intermediate results of the tasks to see which is causing this behavior. Thank you!

06/11/2021, 9:16 PM

How to make Prefect.io format file handler so that log entries will be just like stdout I am trying to have a Prefect.io log file, created by file handler, to look exactly like the log the is printed out by the Prefect.io logger, but I get just the message, without the metadata such as the log level. Any ideas on how to do that ?