Is it possible to return a Socket from a task?

Hi, I'm looking into running Prefect on Kubernetes clusters in different regions, and had a question about handling GCSResults. If we use the

@task(result=GCSResult(bucket=<bucket>))

method of configuring a task result, is the bucket path fixed at flow registration time? If so, is there a way it can be dynamically set at flow run time? What I'm hoping to do is have flows that can be registered to run on different clusters (using agent labels), and have their GCSResult bucket path be configured via an env var on the cluster. That way we can re-use the same flow code across different clusters but have different results buckets depending on the cluster.

Do you need to set the image on ECSRun for the task to use a a specific image or will it associate the Docker storage pushed to ECR in the same flow as the base image to use?

Cryptic errors in ShellTask - Hi, I'm running a flow with a number of shellTasks, and encountering errors like:

2022-04-14 09:42:06-0600] ERROR - prefect.TaskRunner | Task 'ShellTask[0]': Exception encountered during task execution!
Traceback (most recent call last):
  File "/blah/python3.9/site-packages/prefect/engine/task_runner.py", line 880, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "/blah/python3.9/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/blah/python3.9/site-packages/prefect/utilities/tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "/blah/python3.9/site-packages/prefect/tasks/shell.py", line 131, in run
    tmp.write(command.encode())
AttributeError: 'list' object has no attribute 'encode'

I have a couple of questions: • Is there a way to tag shell_tasks such that you can see some clue as to which one failed? • Can I get a better description of the failure At the moment I have

shell_task = ShellTask(log_stderr=True, return_all=True, stream_output=True)

Is it a valid use-case to use Prefect for the sake of having a managed way of running tasks rather than "data pipelines"? I just want the workflow aspect of it.

Can I run Prefect fully on-premise without cloud features whatsoever?

So I have a Fargate ECS cluster, but when I try to scale up the cpu count with ECSRun() I get:

[14 April 2022 2:14pm]: An error occurred (InvalidParameterException) when calling the RunTask operation: No Fargate configuration exists for given values.

. The weird thing is that 4096 appears to be a valid entry: https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-cpu-memory-error.html. Is it possible I screwed up the definition of my fargate cluster?

What are the best practices for managing Docker image names?

docker_storage = Docker(
    registry_url=environ["REGISTRY_URL"],
    dockerfile="./Dockerfile",
    image_name="{edited}-prod-platform-prefect-{project}",
    image_tag="latest",
)

It seems like each flow could demand its own image in order to separate dependencies, which would mean creating an ECR repo for each workflow? I suppose this wouldn't be that difficult to script with Github Actions and aws-cli globbing a directory for workflow names?

Hey, quick question on Prefect Orion - Wanted to know if the caching on task based on the persistent result something going to be supported in the Prefect Orion. The one that is already available in the Prefect 1.0 with @task(target=“{task-name}“)

Is there any documentation on which prefect docker tag has what extras?

Are there any example flows involving pulling from different branches of the same repo using mapping?

prefect deployment create

says it should create or update a deployment but it is failing when the deployment exists (prefect 2.0) ?

Hi, Having some problems using getting around CORS issues with Prefect V2.0 accessible to another host on the lan while running in a docker container. Details in thread per @Kevin Kho

Hello! All my prefect flows are getting scheduled but not running. Any recommendations on where I can start to debug this?

Hi folks, in a prefect flow , for one particular task, I am trying to raise a SKIP signal

from prefect.engine.signals import SKIP
........
........
raise SKIP('Skipping all downstream dependencies.')

With this all the following tasks are being skipped and was expecting the flow state will be

skipped

. Though the following tasks are being skipped but at the end the flow becomes

successful

. I have two questions around : 1. Is this something expected? 2. If this is expected, whats the remedy . Do I need to explicitly use a state handler on top of it to make the flow

skipped

eventually

Hi! Is it possible to increase max scheduled flows in prefect? I found “max_scheduled_runs_per_flow” parameter but it seems that is is only available in prefect server not prefect.

we're still having issues with logs not loading, sounds like there was an incident yesterday but prefect.status.io says it should be resolved now? any known issues that are still being worked on?

Question on state_handler, is there anyway to pass arguments to state_handler? I want to update database table when flow or task fails. The arguments are database config information and credentials

Hi everyone. Is there a migration guide that discusses the changes between 1.0 and Orion? I have a couple of implementations on 1.0 that we'd like to migrate soon after Orion hits GA. One runs on Azure Kubernetes and the other one on AWS ECS. We are using Docker storage for both. I was wondering what you recommend to understand how the different concepts relate in both versions. Thanks!

Hi all, I am wondering if its possible to change parameters of a deployed flow in the Orion UI and trigger a quick run with the new parameters. That would be extremely useful, for example in the process exploring different configurations and sizes when setting up third party cloud services

I'm having problems with tasks stuck at the "Running" state however when I inspected running workers everything seems running fine and results are submitted to my database although nothing is updating prefect state. currently using "prefect cloud" with

prefect 1.2.0-python3.9

docker image. same as https://github.com/PrefectHQ/prefect/issues/3952 for the dev team: https://cloud.prefect.io/bitthebyte/flow-run/b30223e1-5308-48fe-aa0b-9326c6e48860 (this is the stuck workflow) I already tried restarting

Hello guys, I would like to know if in Prefect 2.0 it is possible to request the execution of a specific task and all its dependent upstream tasks. Thanks.

Hello guys, we use VictorOps as our incident management tool and I setup cloudhook to the flow to send notifications to the Victorops as web URL. I am trying to test that out but it seems like, I am not receiving messages to our VictorOps. Does anyone has any idea about that?

Hello folks, I'm running flows on Prefect 1.2 using ECSRun and my setup uses the approach as described in this issue under the Conclusions pt. 5. It uses the

task_definition_arn

with the containers named as

flow

While I'm able to trigger the flows, the flow crashes with the error

copilot/flow/8d31faa7f1ba   File "/usr/local/lib/python3.8/importlib/__init__.py", line 127, in import_module
copilot/flow/8d31faa7f1ba     return _bootstrap._gcd_import(name[level:], package, level)
copilot/flow/8d31faa7f1ba   File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
copilot/flow/8d31faa7f1ba   File "<frozen importlib._bootstrap>", line 991, in _find_and_load
copilot/flow/8d31faa7f1ba   File "<frozen importlib._bootstrap>", line 961, in _find_and_load_unlocked
copilot/flow/8d31faa7f1ba   File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
copilot/flow/8d31faa7f1ba   File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
copilot/flow/8d31faa7f1ba   File "<frozen importlib._bootstrap>", line 991, in _find_and_load
copilot/flow/8d31faa7f1ba   File "<frozen importlib._bootstrap>", line 973, in _find_and_load_unlocked
copilot/flow/8d31faa7f1ba ModuleNotFoundError: No module named '/root/'

However, i've ensured the flows folder from my project are under the PYTHONPATH and I can't entirely figure out what (if anything) that cloudpickle is trying to do to access those flows at the time of execution.

Hi, I'm new to Prefect. It's fantastic to deploy pipelines by using Python code. However, I had faced some issues when deploying a pipeline to Prefect. Here's my scenario. • I want to run Prefect Agent in the K8s cluster, and I successfully plug the agent into our cluster. • I built a docker image in Gitlab CI and pushed the docker image to ECR. Then, I wrote the Dockerfile and reference from here. • I write a flow script and register to the Prefect cloud by using my local computer. Unfortunately, after running the pipeline, I got some errors: the python model was not found. As a result, I can't successfully import the module that I wrote in the project. And here's my flow script.

Failed to load and execute flow run: ModuleNotFoundError("No module named '/Users/xxx/'")

Thanks for helping. 🙇

Hiya! Quick question on best practices in Prefect 1. With a flow-of-flows, is there an ideal way to pass values down to child flows? In particular, secrets? Parameters seem fine for everything else.

Hi, I’m trying Prefect Orion. I am planning to use Amazon S3 bucket for Storage. I wrote the following code. However, I get an error. I have not created the Storage yet… Am I doing something wrong? I’d be glad to hear from you if anyone has any ideas!

s3_storage_block = S3StorageBlock(
    bucket="********",
    profile_name="default",
    region_name="ap-northeast-1",
)

async with get_client() as client:
    block_id = await client.create_block(
        name="********",
        block=s3_storage_block,
        block_spec_id=uuid.UUID("{12345678-1234-5678-1234-567891234567}")
    )

If I have 2 parameters, one as a list to be mapped, and one as a value that should be unmapped, how can I concat the mapped parameters onto the unmapped one?

So I have this deployment in Prefect 2.0:

DeploymentSpec(
    flow_location=str((FLOW_DIR / "salesforce.py").absolute()),
    flow_name="elt-salesforce",
    name="sf-production-elt-job",
    schedule=IntervalSchedule(interval=timedelta(hours=1)),
    tags=["pipeline"],
    flow_runner=DockerFlowRunner(image=f"{IMAGE_REPO}/{DBT_IMAGE}:{TAG}", stream_output=True)
)

And it took awhile to come to me as a requirement but I essentially have 2 steps. Step one requires docker image A to do some data pipeline stuff, step 2 needs my custom dbt docker image B to do some transform AFTER step 1. So these two dependent tasks constitute one flow with each step on independent docker images. A flow runner is configured at a deployment level but I dont see a way to configure it at the task or subflow level. Definitely a key req in current state. Please help!