Hey there, I'm looking for advice or examples of how to dial with secrets(passwords, keys, etc) in prefect 2.0. Can anyone give me a hint where to look for?

Hey there! I am starting to go crazy trying to get the K8s setup to work 😅 I switched from self-hosted API to Cloud 2.0 since I just couldn’t make persistence with K8s PVCs work. Unfortunately, now I get the following error when I start my agent in K8s:

prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<https://api-beta.prefect.io/api/accounts/XXXXXXX/workspaces/XXXXXXXX/work_queues/name/k8s>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'loc': ['path', 'account_id'], 'msg': 'value is not a valid uuid', 'type': 'type_error.uuid'}, {'loc': ['path', 'workspace_id'], 'msg': 'value is not a valid uuid', 'type': 'type_error.uuid'}, {'loc': ['path', 'account_id'], 'msg': 'value is not a valid uuid', 'type': 'type_error.uuid'}, {'loc': ['path', 'workspace_id'], 'msg': 'value is not a valid uuid', 'type': 'type_error.uuid'}, {'loc': ['path', 'workspace_id'], 'msg': 'value is not a valid uuid', 'type': 'type_error.uuid'}], 'request_body': None}
For more information check: <https://httpstatuses.com/422>
An exception occurred.

That error is just so weird that I have no ideas where to look. Any help would be greatly appreciated :) The agent is in a crashloop because of this. I have followed https://discourse.prefect.io/t/how-to-deploy-a-prefect-2-0-agent-to-a-local-kubernetes-cluster-and-connect-it-to-cloud-2-0-backend/979 exactly. The only difference is that I updated to

prefecthq/prefect:2.0b7-python3.9

because the b6 image gave me version incompatibility errors when using the CLI tool.

what is your PREFECT_API_URL, you sure it ends with a uuid

Hey, can I run my flows as a vertex ai custom job in prefect 2.0?

Is it possible to have a Kubernetes Agent run a Flow in a different namespace? Can this be set via the job template? (In 1.0)

Hey everyone! 👋 We're doing a prototype to test out prefect and plan to go with cloud version for staging and production usage, but need to do local development story too and we want to have everything managed by docker-compose locally (and potentially for CI). Unless I'm missing something, https://hub.docker.com/r/prefecthq/prefect assumes you're going to install prefect on host machine and then

prefect

would start it's containers? Is that correct? If so, what's the best path to have everything as a docker-compose file that developers can spin up on request? Ar there any samples for that use case?

Hello - We have a NodeJS script that we wish to execute as a task, the entire code (and all the NodeJS) dependencies are located on a docker image on ECR, we’re trying out an experimental alternative to running a Kubernetes namespaced job, which would be - to run the image itself as the image of the entire flow (and have the command shell to run the NodeJS script as one of the tasks of the flow) the docker image uses a docker base image of

node:12-alpine

(which doesn’t seem to have

pip

and possibly not

python

either) i tried to add:

RUN apk update
RUN apk add py-pip
RUN pip install prefect[github,aws,kubernetes,snowflake]

first steps finished fine, but the prefect installation seems to be taking forever (over 25 minutes already) and also has a lot of weird warning messages like:

Collecting snowflake-connector-python>=1.8.2                                                                                                                     
  Downloading snowflake_connector_python-1.8.7-py2.py3-none-any.whl (168 kB)                                                                                     
  Downloading snowflake_connector_python-1.8.6-py2.py3-none-any.whl (161 kB)                                                                                     
  Downloading snowflake_connector_python-1.8.5-py2.py3-none-any.whl (159 kB)                                                                                     
  Downloading snowflake_connector_python-1.8.4-py2.py3-none-any.whl (161 kB)                                                                                     
  Downloading snowflake_connector_python-1.8.3-py2.py3-none-any.whl (158 kB)                                                                                     
  Downloading snowflake_connector_python-1.8.2-py2.py3-none-any.whl (157 kB)                                                                                     
INFO: pip is looking at multiple versions of six to determine which version is compatible with other requirements. This could take a while.

or :

INFO: This is taking longer than usual. You might need to provide the dependency resolver with stricter constraints to reduce runtime. If you want to abort this 
run, you can press Ctrl + C to do so. To improve how pip performs, tell us what happened here: <https://pip.pypa.io/surveys/backtracking>

any ideas?

Does anyone know of any prefect tasks for singer taps ?

Quick note that there is a typo in the documentation - concurrency limit command examples should have dashes ("concurrency-limit") instead of underscores ("concurrency_limit")

Hi! I am working on setting up a simple CI/CD process that registers new flows each time a push is made to main. My folder structure is inspired by the gitlab data team so at the root of my project i have a folder orchestrate. Within that folder i have a folder tasks - within this there are subdirectories that contain custom modules that store tasks we use throughout our flows. While registering flows via the GitHub action, the Prefect CI is complaining about not being able to find a module in the tasks folder. When I run this locally, I do not get any issues. I am pretty sure it is related to my PYTHONPATH setup within the GitHub flow but I have not had any luck troubleshooting. Has anyone out there run into a similar issue?

A couple of virtual events happening this week! https://prefect-community.slack.com/archives/C036FRC4KMW/p1657042607147229

Hello everyone, I have a task which keeps failing because the

.map()

is trying to run the task more times than necessary and the 4th time, the index is a string. .

product_categories

is a list containing 3 elements,

flow_config

is a dictionary and

product_category_variables

is also a dictionary. When the task runs in prefect cloud there are three successful task runs indexed 0, 1, 2 but then it tries to do another run with the index as a string and it fails saying

Task 'set_dynamic_config_settings['dataset_bucket_path']': Starting task run...

TypeError: list indices must be integers or slices, not str

The task

set_dynamic_config_settings

looks like this within the flow definition

with Flow(name="flow-name") as: 

   product_category_variables = get_run_variables(
        is_zero_nyp=is_zero_nyp_param,
        bucket_base=flow_config["bucket_base"],
        ltv_product_categories=product_categories,
        return_type="vars",
       )

   final_config = set_dynamic_config_settings.map(
        cfg=unmapped(flow_config),
        product_category_variables=unmapped(product_category_variables),
        product_category=product_categories,
        upstream_tasks=[unmapped(product_category_variables)],
       )

Can someone point me to the right direction please? I'm using Prefect 1.2.2 I have a flow (flow_A) that makes an API call, then flow_B will Transform some of that data, and I want these 2 flow to remain independent while running flow_B as close to flow_A as possible. What is the correct way to ensure that flow_B only starts after flow_A is finished? (I don't want any of these flows to trigger each other, I just want flow_B to wait on flow_A). I tried using

wait_for_flow_run

, but I get asked for a

flow_run_id

, while I don't have a

flow_run_id

, I only have a

flow_id

for the parent flow flow_A

will an agent query for flows across multiple tenants?

Hi, how do I enable debug logging for boto+prefect in my flow when running locally with

flow.run()

? I’ve tried adding this code from the boto3 docs to my flow:

import boto3
boto3.set_stream_logger()  # sets logging.DEBUG

However, ^ that code hasn’t changed the

flow.run()

output. For background, I’m trying to use localstack as an AWS stand-in for local pipeline development. The

S3Download

task is giving me errors and I want to debug the boto3 connection to find out if the S3 service URL, bucket, and key are correct.

Is it possible to yield results from a task into another task? I have a task that pulls data from an API in pages, and rather than accumulating ALL of the data into a list before passing it to the next task, it would be much more memory efficient to yield each page at a time.

Hi! In Prefect 2.0, when creating a flow_run for a deployment using API's

prefect.client.create_flow_run_from_deployment()

is it possible to set the name for the flow_run? I know that there is such an option when calling

create_flow_run(name="My flow run name")

that takes as input a flow_model, what about when creating a run from a deployment though?

Hi All, We have installed the prefect server helm version 2022 and dask gateway 2022.4.0 on the google kubernetes engine cluster and registered the flow , logged into the pefect UI and triggered the flow. Then, we encountered the below error Prefect is using the trafiek dask gateway to spin up dask workers and dask scheduler as part of job run. The prefect apollo (Load balancer external IP) is connecting to dask gateway trafiek proxy. Please correct me if i am wrong. prefect UI is the backend service for the Google kubernetes ingress component( IP ) which is leading to the prefect apollo service. We suspect the issue could be source IP ranges in firewall rule which accesses the dask gateway trafiek LB service IP. Please suggest what can be the source IP range in the above firewall rule as per the architecture of the prefect. Can you please let us know which component of the prefect server will communicate with dask trafiek proxy service to deploy and run the flows so that we can consider it for the source IP. Please let us know if any inputs are required. Attaching the error screenshot below.

Hello, maybe this is kind of a silly question but - assuming we have a big CSV that we then split into multiple CSVs (as a task), and then, we wanna run a set of tasks for each of these “mini-CSVs” (either in parallel or sequentially, probably sequentially though) — we’re trying to understand what’s the beach approach here (continued in thread…)

Hello. Is it possible to use

prefect.context.parameters

when setting a

Result

location?

Hi, we just witnessed an error when running a prefect task locally on a windows 11 machine. This error is not popping up on mac. Do you know how to fix that?

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: json_each
[SQL: INSERT INTO flow_run_notification_queue (flow_run_notification_policy_id, flow_run_state_id) SELECT flow_run_notification_policy.id, CAST(? AS CHAR(36)) AS anon_1
FROM flow_run_notification_policy
WHERE flow_run_notification_policy.is_active IS 1 AND (flow_run_notification_policy.state_names = ? OR EXISTS (SELECT 1
FROM json_each(flow_run_notification_policy.state_names) AS json_each
WHERE json_each.value IN (?))) AND (flow_run_notification_policy.tags = ? OR EXISTS (SELECT 1
FROM json_each(flow_run_notification_policy.tags) AS json_each
WHERE json_each.value IN (SELECT 1 FROM (SELECT 1) WHERE 1!=1)))]
[parameters: ('8d01934d-c9c8-4f23-a3af-ec03ee1586a0', '[]', 'Pending', '[]')]
(Background on this error at: <https://sqlalche.me/e/14/e3q8>)

Hi i am unable to start my Prefect Server, getting below error:

graphql_1   | ValueError: Uniqueness violation.
graphql_1   |
graphql_1   | The above exception was the direct cause of the following exception:
graphql_1   |
graphql_1   | Traceback (most recent call last):
graphql_1   |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 674, in await_completed
graphql_1   |     return await completed
graphql_1   |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 659, in await_result
graphql_1   |     return_type, field_nodes, info, path, await result
graphql_1   |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 733, in complete_value
graphql_1   |     raise result
graphql_1   |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 628, in await_result
graphql_1   |     return await result
graphql_1   |   File "/prefect-server/src/prefect_server/graphql/extensions.py", line 52, in resolve
graphql_1   |     result = await result
graphql_1   |   File "/prefect-server/src/prefect_server/graphql/tenants.py", line 14, in resolve_create_tenant
graphql_1   |     "id": await api.tenants.create_tenant(name=input["name"], slug=input["slug"])
graphql_1   |   File "/prefect-server/src/prefect_server/api/tenants.py", line 29, in create_tenant
graphql_1   |     tenant_id = await models.Tenant(name=name, slug=slug).insert()
graphql_1   |   File "/prefect-server/src/prefect_server/database/orm.py", line 222, in insert
graphql_1   |     insert_mutation_name=self.__root_fields__.get("insert"),
graphql_1   |   File "/prefect-server/src/prefect_server/database/hasura.py", line 237, in insert
graphql_1   |     result = await self.execute_mutations_in_transaction(mutations=[graphql])
graphql_1   |   File "/prefect-server/src/prefect_server/database/hasura.py", line 165, in execute_mutations_in_transaction
graphql_1   |     as_box=as_box,
graphql_1   |   File "/prefect-server/src/prefect_server/database/hasura.py", line 89, in execute
graphql_1   |     raise ValueError("Uniqueness violation.")
graphql_1   | graphql.error.graphql_error.GraphQLError: Uniqueness violation.
graphql_1   |
graphql_1   | GraphQL request:2:3
graphql_1   | 1 | mutation ($input: create_tenant_input!) {
graphql_1   | 2 |   create_tenant(input: $input) {
graphql_1   |   |   ^
graphql_1   | 3 |     id

Hi. Is it possible to specify disk space for KubernetesRun run config?

Hello How can I rename a task output ? Is it possible to use with it a task nout=3 ?

Hey guys, I want to pull down a

.csv

file from a google cloud bucket. It seems like this task is the correct way to go? An alternative I can see is just implementing a python callable that uses the python library for google cloud storage. I'd like some feedback if I'm approaching this problem correctly?

Hi Team. In prefect 2.0, is there a way to visualize the flows like how we could do in prefect 1.0 with flow.visualize()?

Hello. I’m trying to figure out why my configured results don’t seem to be respected/used for caching between flow runs. I have a task decorated to use checkpointing. On each run of the flow, I see the task complete with a

Success

state, and the results file has been overwritten. If I’m understanding correctly this should be completing with

Cached

. Any suggestions?

Is the "sql_server" optional dependency (for writing to MSSQL) available in Prefect 2.0? If not, what are alternatives, do I just stick with 1.0 until things like that get implemented into 2.0, or is there a new way to go about connecting to MSSQL?

is it possible to configure the ecs agent so that it sets the flow run container command to include

-l

(I want a login shell). Seems like currently its passing the command like so

overrides
    containerOverrides
        command": [
            "/bin/sh",
            "-c",
            "prefect execute flow-run"
        ]

Also, possible or not, is this command stable / is it safe to customize this?

In Prefect 2, is the default storage option set per-workspace or the same across all workspaces?