Hello all, I am running a prefect ECS agent as a service in AWS ECS. Now I am trying to register a flow run in Prefect Cloud using the Docker storage. I created a

ECSRun

run configuration for the

run_config

argument in the Flow. If I use the

env

argument to provide the

AWS_ACCESS_KEY_ID

and

AWS_SECRET_ACCESS_KEY

it works! However, I don’t want to put my credentials in the source code. I am trying to use the

task_definition_arn

argument, but I am not sure if it is the correct way, because the image of the task should be the Flow docker image. Can you give me any tip or advise? Thank you very much!

Do mapped tasks only behave in a parallel fashion on dask? I am running on k8s using job per flow and all my mapped tasks log in order sequentially - also theres a huge gap between two mapped tasks I was under the impression that if you have;

p1_res = extract_raw_from_api.map(
        asset_location=asset_locations,
        extract_date=unmapped(extract_date),
        extract_period_days=unmapped(extract_period_days),
    )
    p2_res = process_2.map(p1_res)
    p3_res = process_3.map(p2_res)

Then as soon as task 1 from p1_res is done then the corresponding task should start in process_2? As it stands no process_2 tasks start until many of the extract_raw_from_api tasks are already complete...

Hello all, I was wondering if someone can point me in the right direction. (Please let me know if I need to clarify 🙂) I have this working as a POC in Dagster and was wondering how this example would work in Prefect This would be testing a custom task class ... multiple instances of a smaller task ... configurable params/secrets I have a legacy module that accepts 20+ parameters. Some of these parameters will be required (name), some will need to be secrets (tokens), and some vary (description). Depending on the combination of these parameters the module will do specific things: Query a database and dump to a bucket, grab data from a bucket and load in the db, and other tasks. Instead of accepting command line args, I want to "prefect-ify" that module and supply some of the configuration directly into that module. Would this be a combination of parameters and secrets? Can these be loaded from a toml file? Next, I want to wrap that base module into a smaller/specific task: example: query_db_and_dump_to_bucket(). This smaller task will have required values(db host, db username, db pass, table_name). Just to point out these values are not required in the base module. In my flow, I would want to call query_db_and_dump_to_bucket() again and again using different tables names.

I have a (I think simple) question around exposing more complete output to the prefect logger – I'm deliberately crashing a task that uses a

DbtShellTask

to execute a dbt project under the hood, and am not supplying the necessary ENV vars that

dbt

needs to run. If I have prefect run my flow that calls this shell task, the only failure message I see is the very last line of what dbt spits out, i.e.

[2021-01-27 16:34:13+0000] ERROR - prefect.run_dbt | Command failed with exit code 2
[2021-01-27 16:34:13+0000] ERROR - prefect.run_dbt |   Could not run dbt

this is obviously not terribly informative, and I'd like for prefect to forward to me the entirety of the error message that dbt generates, which is:

Encountered an error while reading profiles:
  ERROR Runtime Error
  Compilation Error
    Could not render {{ env_var('DB_USERNAME') }}: Env var required but not provided: 'DB_USERNAME'
Defined profiles:
 - default

For more information on configuring profiles, please consult the dbt docs:

<https://docs.getdbt.com/docs/configure-your-profile>

Encountered an error:
Runtime Error
  Could not run dbt

what kind of additional steps need to be taken in order for me to have prefect forward all the lines of output rather than just the final one? it's not clear to me where or how I would need to adjust the default prefect logger to set this up

Hi all, I'm running a flow on Kubernetes using DaskExecutor with 4 workers, but it doesn't create 4 worker pods. Ideally, I'd like to be able to scale to >1 node (not using an existing Dask cluster). (I tried using dask_kubernetes.KubeCluster, but I get an error: AttributeError("'NoneType' object has no attribute 'metadata'"). I'm not sure if this would solve my problem anyway.) Is there a way I can run a flow such that I can scale up to multiple nodes? Thank you

Hello! I am trying to test a few flows in our prod environment by running a local agent in a docker container that simulates the environment. For some reason the agent is not picking up the registered flows even though they share labels. Does anyone know why this would be happening? For reference, it runs successfully if I run a local agent in my dev environment

Hello, it's the right place to ask something about prefect errors?

Hello all, I have made some progress with using docker storage in my flow, however, I am now getting this error:

ValueError: Your docker image failed to build!  Your flow might have failed one of its deployment health checks - please ensure that all necessary files and dependencies have been included.

The full traceback is

Step 12/22 : WORKDIR /app
 ---> Running in b820b907ef61
Removing intermediate container b820b907ef61
 ---> fee83981b109
Step 13/22 : RUN groupadd --gid ${GID} ${USERNAME}     && useradd --uid ${UID} --gid ${GID} --shell /bin/bash -M ${USERNAME}     && chown -R ${USERNAME} /app     && chgrp -R ${USERNAME} /app
 ---> Running in 7f6c8b47c9e3
groupadd: invalid group ID 'rainbow'

Removing intermediate container 7f6c8b47c9e3
The command '/bin/sh -c groupadd --gid ${GID} ${USERNAME}     && useradd --uid ${UID} --gid ${GID} --shell /bin/bash -M ${USERNAME}     && chown -R ${USERNAME} /app     && chgrp -R ${USERNAME} /app' returned a non-zero code: 3
Traceback (most recent call last):
  File "integration.py", line 11, in <module>
    from change_password import *
  File "/home/mblau/projects/experian-integration/change_password.py", line 5, in <module>
    from integration import modify_config
  File "/home/mblau/projects/experian-integration/integration.py", line 338, in <module>
    flow.register(project_name="test")
  File "/home/mblau/.local/lib/python3.8/site-packages/prefect/core/flow.py", line 1665, in register
    registered_flow = client.register(
  File "/home/mblau/.local/lib/python3.8/site-packages/prefect/client/client.py", line 782, in register
    serialized_flow = flow.serialize(build=build)  # type: Any
  File "/home/mblau/.local/lib/python3.8/site-packages/prefect/core/flow.py", line 1451, in serialize
    storage = self.storage.build()  # type: Optional[Storage]
  File "/home/mblau/.local/lib/python3.8/site-packages/prefect/storage/docker.py", line 360, in build
    self._build_image(push=push)
  File "/home/mblau/.local/lib/python3.8/site-packages/prefect/storage/docker.py", line 427, in _build_image
    raise ValueError(
ValueError: Your docker image failed to build!  Your flow might have failed one of its deployment health checks - please ensure that all necessary files and dependencies have been included.

what am I doing wrong?

Is there a particular "Dask-ic" way to split a big CSV and make sure the compressed chunks are a specific size?

Hi all! I'm having a bit of an issue with an empty UI when running

prefect server start

. I'll post the logs in the thread.

I'm missing something. It's possibile to launch a flow of docker images tasks?

I'm using something like:

def say_hello():
    container = CreateContainer(
        image_name="nonexistentname",
    )
    StartContainer(container_id=container, docker_server_url="arandomtext")
    WaitOnContainer(container_id=container)

and it's always successful. Even with correct data i don't see my image executed

Hi, We would like to do flow registers from Github Actions when a new flow gets merged into master. To accomplish this, we have our graphql end point protected through SSL using a certificate and we define it as an environment variable PREFECT__CLOUD__GRAPHQL=https://protected.grpahql.endpoint.com so prefect from Github Action runner linux image has access to this. So this endpoint is only accessible though https and using ca certificates. I've tested that this end point works on Chrome with the give certificate and it works fine. However, when we try to call it through Github Actions runners i keep getting: prefect.TaskRunner | FAIL signal raised: FAIL('Failed to register flow_test, error: 400 Client Error: Bad Request for url: https://protected.grpahql.endpoint.com/\nThis is likely caused by a poorly formatted GraphQL query or mutation but the response could not be parsed for more details') Here is some of the things i've tried 1) configured the certificate by adding it to the Github actions runner image at /usr/local/share/ca-certificates/graphCert.crt and then 'dpkg-reconfigure ca-certificates' 2) Declared python env variable to tell it where too look for the ca certs REQUESTS_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt Does any one of you have a similar setup where your graphql endpoint is protected in this way? When it's not protected our setup works just fine... Thank you in advance...

Hi Prefect team! Can someone pls confirm or clarify the following understanding that I have about how Prefect works with Docker: • I can run an app inside a docker container, any dependencies needed, e.g. pandas library, will be installed inside that container. Our repo already has a Dockerfile and a docker-compose.yml which build an image and run the app inside a container successfully. Necessary commands to do that are

docker-compose build app

, then

docker-compose run app

.  • Prefect Agent runs flows. The Local Agent runs flows with the local setup. This is not feasible for production since we’d need to manually make sure that whichever machine is hosting the flow has all dependencies installed. Better: run flows within a docker container. This way we don’t have to worry about dependencies, since every machine will simply be running the container. To run flows in containers we need to use the Docker Agent. • The Docker Agent DOES NOT REQUIRE DOCKER STORAGE, but can work with Docker Storage if desired. The Docker Agent can run with a locally available image (image="example/image-name:with-tag"). If Docker Storage is used, you can provide a url (

<http://registry_url|registry_url>

)to where your image is hosted, e.g. Docker Hub, and that url is used to get the image. • I should be able to run the Docker Agent with the image I had previously created with docker-compose. (It doesn’t seem to be able to find the image, even though I specify the absolute path. )

Hi, we’re using a paid plan on Prefect Cloud but we are now seeing this error when trying to register new flows: 

This tenant already has the maximum number of flows

. Don't' see there's any mention of upper limit on flows? Any help will be appreciated!

Hi, I would like to launch a flow where each task is a docker image(with a CMD altredy set, just need some volume mount). For me it's important that each task in UI results as function name and not like each docker task name.(https://docs.prefect.io/api/latest/tasks/docker.html)

Hey, I use Prefect+Dask distributed and in the code the prefect flow is executed multiple times consecutively. The problem is that in the Dask profiler I see that the memory usage keep growing linearly although each execution (prefect flow) can be considered as a "clean run" and the amount of used memory should stay constant. Has anyone tackled similar issue?

I'm trying to use KubeCluster with DaskExecutor in a simple flow on GCP:

@task
def get_mean():
    import dask.array as da
    array = da.ones((1000, 1000, 1000))
    return array.mean().compute()

@task
def output_value(value):
    print(value)

with Flow("Static Dask Cluster Example") as flow:
    res = get_mean()
    output_value(res)

flow.run_config = KubernetesRun(labels=["dev"], image="my-image")

def get_cluster(n):
    pod_spec = make_pod_spec(image="my-image",
                              memory_limit='1G', memory_request='1G',
                              cpu_limit=1, cpu_request=1)
    return KubeCluster(pod_spec, n_workers=n)

flow.executor = DaskExecutor(cluster_class=get_cluster, cluster_kwargs={'n': 2})
flow.storage = GCS(bucket="my-bucket")
flow.register(project_name="tutorial")

However, I get a strange error:

[2021-01-28 12:11:41+0000] ERROR - prefect.CloudFlowRunner | Unexpected error: AttributeError("can't set attribute")
Traceback (most recent call last):
  File "/opt/conda/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/opt/conda/lib/python3.8/site-packages/prefect/engine/flow_runner.py", line 418, in get_flow_run_state
    with self.check_for_cancellation(), executor.start():
  File "/opt/conda/lib/python3.8/contextlib.py", line 113, in __enter__
    return next(self.gen)
  File "/opt/conda/lib/python3.8/site-packages/prefect/executors/dask.py", line 211, in start
    with self.cluster_class(**self.cluster_kwargs) as cluster:  # type: ignore
  File "k8_flow.py", line 33, in get_cluster
  File "/opt/conda/lib/python3.8/site-packages/dask_kubernetes/core.py", line 414, in __init__
    super().__init__(**self.kwargs)
  File "/opt/conda/lib/python3.8/site-packages/distributed/deploy/spec.py", line 274, in __init__
    super().__init__(
  File "/opt/conda/lib/python3.8/site-packages/distributed/deploy/cluster.py", line 69, in __init__
    self.name = str(uuid.uuid4())[:8]
AttributeError: can't set attribute
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/opt/conda/lib/python3.8/site-packages/distributed/deploy/spec.py", line 651, in close_clusters
    if cluster.status != Status.closed:
AttributeError: 'KubeCluster' object has no attribute 'status'
Exception ignored in: <function Cluster.__del__ at 0x7f99f6b89e50>
Traceback (most recent call last):
  File "/opt/conda/lib/python3.8/site-packages/distributed/deploy/cluster.py", line 113, in __del__
    if self.status != Status.closed:
AttributeError: 'KubeCluster' object has no attribute 'status'

This is running on a KubernetesAgent on GKE with the RBAC permissions mentioned in the documentation. The docker images I'm using are running dask-kubernetes==0.11.0, distributed==2020.12.0, prefect==0.14.5. Thank you in advance

Thanks for the swag Prefect Team! You rock! 🚀

Hi everyone! Is there some way to show statistics of a task in the UI? Currently we have a task that replace an entire table of our db with new data. Then we push a message to our Slack with the old and new counts of rows. If the difference is significant, then the task is considered failed. Is there some way to display these counts in the Prefect UI?

Hello all, I have been successful in getting prefect to build and run a flow on a docker container. However, I am getting an error of

Unexpected error: ValueError('Could not infer an active Flow context.')

and I am struggling to find relevant information on how best to debug this. I thank you all in advance for helping!

What is the best way to pass non-serializable objects into tasks? For example, I have a database connection or a sftp connection that I want to reuse. The use is that I have many files I need to write to a sftp, and want to use mapping functionality. But creating many connections is costly (and it seems that this sftp server is limiting the number of connections)

Hi Prefect, What's the recommended way of dealing with Task failures? For example, I want to write to a db table or send a message to Kafka if a certain task has failed. From the documentation it seems that there are two ways of handling failure of the task: • Implement a State Handler to deal with the failed state. • Implement a Task to deal with the failed stage and use a Trigger with

any_failed

? I'm not sure which is the most idiomatic way of dealing with this failure, since updating a database or sending a message to Kafka itself is prone to failure (and "negative engineering" to deal with that failure). Does Prefect have an opinion/guidelines how to deal with this?

How do I run an agent on ECS? I've tried the docs on https://docs.prefect.io/orchestration/agents/ecs.html#agent-configuration but it doesn't work. I always get an error. The error stack is on the thread

Hello all! Quick question: is it possible to pass a logger.xml file to Prefect logger? I want to implement log rotation for each task in my pipeline Thank you!

Q: Is there a workflow

model

that one can export out from Prefect? I intended to use it to auto generate Python code from it

when using the ECS runconfig, is one of the 3 methods for attaching the ecs task (

task_definition_arn

,

task_definition_path

,

task_definition

) to the flow considered to be canonical? i'm trying to weigh the pros and cons of having some other process like terraform handle management of the task definition (thereby only leaving the ARN to be supplied to prefect) vs. using prefect to handle that

Is there any movement towards having a more airflow like backfilling feature - its literally the one thing I miss - I would really love to be able to set a start date for a schedule in the past and for the flows to catch up to the present - currently I think the only way to achieve is to make use of the params to differentiate future runs for past and to manually start all the backfill flows - is there perhaps a better way currenty?

Hello! I am creating a proof of concept for my team using Prefect. I am not sure how to implement something, hope this is the right place to ask. I am stringing together 3 tasks in a flow. I want them to run one after the other, but none of them output anything (they interact with external services). Can I do this using the functional API, or do I have to use the Imperative API since there is no data dependency that I pass along?

If anyone could provide any guidance that would be great. I am trying to pass an instance of a custom task to a “sub flow”. Just wanted to know if that is even possible. My main flow internally calls StartFlowRun for another flow which can take parameters/results from the main flow. The dependencies and DAG look fine in the UI. If I pass a string for "author_info" my subflow/subtask gets that string. If I try to pass author_instance I get an exception:

Unexpected error: TypeError("Object of type 'TestTask' is not JSON serializable",)

TestTask is a custom task type I created and the run method returns a custom return type I created as well.