prefect-community #prefect-community

Andy Dang

07/15/2022, 5:10 PM

Hi! I just started to look into Prefect as an alternative to AWS Step Functions. We use Step Functions to orchestrate our Databricks jobs with AWS Lambda - and Prefect supports Databricks (so the main actual logic is in DB - we just need Prefect to orchestrate tasks) However what I can’t figure out is where to run the prefect agent to manage my Databricks jobs? Do I spin up an EC2 instance and kick it off there? Can I run it as a Docker container in my ECS cluster? Doubt that I can run it as a Lambda (though that would be amazing), but it doesn’t look like there’s an easy way to do it. Any recommendations?

✅ 1

Divya

07/16/2022, 1:30 AM

Hello, I am trying to use Amundsen with Prefect. Is it possible using Prefect if yes are there any working examples? Right now I am trying to schedule an Amundsen code in Prefect but I am getting the error as below. Thank you, Divya

✅ 1

Michael Reynolds

07/16/2022, 11:24 PM

hello, i am hacking away at something and i had a question about prefect 2.0 that i cannot seem to find an answer to.

Michael Reynolds

07/16/2022, 11:33 PM

Copy code

23:29:26.313 | ERROR   | Flow run 'mustard-quetzal' - Encountered exception during execution:
Traceback (most recent call last):
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/engine.py", line 521, in orchestrate_flow_run
    flow_call
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 120, in run_sync_in_interruptible_worker_thread
    cancellable=True,
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 662, in __aexit__
    raise exceptions[0]
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 702, in _run_wrapped_task
    await coro
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/to_thread.py", line 32, in run_sync
    func, *args, cancellable=cancellable, limiter=limiter
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 96, in capture_worker_thread_and_result
    result = __fn(*args, **kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/goron/main.py", line 35, in run_pipeline
    messages = poll_kafka( conf[ 'kafka' ] ).result()
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/futures.py", line 210, in result
    self._result, timeout=timeout, raise_on_failure=raise_on_failure
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 221, in sync
    return run_async_from_worker_thread(__async_fn, *args, **kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 136, in run_async_from_worker_thread
    return anyio.from_thread.run(call)
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/from_thread.py", line 49, in run
    return asynclib.run_async_from_thread(func, *args)
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
    return f.result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 435, in result
    return self.__get_result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/futures.py", line 220, in _result
    return final_state.result(raise_on_failure=raise_on_failure)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/orion/schemas/states.py", line 136, in result
    raise data
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/task_runners.py", line 314, in _run_and_store_result
    self._results[run_key] = await run_fn(**run_kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/engine.py", line 799, in begin_task_run
    client=client,
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/engine.py", line 905, in orchestrate_task_run
    result, serializer="cloudpickle"
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/states.py", line 130, in return_value_to_state
    return Completed(data=DataDocument.encode(serializer, result))
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/orion/schemas/data.py", line 42, in encode
    blob = lookup_serializer(encoding).dumps(data, **kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/serializers.py", line 59, in dumps
    data_bytes = cloudpickle.dumps(data)
  File "/home/myapp/.local/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/home/myapp/.local/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 633, in dump
    return Pickler.dump(self, obj)
TypeError: can't pickle cimpl.Message objects
                                                                                                            (venv) michael.reynolds@mac1319 goron %

Michael Reynolds

07/16/2022, 11:34 PM

Copy code

23:29:26.313 | ERROR   | Flow run 'mustard-quetzal' - Encountered exception during execution:
Traceback (most recent call last):
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/engine.py", line 521, in orchestrate_flow_run
    flow_call
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 120, in run_sync_in_interruptible_worker_thread
    cancellable=True,
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 662, in __aexit__
    raise exceptions[0]
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 702, in _run_wrapped_task
    await coro
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/to_thread.py", line 32, in run_sync
    func, *args, cancellable=cancellable, limiter=limiter
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 96, in capture_worker_thread_and_result
    result = __fn(*args, **kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/goron/main.py", line 35, in run_pipeline
    messages = poll_kafka( conf[ 'kafka' ] ).result()
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/futures.py", line 210, in result
    self._result, timeout=timeout, raise_on_failure=raise_on_failure
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 221, in sync
    return run_async_from_worker_thread(__async_fn, *args, **kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/utilities/asyncio.py", line 136, in run_async_from_worker_thread
    return anyio.from_thread.run(call)
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/from_thread.py", line 49, in run
    return asynclib.run_async_from_thread(func, *args)
  File "/home/myapp/.local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
    return f.result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 435, in result
    return self.__get_result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/futures.py", line 220, in _result
    return final_state.result(raise_on_failure=raise_on_failure)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/orion/schemas/states.py", line 136, in result
    raise data
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/task_runners.py", line 314, in _run_and_store_result
    self._results[run_key] = await run_fn(**run_kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/engine.py", line 799, in begin_task_run
    client=client,
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/engine.py", line 905, in orchestrate_task_run
    result, serializer="cloudpickle"
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/states.py", line 130, in return_value_to_state
    return Completed(data=DataDocument.encode(serializer, result))
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/orion/schemas/data.py", line 42, in encode
    blob = lookup_serializer(encoding).dumps(data, **kwargs)
  File "/home/myapp/.local/lib/python3.7/site-packages/prefect/serializers.py", line 59, in dumps
    data_bytes = cloudpickle.dumps(data)
  File "/home/myapp/.local/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/home/myapp/.local/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 633, in dump
    return Pickler.dump(self, obj)
TypeError: can't pickle cimpl.Message objects

Jelle Vegter

07/17/2022, 12:35 PM

Hey all, (Prefect 1.0) I’m struggling to figure out why flows execute properly locally, but when triggered through Prefect Cloud + Docker Agent running local doesn’t. The flows are stuck in submitted. Anyone have an idea? I can’t find information about the error message “prefect: error: unrecognized arguments: execute flow-run” . Thanks in advance!

Jack Sundberg

07/17/2022, 1:59 PM

Hi everyone, is there a recommended way to set up a Postgres connection pool for the Orion database? I typically use pgbouncer to build a transactional pool, but when I do this with prefect, I end up an error matching a known bug in sqlalchemy.

Emil Østergaard

07/17/2022, 9:05 PM

Hi, In previous versions (tried 2.0b4) it was possible to create an s3 storage block backed by minio, using the

/api/blocks

endpoint. In the 2.0b8 api I see endpoints for block schemas, block_types and block_document, but as far as I can tell none of these create a block? Through the UI, it is not possible to supply the necessary

client_kwargs

and

config_kwargs

. Is this no longer possible, or am I misunderstanding something? For reference I used a request like this:

Copy code

PAYLOAD=$(cat <<EOF
{
    "name": "minio",
    "block_spec_id": ${FILE_STORAGE_ID},
    "data": {
        "base_path": "<s3://prefect-flows>",
        "key_type": "hash",
        "options": {
            "use_ssl": false,
            "key": "blablabla",
            "secret": "blablabla",
            "client_kwargs": {"endpoint_url": "<http://minio:9000>"},
            "config_kwargs":{"signature_version": "s3v4"},
        }
    }
}
EOF
)
BLOCK_ID=$(echo -n "$PAYLOAD" | curl -vs -XPOST -H "Content-Type: application/json" <http://localhost:4200/api/blocks/> -d@- | jq -r '.id')

✅ 1

Stefan

07/17/2022, 9:46 PM

In the API, what is the difference between "Create a flow run from Deployment" and "Create a flow run"? In the latter you have to specify a Deployment id anyway I see.

✅ 1

yu zeng

07/18/2022, 5:30 AM

hi, i call a create_flow_run task on prefect server, but got "ValueError: You have not set an API key for authentication"

✅ 1

Rajeshwar Agrawal

07/18/2022, 6:34 AM

hey Prefect! I have disabled heartbeat and Lazarus for one of my flow using UI, however it seems that prefect is still retrying the flow. I am seeing 2 log messages which matches

Beginning Flow run for '….'

which indicates that flow was retried by prefect. Any ideas? It seems that heartbeat is also not disabled, as we are seeing logs

No heartbeat detected from the remote task; marking the run as failed.

FYI we are using prefect server community Core Version 1.2.0+10.gafda99411

✅ 1

07/18/2022, 7:19 AM

Not sure where to post this but is there a way to get a task to repeat N number of times? Says I've got a pipeline that calls an APi which returns a json array. I need to execute a task per element in said array and report back?

✅ 1

xyzz

07/18/2022, 9:28 AM

is there a way to setup orion on a Kubernetes Cluster with RBAC disallowed?

✅ 1

Riccardo Tesselli

07/18/2022, 9:46 AM

hi all, I’m about to start a new project based on Prefect 2.0. I was wondering if there is a best practice on how to organise the folders. Is it convenient to have separated packages for “flows”, “tasks” and so on, or something else? Is there a good repo based on Prefect 2.0 from which to take inspiration?

✅ 1

Abhishek Mitra

07/18/2022, 10:22 AM

I have a flow 'FlowA' registered in prefect cloud. Can I set a schedule for FlowA from another flow FlowB?

✅ 1

Paul Lucas

07/18/2022, 11:17 AM

Hi I’m trying to implement a dbt flow that runs our process but when I look at the tasks I just get

DbtShellTask

which isn’t very helpful. Is there a way to rename the individual tasks to something more useful? I’ve tried using

name

and

task_run_name

arguments but without any luck.

✅ 1

haris khan

07/18/2022, 12:44 PM

what is Perimeter class in prefect 1.0 in prefect orion ? just change it to a Task class ?

Priyank

07/18/2022, 1:03 PM

Hi there! I am running a local prefect server and now wants to collect some useful data from the database, like start_time, end_time, duration, for each and every flow and their tasks. We can clearly see this data on prefect dashboard but how I can query this data from the db. I tried listing all the tables in database but didn't get how they are linked to each other. Can you share a visual representation of how the tables in the database are linked to each other? and some queries too cause duration and other things are already on the dashboard so prefect must be querying this data somehow? or is there an easy way to get this info. And also can we get how many times a task failed? At the end we want to use this data for our analysis, how much time each task is taking and how it is changing, is there a particular time when our task fails and more? We are using prefect 1.0 for now.

Emil Barbuta

07/18/2022, 1:28 PM

Hello! I'm trying to create a flow of flows in which some child flows depend on others. When trying to make 2 of them run sequentially using the following code snippet,

wait_for_flow_run

fails by throwing

TypeError: Object of type FlowRunView is not JSON serializable

. Am I passing the wrong object type as input?

Copy code

from prefect.tasks.prefect import create_flow_run, wait_for_flow_run

with Flow("run-pipeline") as flow:
    flow_a_id = create_flow_run(flow_name=flow_a.name, ...[SOME INPUT PARAMS])
    flow_a_wait = wait_for_flow_run(flow_a_id)

(I'm using prefect 1.0.0)

Josh Paulin

07/18/2022, 1:55 PM

Is there any documentation on which permissions the Slack app requires?

tas

07/18/2022, 3:01 PM

Hi there! Is there a documentation on how to setup prefect server for dask cluster that is running on the same machine but in docker containers?(So scheduler is in a container and its workers are also in separate containers)? I've created that cluster using our codebase, but then whenever running the flow from Prefect UI, the task goes to dask scheduler, then worker, but then it fails with "requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url". during initialising the run (it fails with that on dask worker). The task runs when I run it locally with flow.run(). Is there a config var that would point it to outside of dask container?

Tarek

07/18/2022, 3:02 PM

Hello Prefect folks! For my flow I am using

KubernetesRun

and

Docker

registry for storage. In our Gitlab CI/CD pipeline, we want to register the flow when the feature branch is merged into main. However one problem we’re facing is that our CICD pipeline runs on docker, and since Prefect builds the docker image in the registration step, our pipeline fails (docker in docker). I’ve been looking for workaround around this and would be grateful for any ideas: I can tell you about what I have tried so far in the comments

Pranit

07/18/2022, 3:18 PM

I installed prefect and was able to make cloud run jobs on my local few days ago. But today I am not able to get prefect server started. It says

-sh: prefect: command not found

Am I missing some step?

Farooque Shaikh

07/18/2022, 3:37 PM

I am trying to run a graphql query in Python using the below code: params={ "date_obj":"2022-07-18T013226.584864+00:00", "curr_obj":"2022-07-18T053226.584864+00:00", } query=client.graphql( { 'query($date_obj: timestamptz=': { 'flow_run(where: {start_time: {_gte: $date_obj}})': { 'flow_id' } } however the parameter value for date_obj is not getting passed. Can someone please help on this

rectalogic

07/18/2022, 5:49 PM

I setup MFA for my account, but it never prompts for the token when I login. Any idea why?

Scott Aefsky

07/18/2022, 6:20 PM

Hi all. I'm trying to create flow runs using the GraphQL API (Prefect 1.0), and am having trouble getting the syntax right to pass a parameter to my flow. Any help you could provide would be greatly appreciated. Details in thread.

Josh Paulin

07/18/2022, 7:50 PM

Hello. Getting an error when trying to create a new project with the same name as one I just deleted

Copy code

Traceback (most recent call last):
  File "/Users/joshuapa/.pyenv/versions/3.9.9/lib/python3.9/site-packages/prefect/client/client.py", line 1055, in create_project
    res = self.graphql(
  File "/Users/joshuapa/.pyenv/versions/3.9.9/lib/python3.9/site-packages/prefect/client/client.py", line 464, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['create_project'], 'message': 'Uniqueness violation.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

Shaoyi Zhang

07/18/2022, 8:35 PM

Hi, we are using Prefect Cloud 1.0. One member on our team is having trouble accepting invitations to a tenant. He already has access to tenant B, but when he tries to accept the invitation for tenant A, he got

401 unauthorized

error. This error applies to both the email invitation and the invitation in Prefect Cloud UI. A different team member was able to accept invitation to tenant A recently.

✅ 1

Christian Nuss

07/18/2022, 8:57 PM

question: is doing

.run(...)

on a task within a task an antipattern?

Emerson Franks

07/18/2022, 11:39 PM

Hopefully an easy one. I have a flow that needs to run 3 functions/tasks in order and I can't seem to find the correct syntax to make them run sequentially. Currently, I have simple code that looks like this: