prefect-community #prefect-community

prefect-community

Edmund Tian

12/21/2022, 3:03 PM

I’m getting a bunch of misc exceptions from Prefect during my Flow runs. Errors in thread (most of them are

sqlite3.OperationalError

). Anyone know what’s going on? Context: I’m using the

@flow

and

@task

decorators for my Python functions in my app hosted on Google Cloud Run. I’m not using Prefect Cloud. I’m on the following prefect version:

prefect==2.3.2

. It’s hard for me to repo because this occurs during my chron job. I haven’t been able to repo locally.

Mark Nuttall-Smith

12/21/2022, 3:19 PM

I've managed to get my flows to run in docker infrastructure, using a remote s3 storage block for deploying flows, which is pretty cool (to me :P). (earlier struggles) I'm curious how people deploying/declaring/installing the prefect dependencies of their flows though 🤔 . Is it correct that the flows themselves are downloaded at runtime, but the dependencies have to be included in the infrastructure image ahead of time? Say I'm trying to orchestrate a Java application. Does the Java application image also need to have Python, prefect and the prefect dependencies pre-installed?

✅ 2

David Steiner Sand

12/21/2022, 6:12 PM

Firstly, thank you all for building and maintaining Prefect, I’ve been using it in production for a while now and it works wonderfully. Similar to the previous question, currently I’m trying to orchestrate a non Python application using Prefect. The ideia is to use the Prefect to periodically run PHP scripts inside Docker containers, however we might also use Prefect for other scripts in the future, written in other languages. I’ve managed to successfully run non-python scripts inside Docker containers through Prefect, but the agents that run these containers does not seem to update the server with the flow run status. Please see this thread to understand what I did.

✅ 1

Jessica Smith

12/21/2022, 7:15 PM

Prefect V1 question - I think my flow concurrency limit is being used up by deleted flow runs. Is there a way to get the flow ids that are running with a particular label?

Paco Ibañez

12/21/2022, 9:46 PM

Hello! I have a few flows that take dataframe as input and return a transformed dataframe. I would like to create an orchestrator flow that calls the flows using

run_deployment

. What would be the most efficient way of passing the dataframes around?

12/21/2022, 9:50 PM

allow_failure does not work… I am trying:

from prefect import task, flow, get_run_logger, allow_failure
from time import sleep


@task(retries=2, timeout_seconds=3)
def test_1(o, test_retries):
    if test_retries:
        sleep(5)
    else:
        raise IOError('Raising some error')
    return True


@flow(name='test allow_failure')
def run_test_flow():
    t = test_1.submit(True, True)
    t = test_1.submit(t, False, wait_for=allow_failure(t))


if __name__ == "__main__":
    run_test_flow()

But it does not appear to be working it does not raise the IOError

Sam Werbalowsky

12/21/2022, 10:24 PM

Is it correct that in order to run tasks concurrently we need to us

submit

? Even if the flow is written in a way that there is no dependencies between tasks?

✅ 1

Gows Shaik

12/22/2022, 4:37 AM

Hi everyone,How can i get cache storage key for file in prefect.Iam using default caching in prefect.I want to get a storage id for file

Jons Cyriac

12/22/2022, 5:03 AM

Hi! I'm trying to use Deployment() to deploy my flow to an orion server. I'm not able to set the client right. Looks like Deployment().apply() is calling get_client() which gets the api_url from profile.toml. i want to set it from my script itself. How do I go about this?

Chris Arderne

12/22/2022, 7:56 AM

Help with Prefect
v2
(more details in thread!) 1. 🏗️ Structuring project: We're using v2 with a GCS Storage block, a KubernetesJob block and a Flow Deployment (all three defined in Python files). Because we're using GCS Storage, the flow files are not

pip installed

so we can only use relative imports in the flow like

from .utils import foo

, which can make local development and bugchecking a bit more error-prone. We will probably shift to just using the Docker images local files, but haven't yet experimented with this in Prefect v2. Are there any best-practises around this? (Project structure in thread) 2. 🏃‍♀️ Running flows: Prefect

v1

had a

prefect run ...

command in the CLI, which made it possible to run a local flow. Prefect`v2` doesn't seem to have this? We can't add a

if __name__ == "__main__"

to the Flow file, because it has relative imports so running it doesn't work. As a workaround I've added a

run.py

in a parent directory that imports and runs it, but this isn't very ergonomic. In either case if I want to pass parameters, I must create my own CLI for doing so... Have I missed something in the CLI, or is there a plan to add this? Or is there a better pattern I should be using instead?

James Zhang

12/22/2022, 10:14 AM

hi everyone, just a quick question, how can i get the current flow run name programmatically using python during the run? couldn’t find the doc atm…

✅ 1

Nic

12/22/2022, 10:26 AM

Can i have two agents logged in to two different workspaces, running on the same pc?

Nic

12/22/2022, 10:38 AM

Can i access the log database hosted in orion cloud if i want to make a local dashboard based on flow runs'?

✅ 1

Mohit Singhal

12/22/2022, 12:46 PM

Hi here, how can I run same subflow multiple times parallely?

Mohit Singhal

12/22/2022, 12:48 PM

can anybody let me know? Thanks in advance

Justin Trautmann

12/22/2022, 2:05 PM

hello 😛refect:refect team, hello community, is there more information available on how to make the ray autoscaler and prefect 2 task concurrency limits work smoothly together? currently, when i submit a large number of tasks with a concurrency limit tag, the ray autoscaler will see the large amount of tasks and scale accordingly, however only a small number of tasks is allowed to run concurrently due to tag limits and most of the upscaled resources won't be used, which causes unnecessary costs especially for tasks with high resource requirements. i think this is due to the fact that the concurrency limit check happens during propose_state which happens inside the call that is submitted to the task runner. is there any way how this can be avoided? thanks a lot for any input.

Daniel Komisar

12/22/2022, 2:39 PM

Prefect v1 question: is there any way to use task looping with a function that creates tasks? I have a group of tasks that I'm using with

apply_map

and what I'd really like to do is have each of those be the body of a loop. It looks like you can only loop one single task though. Am I missing something? Thanks!

aram

12/22/2022, 3:11 PM

could someone advise how to get orion report api logs? tried

PREFECT_DEBUG_MODE

PREFECT_LOGGING_LEVEL

PREFECT_LOGGING_SERVER_LEVEL

. Non helps

Deepanshu Aggarwal

12/22/2022, 4:01 PM

hi! i am self hosting prefect (orion as well as the flows) in kubernetes cluster . i used https://github.com/PrefectHQ/prefect-helm to setup my orion server and https://github.com/anna-geller/dataflow-ops-aws-eks/blob/main/.github/workflows/eks_prefect_agent.yml for deploying agents. im using kubernetes job block as the infrastructure block and s3 block for the storage. My flows have high memory usage compared to cpu usage ( tending to 1-2 cores and 12gb memory) im running 5 flows at a time with 2000-3000 batches of 50 tasks and running 1000s of flows throughout the day . i have implemented task concurrency limits and flow concurrency limits on every work queue. Im consistently experiencing flow run crashes . Besides concurrency limits , what options do i have to control the load and/or are there any methods to control the resource usage of the jobs in my cluster?

:upvote: 1

✅ 1

Kristian Andersen Hole

12/22/2022, 5:19 PM

Hi! I’m also self hosting Orion in a K8s cluster, and running most of our flows as K8sJobs with success. However, we have some flows that need to be run on Fargate as ECSTasks due to some circumstances. The agents spins up the ECSTask just fine, but on startup of prefect inside the ecstask prefect tries to contact prefect-orion, which inevitably fails because it attempts a kube-dns hostname. (the orion instance also isn’t exposed to the outside) I was under the assumption that it only needed a channel to the agent, but thinking about it does make sense that it needs Orion as well. Is this correctly assessed? Any workarounds to this? We would at the very least need to make it target a proper host instead of the kube-dns name. Then perhaps the two can safely be allowed to connect. Self-hosted Orion doesnt have auth, so exposing it to 0.0.0.0/0 might not be an option.

✅ 1

Laraib Siddiqui

12/22/2022, 5:45 PM

When using

from prefect.tasks.shell import ShellTask

Laraib Siddiqui

12/22/2022, 5:47 PM

When using

from prefect.tasks.shell import ShellTask

it throws out and error

ModuleNotFoundError: No module named 'prefect.tasks.shell'; 'prefect.tasks' is not a package

I know i can use

prefect-shell

to overcome this and run python scripts but i couldn't find any resolution on the above error

Uday

12/22/2022, 7:25 PM

Hi Guys! New to Prefect, I have made good progress integrating Prefect with Snowflake and able to write queries and manage workflows. But, I was wondering if there is prefect integration with Snowpark?

✅ 1

Kristian Andersen Hole

12/22/2022, 8:03 PM

Hey, another query for you guys 🙂 Is it possible to serve Orion on a path. Putting it on a hostname + path without doing anything else makes index.html reference assets at root, and not at the path at which the actually now are reachable! Any input appreciated !

✅ 1

Javier Ochoa

12/22/2022, 8:22 PM

Hello Folks. Question. I have a flow defined with several task using "map" I want to run a final task once all map tasks have finished. This is my code:

with Flow(
    FLOW_NAME,
    run_config=UniversalRun(labels=LABELS),
    terminal_state_handler=workflow_terminal_state_handler,
) as flow:

    file_list = list_unprocessed_files(var)

    dataframes = get_dataframes.map(file_list)

    dataframes = filter_dataframe.map(
        dataframes, resource_type_name=unmapped(var)
    )

    # this is the final task
    print_process_summary_log()

Billy McMonagle

12/23/2022, 3:40 AM

Hi there, I'm working on migration a flow from Prefect 1 to 2. I use Snowflake, and am having some trouble with the `prefect-dbt`/`prefect-snowflake` collections. I think there is an issue related to my use of private key Snowflake authentication. Here are some details... by the way, I know many of us have slowed down for the holidays and I'm not in a rush for a response.

👀 2

Amruth VVKP

12/23/2022, 12:52 PM

Anyone running into issues launching Orion from the latest prefect 2.7.4 docker image? My container is crashing on with these logs -

2022-12-23 12:47:33 
2022-12-23 12:47:33  ___ ___ ___ ___ ___ ___ _____    ___  ___ ___ ___  _  _
2022-12-23 12:47:33 | _ \ _ \ __| __| __/ __|_   _|  / _ \| _ \_ _/ _ \| \| |
2022-12-23 12:47:33 |  _/   / _|| _|| _| (__  | |   | (_) |   /| | (_) | .` |
2022-12-23 12:47:33 |_| |_|_\___|_| |___\___| |_|    \___/|_|_\___\___/|_|\_|
2022-12-23 12:47:33 
2022-12-23 12:47:33 Configure Prefect to communicate with the server with:
2022-12-23 12:47:33 
2022-12-23 12:47:33     prefect config set PREFECT_API_URL=<http://0.0.0.0:4200/api>
2022-12-23 12:47:33 
2022-12-23 12:47:33 View the API reference documentation at <http://0.0.0.0:4200/docs>
2022-12-23 12:47:33 
2022-12-23 12:47:33 Check out the dashboard at <http://0.0.0.0:4200>
2022-12-23 12:47:33 
2022-12-23 12:47:33 
2022-12-23 12:47:33 
2022-12-23 12:47:35 Traceback (most recent call last):
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 233, in _catch_revision_errors
2022-12-23 12:47:35     yield
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 443, in _upgrade_revs
2022-12-23 12:47:35     for script in reversed(list(revs))
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/revision.py", line 799, in iterate_revisions
2022-12-23 12:47:35     revisions, heads = fn(
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/revision.py", line 1454, in _collect_upgrade_revisions
2022-12-23 12:47:35     current_revisions = self.get_revisions(lower)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/revision.py", line 527, in get_revisions
2022-12-23 12:47:35     return sum([self.get_revisions(id_elem) for id_elem in id_], ())
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/revision.py", line 527, in <listcomp>
2022-12-23 12:47:35     return sum([self.get_revisions(id_elem) for id_elem in id_], ())
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/revision.py", line 552, in get_revisions
2022-12-23 12:47:35     return tuple(
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/revision.py", line 553, in <genexpr>
2022-12-23 12:47:35     self._revision_for_ident(rev_id, branch_label)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/revision.py", line 624, in _revision_for_ident
2022-12-23 12:47:35     raise ResolutionError(
2022-12-23 12:47:35 alembic.script.revision.ResolutionError: No such revision or branch 'f7587d6c5776'
2022-12-23 12:47:35 
2022-12-23 12:47:35 The above exception was the direct cause of the following exception:
2022-12-23 12:47:35 
2022-12-23 12:47:35 Traceback (most recent call last):
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/starlette/routing.py", line 671, in lifespan
2022-12-23 12:47:35     async with self.lifespan_context(app):
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/starlette/routing.py", line 566, in __aenter__
2022-12-23 12:47:35     await self._router.startup()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/starlette/routing.py", line 648, in startup
2022-12-23 12:47:35     await handler()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/api/server.py", line 345, in run_migrations
2022-12-23 12:47:35     await db.create_db()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/database/interface.py", line 55, in create_db
2022-12-23 12:47:35     await self.run_migrations_upgrade()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/database/interface.py", line 63, in run_migrations_upgrade
2022-12-23 12:47:35     await run_sync_in_worker_thread(alembic_upgrade)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 69, in run_sync_in_worker_thread
2022-12-23 12:47:35     return await anyio.to_thread.run_sync(call, cancellable=True)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
2022-12-23 12:47:35     return await get_asynclib().run_sync_in_worker_thread(
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
2022-12-23 12:47:35     return await future
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
2022-12-23 12:47:35     result = context.run(func, *args)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/database/alembic_commands.py", line 24, in wrapper
2022-12-23 12:47:35     return fn(*args, **kwargs)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/database/alembic_commands.py", line 53, in alembic_upgrade
2022-12-23 12:47:35     alembic.command.upgrade(alembic_config(), revision, sql=dry_run)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/command.py", line 322, in upgrade
2022-12-23 12:47:35     script.run_env()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 569, in run_env
2022-12-23 12:47:35     util.load_python_file(self.dir, "env.py")
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/util/pyfiles.py", line 94, in load_python_file
2022-12-23 12:47:35     module = load_module_py(module_id, path)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/util/pyfiles.py", line 110, in load_module_py
2022-12-23 12:47:35     spec.loader.exec_module(module)  # type: ignore
2022-12-23 12:47:35   File "<frozen importlib._bootstrap_external>", line 883, in exec_module
2022-12-23 12:47:35   File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/database/migrations/env.py", line 147, in <module>
2022-12-23 12:47:35     apply_migrations()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 201, in coroutine_wrapper
2022-12-23 12:47:35     return run_async_from_worker_thread(async_fn, *args, **kwargs)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 152, in run_async_from_worker_thread
2022-12-23 12:47:35     return anyio.from_thread.run(call)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/anyio/from_thread.py", line 49, in run
2022-12-23 12:47:35     return asynclib.run_async_from_thread(func, *args)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
2022-12-23 12:47:35     return f.result()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 458, in result
2022-12-23 12:47:35     return self.__get_result()
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
2022-12-23 12:47:35     raise self._exception
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/database/migrations/env.py", line 141, in apply_migrations
2022-12-23 12:47:35     await connection.run_sync(do_run_migrations)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/sqlalchemy/ext/asyncio/engine.py", line 548, in run_sync
2022-12-23 12:47:35     return await greenlet_spawn(fn, conn, *arg, **kw)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 128, in greenlet_spawn
2022-12-23 12:47:35     result = context.switch(value)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/prefect/orion/database/migrations/env.py", line 129, in do_run_migrations
2022-12-23 12:47:35     context.run_migrations()
2022-12-23 12:47:35   File "<string>", line 8, in run_migrations
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/runtime/environment.py", line 853, in run_migrations
2022-12-23 12:47:35     self.get_context().run_migrations(**kw)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/runtime/migration.py", line 611, in run_migrations
2022-12-23 12:47:35     for step in self._migrations_fn(heads, self):
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/command.py", line 311, in upgrade
2022-12-23 12:47:35     return script._upgrade_revs(revision, rev)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 431, in _upgrade_revs
2022-12-23 12:47:35     with self._catch_revision_errors(
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/contextlib.py", line 153, in __exit__
2022-12-23 12:47:35     self.gen.throw(typ, value, traceback)
2022-12-23 12:47:35   File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 265, in _catch_revision_errors
2022-12-23 12:47:35     raise util.CommandError(resolution) from re
2022-12-23 12:47:35 alembic.util.exc.CommandError: Can't locate revision identified by 'f7587d6c5776'
2022-12-23 12:47:35 
2022-12-23 12:47:35 Application startup failed. Exiting.
2022-12-23 12:47:36 Orion stopped!

Logs while trying to launch Prefect Orion on local Ubuntu terminal with 2.7.4 -

___ ___ ___ ___ ___ ___ _____    ___  ___ ___ ___  _  _
| _ \ _ \ __| __| __/ __|_   _|  / _ \| _ \_ _/ _ \| \| |
|  _/   / _|| _|| _| (__  | |   | (_) |   /| | (_) | .` |
|_| |_|_\___|_| |___\___| |_|    \___/|_|_\___\___/|_|\_|

Configure Prefect to communicate with the server with:

    prefect config set PREFECT_API_URL=<http://127.0.0.1:4200/api>

View the API reference documentation at <http://127.0.0.1:4200/docs>

Check out the dashboard at <http://127.0.0.1:4200>



/usr/lib/python3.10/contextlib.py:142: SAWarning: Skipped unsupported reflection of expression-based index ix_flow_run__coalesce_start_time_expected_start_time_desc
  next(self.gen)
/usr/lib/python3.10/contextlib.py:142: SAWarning: Skipped unsupported reflection of expression-based index ix_flow_run__coalesce_start_time_expected_start_time_asc
  next(self.gen)

👍 1

✅ 1

Chris Gunderson

12/23/2022, 3:29 PM

Hi Prefect Team - We now have a working dev environment with Prefect 2. The ECS Service is running the agent properly. However, for each flow we are creating a new task definition. Will we need to create a separate ECS Service to run the flows? If we create a separate ECS Service and Task Definition for the flows, do we put this task definition in the Prefect block for the ECS Task?

Robert Esteves

12/23/2022, 6:55 PM

How do I import a function from another file in a Prefect 1.0 script? My Prefect script is stored in a folder in my repo. I also stored the function script within the same folder, but whenever I try to execute the Flow from Prefect IO it fails (No Module Found Error) here is my example: github |_my_repo|_

Pekka

12/24/2022, 2:06 PM

Merry Christmas 🥳 🎄

🎄 9