prefect-community #prefect-community

itay livni

01/08/2021, 2:04 AM

Hi - I got an error message in Prefect Cloud

Failed to load and execute Flow's environment: SyntaxError("invalid or missing encoding declaration for '/home/.../MOC/etl_moc.py'")

-After the flow began and executed some tasks successfully. The flow was run locally. Is this a bug or something not configured correctly on my end? Thanks

Joël Luijmes

01/08/2021, 12:32 PM

When I cancel a flow in the UI, it stops further execution of tasks right? I got two questions/remarks: • Is the active task killed in some way? Or will it wait until it completes? • When I use a resource manager, I noticed that unfortunately the cleanup will not be called 😕 . Is there some way I can cancel my flow, but still cleanup resources? Maybe force it with a state handler, if state handlers still are triggered?

Dan Corbiani

01/08/2021, 2:31 PM

Hi, I'm new to prefect and I'm trying to use it to evaluate the performance of code / parameter changes. I've created a basic flow but cannot figure out how to get historical metrics on each of the tasks. Is there something that I'm missing? My hope would be to get a table that looks like: | Flow Name | Parameters | Task Name | Start Time | End Time | Duration | I've tried going through the graphQL api and I've looked through the UI. Any direction would be appreciated.

Jeff Williams

01/08/2021, 2:39 PM

Is there a place where the exit codes are documented? When I stop the server on my GCP implementation, I see several components exit with code 137. This may be perfectly normal, but it would be nice to know what the exit codes mean. Also, the window that I started the server in does not return to a command prompt when it is stopped. Is that normal? It tells me that "tmp_ui_1 exited with code 137" and that is the last message I get.

Marwan Sarieddine

01/08/2021, 2:58 PM

Hi folks, we ran into an issue where one of our child tasks ran until Success then was restarted for some reason causing our flow to crash. Any idea why this happened ? and is there a way to prevent this ? We are running a kubernetes agent and a dask kubernetes execution environment on EKS...

Vipul

01/08/2021, 3:27 PM

I am trying to setup a project with Prefect and I do have setup and teardown process. I could register the flow and project during the setup and could delete the project during teardown but I am not sure how can I remove the entry for Prefect Agent from Prefect Server. Any suggestion? Thanks

Matt

01/08/2021, 8:28 PM

Has anyone used Prefect to manage Dataform runs and mind sharing with me how they designed it?

👀 1

watching 1

matta

01/09/2021, 2:46 AM

Getting this whenever I try to run a Flow with a Mapped task on Prefect Server:

Copy code

Failed to retrieve task state with error: ValidationError({'_schema': 'Invalid data type: None'},)
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/prefect/engine/cloud/task_runner.py", line 193, in initialize_run
    map_index=map_index,
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 1387, in get_task_run_info
    state = prefect.engine.state.State.deserialize(task_run_info.serialized_state)
  File "/usr/local/lib/python3.6/site-packages/prefect/engine/state.py", line 362, in deserialize
    state = StateSchema().load(json_blob)
  File "/usr/local/lib/python3.6/site-packages/marshmallow_oneofschema/one_of_schema.py", line 144, in load
    raise exc
marshmallow.exceptions.ValidationError: {'_schema': 'Invalid data type: None'}

Using 0.13.19 right now Looks like some kind of serialization problem?

Dr. Jones

01/09/2021, 11:26 AM

I get this error message by the Prefect agent when a scheduled flow is about to start. Can you help me? 🙂

Copy code

[2021-01-09 11:20:53,118] ERROR - My_Agent | 400 Client Error: Bad Request for url: <http://apollo:4200/>

The following error messages were provided by the GraphQL server:

    GRAPHQL_VALIDATION_FAILED: Cannot query field "run_config" on type "flow_run".

The GraphQL query was:

    query {
            flow_run(where: { id: { _in: ["84b340e4-9f74-425b-9abd-27247fdc3759"] }, _or: [{ state: { _eq: "Scheduled" } }, { state: { _eq: "Running" }, task_runs: { state_start_time: { _lte: "2021-01-09T11:20:50.061124+00:00" } } }] }) {
                id
                version
                state
                serialized_state
                parameters
                scheduled_start_time
                run_config
                flow {
                    core_version
                    name
                    storage
                    id
                    environment
                    version
            }
                task_runs(where: { state_start_time: { _lte: "2021-01-09T11:20:50.061124+00:00" } }) {
                    serialized_state
                    version
                    task_id
                    id
            }
        }
    }

The passed variables were:

    null

Prash S

01/09/2021, 6:41 PM

Hello everyone 🙂 I'm trying to run flows on a Kubernetes cluster (Minikube for now) and I'm struggling to with an issue -- I see agents alive and responsive, but they don't pick up any tasks. I'm running on Prefect Cloud, using the DaskExecutor with Docker storage (passing in a Dockerfile which built fine). Code/more details in thread. Would appreciate any help, thanks!

jeff n

01/10/2021, 12:02 AM

Sorry if this is basic. I am trying to use GitHub as a storage location for my flow. However the flow has a requirement on another local python file. Since the agents are supposed to use the storage to pull down the flow code for running.

Josh Pitts

01/10/2021, 2:17 AM

Hi all, I’m new to Prefect, and we are hoping to use build out our data workflows using prefect. I am trying to run the local file example from the docs:

Josh Pitts

01/10/2021, 2:17 AM

Copy code

@task(target="func_task_target.txt", checkpoint=True, result=LocalResult(dir="~/.prefect"))
def func_task():
    return 99

Josh Pitts

01/10/2021, 2:17 AM

but no file is created.

Josh Pitts

01/10/2021, 2:18 AM

when I run this in a flow via flow.run() Can anyone advise where to explore? The docs are not so clear on this.

Josh Pitts

01/10/2021, 2:30 AM

Though that works to set

PREFECT__FLOWS__CHECKPOINTING=true

, what is actually happening regarding

target

checkpoint=True

, and

result=LocalTarget()

? or are those just ignored?

Josh Pitts

01/10/2021, 3:12 AM

Also, should this pattern be used as a good practice for a final output that is a file system file? or should we just write our own file handling

Danny Vilela

01/10/2021, 11:35 PM

Hi everyone! I’m trying to better understand the Great Expectations plugin for Prefect. I currently have a ETL pipeline I’d like to Prefect-ify into tasks, which doesn’t sound too difficult. I would also want to use Great Expectations to declare expectations around any given task’s inputs and outputs (e.g., assert some input’s column is unique, or that the output table doesn’t drop any rows). My question is: is this when I’d use the Great Expectations plugin? As far as I can tell, I don’t need the heavier offerings from Great Expectations; just inline data assertions to make sure my pipeline is “correct”. Please let me know if this question is better suited for a different channel or the Great Expectations Slack group. Thank you!

Pedro Machado

01/11/2021, 4:04 AM

Hi there. I am having trouble with shell tasks and logging. For example, when I run

DbtShellTask

, it doesn't show output why it's running or when it succeeds. The only time I see the detailed output is when there is a failure. I tried passing these args and it still doesn't show streamed output in the logs, not even after the task completes.

Copy code

return_all=True,
log_stdout=True,
log_stderr=True,

I also set the

PREFECT__LOGGING__LEVEL=DEBUG

everywhere I could think of: prefect server, agent, docker storage, run config. I see DEBUG level messages on Prefect Server, but nothing from the task. I am running 0.14.2 Any ideas?

Maurits de Ruiter

01/11/2021, 9:54 AM

Quick question. I'm looking for a way to start and wait for a Fargate Task. In the docs I see possibilities to start Lamda's, but nothing about ECS and Fargate. Am I correct to assume this isn't possible by default?

Jarek Piotrowski

01/11/2021, 1:27 PM

Hello Prefect Community! I’m working on a new tool that solves the main problem we currently see in documentation: it is very far from the codebase! That is why we develop a tool that enables linking code with documentation, so that: • developers or technical writers responsible for Prefect documentation can get notifications on Github when a code that is linked to docs changes, • new contributors, while reading documentation, can quickly view source code connected to a given functionality so that they can quickly understand the codebase. Do you think that such a tool would be useful to you? More information and preview is on our website: https://www.hastydocs.com All feedback is very appreciated! Cheers, Jarek

Vincent

01/11/2021, 2:24 PM

I want to ask the Dask Experts whether they can help identify a scheduler bottleneck I have experienced. I have a flow that schedules 22 parallel mapped jobs (each with 65 items) allowing for 1430 parallel jobs. When I schedule the job on an k8s cluster with 375 nodes (each with a worker that runs 4 threads), my scheduler throws messages out like the below. The CPU utilization of the scheduler also goes to 100%. These jobs ultimately die when all the workers get disconnected.

Copy code

distributed.utils_perf - WARNING - full garbage collections took 10% CPU time recently (threshold: 10%)
...
distributed.core - INFO - Event loop was unresponsive in Scheduler for 7.07s.  This is often caused by long-running GIL-holding functions or moving large chunks of data. This can cause timeouts and instability.
...
distributed.comm.tcp - INFO - Connection closed before handshake completed
...
distributed.scheduler - INFO - Close client connection: Client-worker-3c6d8642-53b5-11eb-800e-32b98c347770

When I scale the job down to 250 nodes and 3 threads per worker, I still get 100% utilization, but it is slightly more stable. where the only warning messages is

Copy code

distributed.utils_perf - WARNING - full garbage collections took 11% CPU time recently (threshold: 10%)

Thanks for any advice!

Equipe AI HOC

01/11/2021, 3:06 PM

Hello. Is there a IP whitelist for Prefect Cloud? Need to connect onprem to Cloud but I am getting blocked by firewall. Need to add Prefect to whitelist

👀 1

Brett Naul

01/11/2021, 5:08 PM

I might be missing something obvious but did the ability to filter in the UI to a given state (e.g. all Failed tasks) go away in the switch from "Gantt" to "Overview"?

Riley Hun

01/11/2021, 6:43 PM

Hello - Sorry if this is a bad question, but we want to use Dask Gateway as the executor for our Prefect Flows, but I believe the clusters generated by Dask Gateway delete themselves when in idle. So, if we have a flow that runs maybe once a day, the flow will likely fail the following day because there's no Dask Cluster to execute it (it shut down when not in use). Is there a workaround for this such that we can tell Prefect to start up a cluster from Dask Gateway each time it runs its flow, or do we have to ensure that the clusters from Dask Gateway do not delete themselves when in idle?

Billy McMonagle

01/11/2021, 7:18 PM

Hi there! Is there an environment variable that can be set on Kubernetes Agents in order to populate environment variables for flow runs managed by that agent? I see in the docs that you can run:

Copy code

prefect agent <AGENT TYPE> start --env KEY=VALUE --env KEY2=VALUE2

Instead, is it possible to do this by setting an environment variable on the agent, similar to how labels are set?

Copy code

- name: PREFECT__CLOUD__AGENT__LABELS
  value: "[]"

Peter Roelants

01/11/2021, 7:20 PM

Hi Prefect Community, I'm new to Prefect and am trying to learn to work with the tool. However, I'm struggling to understand how to properly decouple building of Storage artefacts, and registering/running these artifacts at a later moment in time. It seems that with

flow.register

the creation and registration need to happen in the same call. Is there an example somewhere on how to decouple these steps? For example how to create and store a Docker build artefact that encapsulate a flow, and running/registering the flow stored in the Docker artefact at a later time without access to the original flow file.

Joël Luijmes

01/11/2021, 7:33 PM

Hey got another question concerning retries. • What happens if I don’t specify way to persist results? Is it possible that an upstream task is rerun to get the input? Or are teh inputs stored in memory? • And if if the setup of the context manager is the upstream, yield that the same behavior? • And does anything change if I run the flow on my kubernetes agent or locally? I noticed for example that - running locally - my context manager setup task was ran again on a task failure. I’m trying to figure out why exactly, and how to mitigate 😅 And, what is the best way to persist intermediate results for kubernetes? A local volume, such that it deletes all results after completion?

Jeremy Phelps

01/11/2021, 10:16 PM

Hi all, what version of the

google.cloud

Python library does Prefect depend on?

Jeremy Phelps

01/11/2021, 11:10 PM

How do you get your Google credentials into Kubernetes? I have a Google credentials JSON file on the Agent machine, but when my flow tries to run in Kubernetes, this error happens:

Copy code

Failed to load and execute Flow's environment: DefaultCredentialsError('Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see <https://cloud.google.com/docs/authentication/getting-started>')