prefect-community #prefect-community

Hi all -- I have a pipeline flow that is working perfectly fine on one cluster, but when running it on another I am seeing these errors thrown consistently. There are no errors related to the actual work, and I am executing this workflow against a known dataset that is weel understood, so I am sure the problem is not there. Any ideas? It is almost like a task ID can not be pulled from the database or something. I am running a postgres data and orion server in a VM near said clusters.

Copy code

Encountered exception during execution:
Traceback (most recent call last):
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/engine.py", line 610, in orchestrate_flow_run
    result = await run_sync(flow_call)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 68, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/bin/process_holography.py", line 431, in holography_flow
    return_me = log_and_wait_futures(return_me)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/tasks.py", line 360, in __call__
    return enter_task_run_engine(
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/engine.py", line 733, in enter_task_run_engine
    return run_async_from_worker_thread(begin_run)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 148, in run_async_from_worker_thread
    return anyio.from_thread.run(call)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/anyio/from_thread.py", line 49, in run
    return asynclib.run_async_from_thread(func, *args)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
    return f.result()
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/concurrent/futures/_base.py", line 446, in result
    return self.__get_result()
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/engine.py", line 874, in get_task_call_return_value
    return await future._result()
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/futures.py", line 237, in _result
    return await final_state.result(raise_on_failure=raise_on_failure, fetch=True)
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/states.py", line 86, in _get_state_result
    raise MissingResult(
prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

✅ 1

Thomas Fredriksen

11/16/2022, 12:21 PM

Hi there! I recall that flows would be stored as pickled objects with a

.flow

-postfix in Prefect 1. Is it still possible to store flows in this way in Prefect 2? From what I tell from the source, it may not be possible https://github.com/PrefectHQ/prefect/blob/a24c725629e1a6aef31bcb577d4a853a86fc1cbe/src/prefect/utilities/importtools.py#L191

✅ 1

Joshua Grant

11/16/2022, 3:22 PM

Has anyone had any luck creating a

DockerRegistry

block with AWS ECR?

✅ 1

Jenia Varavva

11/16/2022, 4:21 PM

Hi, I’m trying to create a flow (Prefect v1) that calls two other flows on different API URLs, e.g.

team1_url

and

team2_api

. I wanted to use the create_flow_run() task, but since it constructs a

Client()

with no args, I’m having trouble pointing it at the right API_URL. My initial instinct was to do something like:

Copy code

config_team1 = prefect.context.config.copy()
config_team1.cloud.api = 'team1_url'
with Flow('parent'):
  with prefect.context(config=config_team1):
    create_flow_run(...)  
  with prefect.context(config=config_team2):
    create_flow_run(...)

This, unfortunately, doesn’t work as the context is set at the flow definition time, not at task execution (i.e. not when

Client()

is executed). An alternative that got me closer was something like:

Copy code

with prefect.context(config=config_team1):
  flow.run()

which indeed overrides the config value for the

Client()

call inside the

create_flow_run()

task, but doesn’t look like it would allow me using two different urls in the same flow. Is there a better way around this than copying/adjusting

create_flow_run()

to allow overriding

api_url

itself?

👀 1

Dev Dabke

11/16/2022, 4:28 PM

I have a very particular question: I have a central

logger

created as a singleton in a python module. However, sometimes, this module is invoked in a prefect flow. The logs from this logger are not displaying during the flow run, nor or they logging to orion. I am not invoking

get_run_logger

anywhere. The issue is that the module is only sometimes invoked in the context of a flow; it’s a shared library. Can I register prefect as a handle to this logger that will only log when in a prefect context?

Mac

11/16/2022, 4:29 PM

Hi, is there an equivalent of

create_flow_run.map()

from prefect v1 in v2? I am trying to run a few subflows at once, then have the flow wait to get successful responses from all of the subflow before moving on the the next task

Jai P

11/16/2022, 5:27 PM

👋 hullo, prefect 2 question! my team is trying to figure out in what cases you might want to call a task directly, submit a task, or submit a task and wait for the result, and if there's a general recommendation on those types of things? more clarification in thread

✅ 1

Thomas Fredriksen

11/16/2022, 7:29 PM

Hi there, I am trying to use the

DaskTaskRunner

with the

KubeCluster

-type from

dask-kubernetes

, however I am encountering the following error:

Copy code

Crash details:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/task_runners.py", line 161, in start
    await self._start(exit_stack)
  File "/usr/local/lib/python3.10/site-packages/prefect_dask/task_runners.py", line 300, in _start
    self._client = await exit_stack.enter_async_context(
  File "/usr/local/lib/python3.10/contextlib.py", line 619, in enter_async_context
    result = await _cm_type.__aenter__(cm)
  File "/usr/local/lib/python3.10/site-packages/distributed/client.py", line 1398, in __aenter__
    await self
  File "/usr/local/lib/python3.10/site-packages/distributed/client.py", line 1213, in _start
    await self._ensure_connected(timeout=timeout)
  File "/usr/local/lib/python3.10/site-packages/distributed/client.py", line 1276, in _ensure_connected
    comm = await connect(
  File "/usr/local/lib/python3.10/site-packages/distributed/comm/core.py", line 291, in connect
    comm = await asyncio.wait_for(
  File "/usr/local/lib/python3.10/asyncio/tasks.py", line 445, in wait_for
    return fut.result()
  File "/usr/local/lib/python3.10/site-packages/distributed/comm/tcp.py", line 487, in connect
    ip, port = parse_host_port(address)
  File "/usr/local/lib/python3.10/site-packages/distributed/comm/addressing.py", line 95, in parse_host_port
    port = _default()
  File "/usr/local/lib/python3.10/site-packages/distributed/comm/addressing.py", line 73, in _default
    raise ValueError(f"missing port number in address {address!r}")
ValueError: missing port number in address '<Not Connected>'

The taskrunner is set dynamically:

Copy code

flow.task_runner = DaskTaskRunner(
            cluster_class=KubeCluster,
            cluster_kwargs={
                "name": f"{flow.name}",
                "namespace": "prefect",
                "image": self._storage.get_name(),
                "n_workers": 1,
                "resources": {
                    "limits": {
                        "cpu": resources.cpu,
                        "memory": resources.mem,
                    }
                },
            },
            adapt_kwargs={
                "minimum": self._min_workers,
                "maximum": self._max_workers,
            },
            client_kwargs={
                "set_as_default": True
            }
        )

Does anyone who might be causing this?

Vadym Dytyniak

11/16/2022, 7:32 PM

Hi. What is the replacement for

LocalRun

in Prefect 2?

✅ 1

badasstronaut

11/16/2022, 7:33 PM

Greetings again Prefect Community. I am trying to pass AWS tags to tasks run by Prefect. I am using

prefect_aws

and here’s an example of my ecs.py infra block:

Copy code

ecs = ECSTask(image='custom-prefect-dask:latest',
              cpu='512',
              memory='2048',
              cluster='arn:aws:ecs:us-east-1:...Cluster',
              execution_role_arn='arn:aws:iam::...Role',
              task_role_arn='arn:aws:iam:...Role',
              task_start_timeout_seconds=300,
              configure_cloudwatch_logs=True,
              launch_type='FARGATE_SPOT',
              vpc_id='vpc-...',
              task_customizations=[dict(op='replace',
                                        path='/tags',
                                        value=[dict(key='TAG_1', value='Prefect'),
                                               dict(key='TAG_2', value='POC')]),
                                   dict(op='replace',
                                        path='/networkConfiguration/awvpcConfiguration/assignPublicIp',
                                        value='DISABLED')],
              env={'PREFECT_API_URL': '<http://prefect.internal>..../api'})

Expected result: The

prefect

task reflects the tags provided in

task_customizations

Actual result: Only the default

<http://prefect.io/|prefect.io/>

tags are generated on the

prefect

task EDIT: This turned out to be due to an outdated version of

prefect-aws

. The code block above works as expected.

✅ 1

Vishnu Duggirala

11/16/2022, 7:43 PM

This is a question for prefect-aws.

Copy code

The image to use for the Prefect container in the task. If this value is not null, it will override the value in the task definition. This value defaults to a Prefect base image matching your local versions.

it is not overriding the image when I gave the task definition ARN. Does it only override if I give the task definition?

✅ 1

Madison Schott

11/16/2022, 7:54 PM

I'm curious, has anyone used Rivery.io and Prefect together? I don't see any integrations specifically built out, probably because they also offer orchestration- but wondering how complicated it would be to sync data with Rivery from a Prefect flow. Thanks!

✅ 1

Ashley Felber

11/16/2022, 8:00 PM

Hello, I just updated a Python script to work as a Prefect flow. I executed it locally and it works but it took ~25 minutes to run. However, if I execute the python script locally, without Prefect, it takes ~3 minutes. Why would this be the case?

dados grupoa

11/16/2022, 8:07 PM

Hello guys, I'm trying to migrate from prefect 1 to 2, but I can't find how to put dynamic names on tasks, it would be the equivalent of this, in prefect 1 https://docs-v1.prefect.io/core/concepts/templating.html. Can anybody help me ?

✅ 1

Patrick Alves

11/16/2022, 8:15 PM

Hi there, I have a ETL flow running on Prefect 2.6.4 that runs in a Kubernetes Cluster. I am having issues with the KubernetJob retries. I’ve create a Job template where I set the backoffLimit as 0, so it should not retry any failed flow/Job. (I will add the full template as a comment. ) The point is that Prefect is not setting the backoffLimit in Job manifest when it creates the KubernetesJob. See the config of the Job created by Prefect (see the 2nd image). I really need to avoid these retries, any tips?

Sander

11/16/2022, 9:32 PM

Hi, I’m working on setting up an ephemeral dask cluster in kubernetes managed by a KubeCluster operator. All orchestrated by prefect. So far so good. But I’m running into some weird error that I don’t find obvious to solve. See thread.

Andrew P

11/16/2022, 10:07 PM

Hey folks, I'm getting a bunch of

| ERROR | prefect.agent - Failed to create work queue 'default'.

when trying to connect to my Prefect Cloud work queue from ECS. I've verified that the

PREFECT_API_URL

looks right (basically,

"<https://api.prefect.cloud/api/accounts/${prefect_account_id}/workspaces/${prefect_workspace_id}>"

, and the two variables when substituted in the env var are the same as in my prefect cloud UI url). Is there any other failure mode that might result in this? Can I pass a debug flag to see if the failure is eg a 403 vs a lack of dns or something? No google results and the error seems pretty opaque. Starting with the command

["prefect","agent","start","-q","default"]

using the image

prefecthq/prefect:2-python3.10

✅ 1

Scott Walsh

11/17/2022, 12:15 AM

Prefect 2.4.5 I have a parent_flow that is registered to a deployment in Orion. I am using a kubernetes block to run this parent_flow as a k8s job. Within that parent_flow I have two sub flows, sub_flow1 and sub_flow2. Both of these sub flows are also registered to their own deployments that run as k8s jobs. I am triggering the subflows using client.create_flow_run_from_deployment(**). Is there a way that I can pass the results from sub_flow1 into sub_flow2 here? And, can I somehow wait to start sub_flow2 until sub_flow1 is complete? The reason I have the sub flows built as their own deployments is so they can run using different resources, maybe one needs a GPU and the other does not. Here is some code for what I am describing:

Copy code

import asyncio
from prefect.client import get_client
from prefect import flow, task

@task
async def child_task1():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(name='sub_flow1/sub_flow1')
        response = await client.create_flow_run_from_deployment(deployment.id)

@task
async def child_task2():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(name='sub_flow2/sub_flow2')
        response = await client.create_flow_run_from_deployment(deployment.id)

@flow
async def parent_flow():
    await child_task1.submit()
    await child_task2.submit()

if __name__ == "__main__":
    asyncio.run(parent_flow())

✅ 1

Carlo

11/17/2022, 12:23 AM

Hello. We were originally using

ECSTask

with the

task_definition

arg. Today we tried transitioning to

task_definition_arn

However, it's still trying to register a task definition every time the agent picks up a flow to run. Details in thread...

✅ 1

Ben Muller

11/17/2022, 12:57 AM

Hey if I have a flow like so:

Copy code

@flow
def my_flow(
    start: str = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d"),
    end: str = datetime.utcnow().strftime("%Y-%m-%d"),
):

Will these default parameters be set at deploy time or will they update at run time ?

✅ 1

Nick Batchelder

11/17/2022, 1:03 AM

Hey all, is it possible to add custom k8s labels to a flow run at the time a flow run is created? I see these three labels are set by default, but it would be super handy if I could set custom labels like

username=nick

at runtime.

Deepanshu Aggarwal

11/17/2022, 8:15 AM

hi! im running into this error

Copy code

httpx.RemoteProtocolError: <ConnectionTerminated error_code:ErrorCodes.NO_ERROR, last_stream_id:19999, additional_data:None>

anyone who faced similar issue ? adding the trace stack in comments

Ben Muller

11/17/2022, 8:31 AM

Hey - what does it mean when we get a message saying:

Copy code

lock document has schema checksum sha256:96937dc8f6bd314a1cbeae7623159848872249e742b3bac8d7af39da4e2dfe79 which does not match the schema checksum for class 'ECSTask'. This indicates the schema has changed and this block may not load.

How do i debug and or make sure everything is all good?

➕ 1

✅ 1

Mohit Singhal

11/17/2022, 8:41 AM

Hi team, I am new to prefect and I have a question. If I am running a flow daily and it failed for yesterday and now when I will be retrying it today, is there any date parameter which gives me yesterday's date instead of today's date as I will be fetching some values based on date from the api in my flow.

✅ 1

Manuel Garrido Peña

11/17/2022, 9:32 AM

hey folks, maybe this is a stupid question, but Why isnt there a way to trigger a dag from the CLI with custom parameters in prefect 2.0? in 1.0, you could do

params= X=y

, why did we remove such awesome feature?

✅ 1

Zinovev Daniil

11/17/2022, 11:26 AM

Copy code

Hi all. I had a problem right after installing prefect orion. UI is not loading. I am getting a blank page with a title. JS is not fully loaded. It breaks at 1.1 mb. Perhaps someone solved such a problem?

✅ 1

Rajeshwar Agrawal

11/17/2022, 1:27 PM

Hey @Prefect, One of the persistent issues that we were facing with prefect V1 was the unreliability around heartbeats and lazarus issues. This was promised to be fixed in v2, for which I recently did a quick evaluation of release 2.6.7 (community). Unfortunately, my initial investigation doesn’t seem to give great results. For example, any changes to k8s pod running a KubernetesJob type flow (such as pod restarts, pod terminated due k8s operations etc) do not show up as appropriate status updates on prefect flow. For instance, if a pod is terminated, the flow status remains Running indefinitely (flow timeouts wont solve the issue here, as our flow execution times follows the bell curve). I believe a combination of hearbeat and lazarus in v1 worked around this by marking flows which didn’t report back for a long time as crashed, which were retried by Lazarus. A similar functionality seems to be completely missing in v2. A github issue for this also exists https://github.com/PrefectHQ/prefect/issues/7371#issuecomment-1295604573 Is this something that’s already on the roadmap, which can expect to be released in near future? Are there any other flow execution patterns, that would work with k8s compute infrastructure, which solves the above mentioned problem? Here are some other observations that I found in v2, which are different from v1 1. It’s not possible to Cancel a flow run. We can delete it, but that does not terminate the underlying k8s job associated with the run. It seems that delete will only stop tracking the status of flow run. 2. The flows are no longer organised in projects. Previously, were able to do a separate deployment of a single flow into multiple projects, which allowed us to mimic multi-tenant capabilities, and manage different versions of same flow for different tenants. A similar functionality exists in v2, which allows multiple deployments for each flow, however this is less intuitive from multi-tenant perspective, as its more convenient to view and manage all flows of single project at once place, rather than manage different deployments of different tenants within the same flow 3. Date based filtering of logs of flow run is not supported. We need to keep scrolling down to reach the last line of log, this is not ideal for long-running flows with large log output 4. UI Issue - While working with Kubernetes Job Block, I observed that its not possible to unset any optional fields. 5. The time elapsed shown on UI remains at 0 secs unless flow reaches a terminal state

🎯 3

💅 1

Jared Robbins

11/17/2022, 2:06 PM

I’m looking at the code and don’t see any reason my mattermost webhook wont work for notications. I’m able to use the webhook from the container that prefect is running in, but it just doesn’t send anything for flow state changes. Where could i investigate to see if there is an error log somewhere? I’m labeling it as a slack webhook Edit: i tested apprise and found the issue is there.

✅ 1

Jon

11/17/2022, 2:19 PM

hi, I'm on prefect 1 using this pattern to pass a task's result from a child flow to a parent flow: https://discourse.prefect.io/t/how-to-pass-data-from-a-child-flow-to-another-task-in-a-parent-flow/126 the child flow is mapped with

create_flow_run.map

. How to get the results of the mapped flow? I'm getting this error:

Copy code

└── 09:11:14 | ERROR   | Task 'get_task_run_result': Exception encountered during task execution!
Traceback (most recent call last):
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 880, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/tasks/prefect/flow_run.py", line 220, in get_task_run_result
    flow_run = FlowRunView.from_flow_run_id(flow_run_id)
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/backend/flow_run.py", line 585, in from_flow_run_id
    flow_run_data = cls._query_for_flow_run(where={"id": {"_eq": flow_run_id}})
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/backend/flow_run.py", line 627, in _query_for_flow_run
    result = client.graphql(flow_run_query)
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/client/client.py", line 464, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'message': 'parsing UUID failed, expected String, but encountered Array', 'locations': [{'line': 2, 'column': 5}], 'path': ['flow_run'], 'extensions': {'path': '$.selectionSet.flow_run.args.where.id._eq', 'code': 'parse-failed', 'exception': {'message': 'parsing UUID failed, expected String, but encountered Array'}}}]

✅ 1

Taylor Curran

11/17/2022, 2:47 PM

@Kehinde Adewusi Has the question: When using Prefect 1 databricks integration, is it possible return the logs from the databricks jobs so that they show up in the Prefect logs in the Prefect UI? Can flow run logs be sent from Prefect to elastic search?

1️⃣ 1

👍 2