prefect-community #prefect-community

prefect-community

Ofir

08/26/2022, 3:57 PM

So:

with Flow("Data preprocess" as data_preprocess_flow:
  ...

with Flow("Visualization" as visualization_flow:
  ...

vs:

with Flow("Monolithic data pipeline") as monolithic_flow:
   res = data_preprocess(dataframe)
   res = train(res)
   ...

✅ 1

Ofir

08/26/2022, 3:58 PM

The question boils down to two key questions actually: 1. Is interflow dependencies supported? 2. If it is supported, what are the tradeoffs between the two approaches? what is the standard practice?

✅ 1

Nathaniel Russell

08/26/2022, 4:35 PM

storage = S3(
    bucket=s3_bucket_name,
    key=f"flows/banner_canvas_flow.py",
    stored_as_script=True,
    local_script_path=f"flows/banner_canvas_flow.py"
)

In prefect 1.0 I could store my task code in an S3 bucket and use that implementation of the tasks whenever I ran a flow with storage=storage. How would I accomplish this in prefect 2.0?

✅ 1

Philip MacMenamin

08/26/2022, 5:05 PM

Hi, I'm wondering if there's a way to attach a key to a

context

value on a per run basis?

👀 1

✅ 1

Matt Delacour

08/26/2022, 5:15 PM

How to use the

_or

statement in GraphQL? Cannot make it work on my side. There is nothing about

_or

_and

in the doc nor in Github

👀 2

✅ 1

Neil Natarajan

08/26/2022, 7:06 PM

I keep encountering the following Runtime error:

RuntimeError: Orion requires sqlite >= 3.24.0 but we found version 3.7.17

how do I go about fixing this error

✋ 1

✅ 1

Leon Kozlowski

08/26/2022, 7:46 PM

I am trying to migration from a 0.15.12 setup using

DockerStorage

and

KubernetesRun

- I have been looking through previous threads and it seems I should be using

DockerPackager

, but it looks like a

Deployment

doesn’t accept the

packager

argument anymore - is there guidance on migrating from a DockerStorage/Kubernetes run workflow to 2.0?

Kevin Grismore

08/26/2022, 7:56 PM

Anyone know why I get

KeyError: "No class found for dispatch key 'gcs' in registry for type 'Block'."

when trying to run a flow stored on GCS using a GCS storage block?

✅ 1

Ilya Galperin

08/26/2022, 8:30 PM

We’re experiencing really strange behavior when creating a storage block and deployment using the Python Deployment object. When we create the deployment the first time, the flow is picked up and runs on our Kubernetes-hosted agent, but subsequent flow runs will seemingly randomly go into a failed state almost instantly with no error message and never get picked up by the agent. I notice when our flow does successfully run, we get this warning message in the container’s log:

/usr/local/lib/python3.10/site-packages/prefect/deployments.py:48: UserWarning: Block document has schema checksum sha256:0ec43f8010cee4adbf73aebcc58f1e45986d765c2a224dfc9cd5428f98c516f8 which does not match the schema checksum for class 'S3'. This indicates the schema has changed and this block may not load.
  storage_block = Block._from_block_document(storage_document)

Deleting and re-creating the block and deployment will sometimes cause it to work again but again, only on an intermittent basis. The flow code itself does not touch or interact with Prefect block storage. Has anyone experienced this or have any idea what might be causing the issue?

✅ 1

Jai P

08/26/2022, 8:37 PM

👋 I see mention of task checkpointing (storing the output of a task) in prefect 1 docs, but no mention of it in prefect 2 yet. I'd assume that you all are planning to eventually add that, is that correct? If so, is there a rough timeline for it?

✅ 1

Seth Coussens

08/26/2022, 8:55 PM

Anybody have anyluck with a storage block that you can use for multiple flows that doesn't just write over your files when you build the deployment?

✅ 1

Brandon T. Kowalski

08/26/2022, 9:19 PM

I am currently attempting to get a Hello World running on an Agent deployed on ECS started from the 2.0 Cloud UI. I have added a storage block in S3 and see that the CLI uploaded code there. When I go to run the Flow via the Deployment the Agent on ECS receives the job but immediately fails. CloudWatch reports the following and I can figure out how to proceed.

20:53:32.166 | INFO    | prefect.agent - Submitting flow run 'adf8ae6e-675d-40dd-8a32-078aecbc248b'
20:53:32.359 | ERROR   | prefect.agent - Failed to submit flow run 'adf8ae6e-675d-40dd-8a32-078aecbc248b' to infrastructure.

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/agent.py", line 200, in submit_run
    await self.task_group.start(submit_flow_run, flow_run, infrastructure)
  File "/usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 807, in start
    return await future
  File "/usr/local/lib/python3.9/site-packages/prefect/infrastructure/submission.py", line 47, in submit_flow_run
    infrastructure = _prepare_infrastructure(flow_run, infrastructure)
  File "/usr/local/lib/python3.9/site-packages/prefect/infrastructure/submission.py", line 35, in _prepare_infrastructure
    "env": {**base_flow_run_environment(flow_run), **infrastructure.env},
AttributeError: 'S3' object has no attribute 'env'

Any pointers?

✅ 1

Chris Gunderson

08/26/2022, 9:36 PM

Hi Team - I'd like to add this function in a utility script

from prefect.tasks.notifications import SlackTask
class Alert:

    def alert_on_special_failure(task, old_state, new_state):
        if new_state.is_failed():
            if getattr(new_state.result, "flag", False) is True:
                errMsg = '--- LOADER ERROR ---'
                msg = "{}\nTask: `{}` FAILED.\nThe loader process failed: `{}`".format(
                        errMsg, task.name, new_state.result.value)
                SlackTask().run(message = msg, webhook_secret = "SLACK_PREFECT_DEV")
        return new_state

This was previous called like this:

@task(name = 'send API request',
      max_retries = 3,
      retry_delay = timedelta(minutes = 5),
      state_handlers = [alert_on_special_failure])
def post_request_process(

If we add this alert on special failure object, will we need to pass in the task, old_state, and new_state? Were these variables global?

👀 1

✅ 1

Hedgar

08/27/2022, 7:41 AM

My current challenge: my system’s Sqlite3 version is the latest, 3.39, but the Sqlite3 version in the python terminal is 3.19. , Perfect 2 looks for Sqlite3 from Python! How can I upgrade Python’s sqlite3 version?

Ofir

08/27/2022, 11:14 AM

Seems like the v1 docs are outdated:

git clone --depth 1 <https://github.com/PrefectHQ/prefect.git>
cd prefect/examples/tutorial

👍 1

✅ 2

Ofir

08/27/2022, 11:14 AM

https://docs-v1.prefect.io/core/tutorial/01-etl-before-prefect.html

✅ 1

Ofir

08/27/2022, 11:15 AM

ofir@Ofirs-MacBook-Pro-2 prefect % ls examples/
ls: examples/: No such file or directory
ofir@Ofirs-MacBook-Pro-2 prefect %

✅ 1

Ofir

08/27/2022, 11:28 AM

ofir@Ofirs-MacBook-Pro-2 prefect % git checkout 1.x
Branch '1.x' set up to track remote branch '1.x' from 'origin'.
Switched to a new branch '1.x'
ofir@Ofirs-MacBook-Pro-2 prefect % ls examples
conditional.py  kafka           mapping.py      old             parameters.py   tutorial
ofir@Ofirs-MacBook-Pro-2 prefect % ls examples/tutorial
01_etl.py                       03_parameterized_etl_flow.py    05_schedules.py                 aircraftlib
02_etl_flow.py                  04_handle_failures.py           06_parallel_execution.py        requirements.txt
ofir@Ofirs-MacBook-Pro-2 prefect %

✅ 1

Ofir

08/27/2022, 11:29 AM

I guess the documentation should update to:

git clone --branch 1.x --depth 1 <https://github.com/PrefectHQ/prefect.git>
cd prefect/examples/tutorial

✅ 1

Keith Hickey

08/27/2022, 2:15 PM

FYSA, I posted over on #prefect-ui, looking for some insight on the Radar view and diagramming control flow with it. If you have any tips or share similar questions, please add a threaded-reply on that message. https://prefect-community.slack.com/archives/C0192RWGJQH/p1661609573510969

✅ 1

Venkat Ramakrishnan

08/28/2022, 4:04 AM

Hi, I am getting the following error when I run Prefect agent looking for flows deployed from a python script running on a Windows laptop. Further details in the thread. 09:28:36.300 | INFO | prefect.agent - Submitting flow run 'ae7d9532-0a2b-42a6-bbfb-18f3f987098a' 09:28:41.921 | INFO | prefect.agent - Submitting flow run 'ae7d9532-0a2b-42a6-bbfb-18f3f987098a' 09:28:46.342 | INFO | prefect.infrastructure.process - Opening process 'bouncy-dalmatian'... 09:28:46.342 | INFO | prefect.agent - Completed submission of flow run 'ae7d9532-0a2b-42a6-bbfb-18f3f987098a' 09:28:50.680 | ERROR | Flow run 'bouncy-dalmatian' - Flow could not be retrieved from deployment. Traceback (most recent call last): File "D:\Raghu\venv\lib\site-packages\prefect\engine.py", line 254, in retrieve_flow_then_begin_flow_run flow = await load_flow_from_flow_run(flow_run, client=client) File "D:\Raghu\venv\lib\site-packages\prefect\client.py", line 104, in with_injected_client return await fn(*args, **kwargs) File "D:\Raghu\venv\lib\site-packages\prefect\deployments.py", line 55, in load_flow_from_flow_run await storage_block.get_directory(from_path=None, local_path=".") File "D:\Raghu\venv\lib\site-packages\prefect\filesystems.py", line 100, in get_directory shutil.copytree(from_path, local_path, dirs_exist_ok=True) File "D:\Raghu\Python\lib\shutil.py", line 566, in copytree with os.scandir(src) as itr: FileNotFoundError: [WinError 3] The system cannot find the path specified: 'C:\\Users\\HOME\\AppData\\Local\\Temp\\tmpl2td_44jprefect' 09:28:51.663 | INFO | prefect.infrastructure.process - Process 'bouncy-dalmatian' exited cleanly.

✅ 1

Anat Tal Gagnon

08/28/2022, 10:54 AM

Hi there! does anybody knows if there's a command to check which agents has started in the background?

✅ 1

Anat Tal Gagnon

08/28/2022, 11:02 AM

and... how can I run "prefect deployment run" with custom parameters?? from the cli

✅ 1

Benson Mwangi

08/28/2022, 5:17 PM

Hello, using prefect version 0.15.9 and was wondering if there's a way to get the prefect cron scheduler to work at sub minute intervals? At-least 30 seconds. Thank you!

✅ 1

ibrahem

08/29/2022, 7:19 AM

Hi, Can we use external database for Prefect 2.0 instead of sqlite?

✅ 1

Faheem Khan

08/29/2022, 7:50 AM

Hi everyone, am I missing something in the latest upgraded prefect from 2.0.4 to 2.2.0. I started agent with a random queue name i.e. q1 and added "--work-queue q1" argument to prefect deployment. now when I start my flow it freezes after sometime. the same code was working fine with prefect 2.0.4. I am using Dask, docker containers and minio storage. Cheers

✅ 1

Andreas Nord

08/29/2022, 8:01 AM

Hi! For which release is it planned to add docker image as a way of storing flows?

✅ 1

Enrique

08/29/2022, 9:20 AM

Hi Prefect! 🙂 I'm updating my flows from 1.X to 2.X. Is there a way to set a task timeout as we had in the previous versions? thanks a lot!

✅ 1

Hedgar

08/29/2022, 11:27 AM

I have flow that suppose to scrape data, clean and save data to CSV file via pandas

to_csv()

method. The path was created by pathlib:

file-path = Path("data/fresh-data.csv”)

file-path.parent.mkdir(parents=True,exist_ok=True)

Why can't I see fresh data, is there an extra thing am missing. Just trying stuff with prefect 2 on my local system

✅ 1

Brandon T. Kowalski

08/29/2022, 12:16 PM

Quick question about Flow and Deployment IDs. Are these IDs static for the lifetime of the respective flow and deployment? (i.e. don't change on update)

✅ 1