Hello, I am a beginner at prefect pipeline orchestration. This is the first time I've attempted to dockerize my prefect pipeline and register it to local UI server. After submitting my flow, it seems to be stuck with a backlog of "late runs".

Hi folks, I am facing this error - whenever my flow involves spinning up a considerable number of workers (more than 100 dask workers) - note the same flow runs fine when I set a smaller number of workers… (the error is invoked after the flow goes into Running state, but before any tasks are run)

Unexpected error: TimeoutError()
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 410, in get_flow_run_state
    with executor.start():
  File "/usr/local/lib/python3.7/contextlib.py", line 112, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/executors/dask.py", line 239, in start
    with Client(self.address, **self.client_kwargs) as client:
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 744, in __init__
    self.start(timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 949, in start
    sync(self.loop, self._start, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/distributed/utils.py", line 339, in sync
    raise exc.with_traceback(tb)
  File "/usr/local/lib/python3.7/site-packages/distributed/utils.py", line 323, in f
    result[0] = yield future
  File "/usr/local/lib/python3.7/site-packages/tornado/gen.py", line 735, in run
    value = future.result()
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 1046, in _start
    await self._ensure_connected(timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/distributed/client.py", line 1108, in _ensure_connected
    await asyncio.wait_for(self._update_scheduler_info(), timeout)
  File "/usr/local/lib/python3.7/asyncio/tasks.py", line 449, in wait_for
    raise futures.TimeoutError()
concurrent.futures._base.TimeoutError

it seems to me the solution to avoid this timeout is to add a task that waits for the workers to be ready - anyone else run into this ? (I am using a DaskKubernetesEnvironment for execution on AWS EKS)

Hi! I am trying to setup prefect with dask on multiple machines, but I keep getting this error on the remote machines and I think I need to change some config or set some env variable but I can't figure out what that is

Hello, I'm using

slack_notifier

as my flow's state_handler, but it does not seem to get called when my task fails. I get notified when it starts and ends. My pipeline is set such that failing_task -> task (trigger_will_fail) -> recovery_task (task triggers only on fail and sets state to success, otherwise is skipped). I ideally want to be notified that we encountered a failure, even though we do end up doing failure recovery and setting the final state to a success.

I cannot find my dockerized prefect flow in my ".prefect/flows" directory. I'm attempting to run this code, but to no avail because the path to the file doesn't exist:

from prefect import Flow
my_flow = Flow.load('thinknum-etl')
my_flow.run()

I followed the prefect quickstart but my web ui can't seem to connect to the graphql

leading to the dashboard to constantly be stuck in this state

@Jeremiah, you’re out of date. You guys are #1 now 🙂

Hi everyone. Is anyone familiar with the manner in which a manual_only task can be triggered from the UI? The logger results in

finished task run for task with final state: 'Paused'

, and I'm not sure as to how this would be fixed.

hi! is there a way to make a parameter default to the current datetime?

pip install prefect[aws,dask_cloudprovider]

currently fails to install (without some extra work) due to conflicting dependencies. A solution I found was:

pip install --use-feature=2020-resolver prefect[aws,dask_cloudprovider]

If you pin versions in requirements.txt or use pipenv or pip-tools, you can then run

$ pip freeze | grep boto
aiobotocore==1.0.7
boto3==1.12.32
botocore==1.15.32

to see the versions you need to pin to.

Happy Wednesday - I'm encountering issues with several of the docker images. They seem to be related to the healthcheck.py file. I've tried

prefecthq/prefect:0.12.1-python3.7

and

prefecthq/prefect:0.12.5-python3.7

, which resulted in the following error:

AttributeError: module 'types' has no attribute 'CellType'

I've also tried

FROM prefecthq/prefect:0.11.5-python3.7

, which resulted in this error:

ModuleNotFoundError: No module named 'prefect.core.parameter'

Hello, I'm getting the

Can't connect to  <http://localhost:4200/graphql>

error when running my server on a remote instance. I've upgraded my prefect version so that might be why. Here's my config.toml

[server]
 [server.ui]
  graphql_url = "<http://url:4200/graphql>"
 [server.api]
  graphql_url = "<http://url:4200>"

Here's my output for

prefect diagnostics

{
  "config_overrides": {
    "backend": true,
    "graphql_url": true,
    "server": {
      "api": {
        "SLACK_WEBHOOK_URL": true,
        "graphql_url": true
      },
      "ui": {
        "graphql_url": true
      }
    }
  },
  "env_vars": [],
  "system_information": {
    "platform": "Linux-5.3.0-1032-aws-x86_64-with-Ubuntu-18.04-bionic",
    "prefect_version": "0.13.2",
    "python_version": "3.6.8"
  }
}

Hello - I'm getting the following error:

ValueError: Task <Task: ThinkNumGetHistoryTask> has retry settings but some upstream dependencies do not have result types

Would anyone know how to address this error? I tried adding retry to all the tasks in the flow, but didn't seem to resolve the issue.

Really sorry to keep bombarding the channel with so many questions - I think this will be the last one for the day. My question is: when you dockerize your flow, does this docker image replace the image of your dask workers for the purpose of executing your flow? I was under the impression that it does. However, I am getting this error w/ a custom library when running the flow from Prefect Server.

ModuleNotFoundError: No module named 'alternative_data_pipelines'

Note that the package

alternative_data_pipelines

exists in my docker file and has been pip-installed. No issues with the health checker - everything has been registered successfully and runs ok locally and from the docker container.

This message was deleted.

Hi - (1) Does

flatten

filter out

None

by default?

Hey folks, we've been putting together a bunch of custom Tasks for internal use. Lets say one of those is called DoSomething. I'm wondering what the correct approach is for dealing with Task parameters. Providing the task with a Parameter in the flow fails because the task doesn't know what to do with it, and calling

run

on the parameter in the flow definition is clearly wrong. The Parameter instance/s are passed in as a list since the users of the Task need to be able to pass an arbitrary number of unnamed parameters. Are there any examples of unpacking the parameters in an efficient way? So far I have

params = [param.run() if isinstance(param, Parameter) else param for param in params]

I've also tried running the Parameter through a

@task

function but when I pass the results of that task to DoSomething it complains that FunctionalTask is not compatible.

Hi all, just getting started and coming from Airflow, I cannot find documented where task .py files should live (equivalent of airflow dags folder)? Thanks in advance!

Can I set an upper bound for tasks run in parallel using

apply_map

? My problem is, that when I map more than 60 Tasks at once, each having an open connection to an external database, the flow will remain in a running state and never terminate.

💃 Just arrived! Love it that I am a part of something so awesome. I am new to Slack and Open Source contributions. Can someone perhaps help me to get started to contribute to this project?

Hello the community, trying to get acquainted with Prefect to keep up with my consultants here.

I of course have a question 🙂 . All of my so-called ETLs look the same, but it doesn't match the tempaltes I can find in the doc. • I have a task returning a list of file paths ==> extract • I have a task transforming these files in a list of dict • for each dic I have a task to load I just want to be able to map the transform to each file, then map each returned dic to its load task. However I can't find a way to return only the relevant list to the relevant load task

Hi folks - I am facing DiskPressure warnings and dask-worker pods getting evicted when running the flow on EKS

78s         Warning   Evicted                pod/dask-worker-855c59c78d-qwv5v       The node had condition: [DiskPressure].
78s         Warning   Evicted                pod/dask-worker-855c59c78d-fvc4q       The node had condition: [DiskPressure].

I am wondering if someone faced similar issues ?

If I am to load 50 tables from 1 db to another. Would you implement that as 1 flow per table or 1 flow mapped over all tables? I guess if I make a loop to register 1 flow per table it would be easier to rerun only one table. And maybe easier with debugging. But it would be more cluttered in the UI.

The UI starts crashing with large flows it seems - I wasn't able to mark the flow as failed, and then when I return to the dashboard, I cant access any of the flows - and it shows 0 for flows, and runs, even though I have definitely more than one flow and run ...

Hello friends, I’m running a

BigQueryTask

on Prefect Cloud and it failed with the error “No heartbeat detected from the remote task; marking the run as failed”. Was this Zombie killer? It worked fine locally. How can I ensure this long-running task completes?

Hello! is there an option that allows a flow to execute in only one instance? in my case, the flow has not finished yet, but the scheduler started it again.