prefect-community #prefect-community

Hello. I'm using Prefect 2.3.2. I would like to choose the name for each of my flow runs (not use the adjective-noun default) so that they are easier to search for in the Orion UI. I am processing images, and would like to name each flow run after its corresponding image. I am not currently using Agents or Deployments or Queues, just importing prefect into python scripts to help organize pipeline code and get each retries and convenient logging. I there a way that I can choose the flow run name when my flow is called? I have tried OrienClient.create_flow_run, but this seems to depend on using agents/deployments. Is there (or could there be) a simpler way I can do this directly when a new flow run starts?

✅ 1

CA Lee

09/11/2022, 8:27 AM

Hello, has anyone encounted an error while setting

is_schedule_active

property in deployment.yaml? Error:

Copy code

'deployments/func-deployment.yaml' did not conform to deployment spec: ValidationError(model='Deployment', errors=[{'loc': ('is_schedule_active',), 'msg': 'extra fields not permitted', 'type': 'value_error.extra'}])

This is using S3 block storage. If using LocalFileStorage, I believe this error does not occur. Prefect version: 2.3.2 Removing the field:

is_schedule_active: false

, the deployment can be applied successfully, but the user would need to login to the UI to manually disable the schedule. This is especially painful if triggered on each CI/CD run.

✅ 2

Ammar Chalifah

09/11/2022, 6:23 PM

Hi people! I'm trying to run Prefect locally inside a docker container and access the server UI through localhost. To do so, I set the Prefect Orion API Host to 0.0.0.0 (because the default host is localhost, and I can't access localhost through port forwarding from my local machine). However, after building the deployment and applying it, it seems the change is not reflected in the Prefect server UI. I know there is something wrong but I can't spot the mistake. Is there anyone that can help? Thank you 😄 UPDATE: The error in Orion UI no longer exists after I changed the API URL with this env var

Copy code

PREFECT_API_URL=http//0.0.0.0:4200/api

👍 1

✅ 1

Chris L.

09/11/2022, 6:51 PM

Hello Prefect community! Not sure if this is supposed to be a feature or a bug...but all our k8s deployed flows started failing the last two days with this error on the K8s Agent (v2.3.2): FIXED: Specified`command` arg explicitly as

[ "python", "-m", "prefect.engine" ]

for

KubernetesJob

Copy code

18:49:52.809 | INFO    | prefect.agent - Submitting flow run 'faa2ebac-ee94-4384-9a4f-16ad28d6fa7e'
....  
  File "/usr/local/lib/python3.9/site-packages/prefect/blocks/core.py", line 171, in __init__
    super().__init__(*args, **kwargs)
  File "pydantic/main.py", line 342, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for KubernetesJob
command
  none is not an allowed value (type=type_error.none.not_allowed)

✅ 1

James Constable

09/12/2022, 2:53 AM

Hi All. I am experimenting with Prefect 2.0 deployments. I am using azure storage as my deployment storage. The simplest example project I can make takes ~25mins to build the deployment. This is way too slow for any kind of CI/CD situation. Is this normal? Is it my connection? Am I doing something wrong?

Ahmed Ezzat

09/12/2022, 4:07 AM

I'm unable to access prefect 2.0 cloud using Github login or reset my own password

Naila Chennit

09/12/2022, 8:41 AM

Hello Prefect community, I would like to know why can't we set a task timeout anymore with prefect 2 please ?

✅ 1

Eli Treuherz

09/12/2022, 9:01 AM

Is there a Prefect 2 equivalent to doing

raise SKIP("not applicable")

in order to skip a task? If I just

return Cancelled(message="not applicable")

instead of returning my task’s actual data, will that have the same effect?

✅ 1

Dennis Hinnenkamp

09/12/2022, 12:10 PM

Hi all, I am trying to read the configuration from a

snowflake_connector_block

for a dbt shell task, but I get the following error when I use the created `dbt_cli_profile`:

Copy code

14:07:11.533 | ERROR   | Task run 'trigger_dbt_cli_command-321ca940-0' - Encountered exception during execution:
Traceback (most recent call last):
  File "/Users/d.hinnenkamp/Desktop/test-prefect/.venv/lib/python3.8/site-packages/prefect/engine.py", line 1185, in orchestrate_task_run
    result = await task.fn(*args, **kwargs)
  File "/Users/d.hinnenkamp/Desktop/test-prefect/.venv/lib/python3.8/site-packages/prefect_dbt/cli/commands.py", line 136, in trigger_dbt_cli_command
    profile = dbt_cli_profile.get_profile()
  File "/Users/d.hinnenkamp/Desktop/test-prefect/.venv/lib/python3.8/site-packages/prefect_dbt/cli/credentials.py", line 107, in get_profile
    "outputs": {self.target: self.target_configs.get_configs()},
  File "/Users/d.hinnenkamp/Desktop/test-prefect/.venv/lib/python3.8/site-packages/prefect_dbt/cli/configs/snowflake.py", line 78, in get_configs
    configs_json = super().get_configs()
  File "/Users/d.hinnenkamp/Desktop/test-prefect/.venv/lib/python3.8/site-packages/prefect_dbt/cli/configs/base.py", line 53, in get_configs
    return self._populate_configs_json({}, self.dict())
  File "/Users/d.hinnenkamp/Desktop/test-prefect/.venv/lib/python3.8/site-packages/prefect_dbt/cli/configs/base.py", line 34, in _populate_configs_json
    configs_json = self._populate_configs_json(configs_json, value)
  File "/Users/d.hinnenkamp/Desktop/test-prefect/.venv/lib/python3.8/site-packages/prefect_dbt/cli/configs/base.py", line 37, in _populate_configs_json
    raise ValueError(
ValueError: The keyword, _is_anonymous, has already been provided in TargetConfigs; remove duplicated keywords to continue

My current code looks like this:

snowflake_connector_block = await SnowflakeConnector.load("wld-snowflake-connection")

connector = snowflake_connector_block

target_configs = SnowflakeTargetConfigs(

connector=connector

dbt_cli_profile = DbtCliProfile(

name="profile_name",

target="dev",

target_configs=target_configs

debug = await trigger_dbt_cli_command.submit(

"dbt debug",

profiles_dir='profile',

project_dir='wld-dbt-warehouse/dbt-project',

overwrite_profiles=True,

dbt_cli_profile=dbt_cli_profile,

return_state=True

✅ 1

Vadym Dytyniak

09/12/2022, 12:22 PM

Hello. I need to pass imagePullSecrets in KubernetesJob. I use customizations arg:

Copy code

customizations=[
    {
        "op": "add",
        "path": "/spec/imagePullSecrets",
        "value": [{'name': 'dockerhub'}],
    },
],

What is wrong with passed value? I still get an access denied error.

✅ 1

Niels Prins

09/12/2022, 1:14 PM

is it corrct that the Orion gui does not show

filters

when creating a WorkQueue? <EDIT> Yes this is correct, see thread

✅ 1

Prasanth Kothuri

09/12/2022, 2:03 PM

Hi all, is the following schedule allowed? it allows me to register the flow, so assuming thats ok

✅ 1

Prasanth Kothuri

09/12/2022, 2:03 PM

Copy code

schedule = Schedule(clocks=[CronClock("0 07,12,14 * * 1-6"),CronClock("0 10 * * 0")])

Prasanth Kothuri

09/12/2022, 2:05 PM

Kelvin Garcia

09/12/2022, 2:51 PM

Hi all, I have some flows that fail but when I try to restart it fails because of a different reason, it seems like it doesn't remember the output of a successful task, what should I do to persist data on restart?

👀 1

✅ 1

Josh Cowles

09/12/2022, 3:19 PM

Is there a publicly available roadmap somewhere? Or is it internal to the prefect team only? From this community post it says that vertex in prefect 2 is on the roadmap, but no further info on when it might be coming

✅ 1

Sam Garvis

09/12/2022, 5:36 PM

https://github.com/PrefectHQ/prefect/issues/6779 We are on Prefect 2.3.2 using GCS storage and k8s helm agents. I just commented on somebody else's issue on GitHub, but just to reiterate, this error is the only thing that is preventing our team from truly upgrading to Prefect 2.0. It has been going on since we started transitioning and it won't go away.

ValueError: Path /root/.prefect/storage/72ed22dddb494741872cffe99e581ee9 does not exist.

It's an incredibly vague error that is hard to debug. However, the flows all complete successfully, they just show as crashed. But if we were to transition to 2.0, we need reliable error monitoring, and we can't do that with 90% of flows crashing for no reason.

✅ 1

Nathaniel Russell

09/12/2022, 6:42 PM

how does one run a deployment in python? I have found only how to make a deployment or run a flow separate from a deployment.

✅ 1

Kelvin Garcia

09/12/2022, 7:20 PM

Hi all, I see in the docs (https://docs-v1.prefect.io/core/concepts/persistence.html#input-caching) that prefect does automatic input caching, so in this case I have a task that returns a dict

Copy code

{
        "jobId": reportId,
        "s3Location": s3Location,
        "validationSettings": validation_settings,
    }

but when the flow fails and I need to restart a task that uses that dict as a dependency I get this error

Copy code

reportId = settings["jobId"]
TypeError: 'NoneType' object is not subscriptable

why the second time I run that task the input is

None

, shouldn't it cache the input for restarts?

✅ 1

Leon Kozlowski

09/12/2022, 7:55 PM

Hi all, I’m try to deploy an orion agent - I’m getting the following error:

Copy code

prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<https://api.prefect.cloud/work_queues/>'

Not sure if I have my api url wrong or some other configuration incorrect

✅ 1

jack

09/12/2022, 9:15 PM

Say I want to start a second prefect agent (running prefect 2.0) so we can start developing some flows to run on prefect 2.0, but want all existing flows to continue running via the prefect 1.x agent. Is there a way to do that without needing to change the labels on our existing flows?

✅ 1

Chris Gunderson

09/12/2022, 10:02 PM

Hi Team - I tried to use this flow to flow in Prefect 1.0. https://docs-v1.prefect.io/core/idioms/flow-to-flow.html . I ended up getting this error: [2022-09-12 165540-0500] ERROR - prefect.sod-loader-parent-flow | Unexpected error occured in FlowRunner: ValueError('Cycle found; flows must be acyclic!')

✅ 2

Niels Prins

09/13/2022, 8:07 AM

Small question (or 2) about prefect 2.0 Trying to build a CICD solution with prefect 2.0. There are a few things I’m still struggling with; • What is the advantage of storing my flow on another source than Git? Using GIT gives me a single source of truth inline with the standards in my organization. • Where does my flow store its cache while running? I want to be able to restart a task when my flow crashed, while running on K8s. Am I correct when assuming that I need to configure External Storage with put-directory and write-path for this? • The MINIO example on the website is trying to connect to an AWS endpoint, probably due to s3fs. Can I just follow the example on https://docs.prefect.io/concepts/blocks/, create my own PyMinio based block and use that? Many thanks in advance

✅ 1

Afaque Ahmad

09/13/2022, 8:08 AM

Hi Prefect Team, We're planning to use Prefect for creating our data flows, If I use the open source one, Do I need to host Prefect on my own cloud provider?

✅ 1

Andreas Nord

09/13/2022, 8:26 AM

Hi! My flow seems to have finished by still show up as running in Prefect Cloud. This is the last logged message in the agent log (in the execution environment)

Copy code

08:12:06.349 | INFO    | prefect.infrastructure.docker-container - Docker container 'practical-bear' has status 'exited'

But the log on prefect cloud is stuck on Executing of the last task of the flow.

✅ 1

Saurabh Indoria

09/13/2022, 9:01 AM

Prefect adjusts the run time at the end of run. Any reason for this? Feels a bit sketchy..

Screen Recording 2022-09-13 at 2.06.41 PM.mov

✅ 1

Dennis Hinnenkamp

09/13/2022, 9:20 AM

I'm trying to use a shell_run_command to run some soda.io scans again my dbt models with prefect. The execution is correct but if the test itself fails, the tasks returns an exception and so my flow is also failing. Is there anything I can do to prevent the task from terminating in an incorrect state?

Copy code

09:19:54.856 | INFO    | prefect.agent - Submitting flow run '9e981ee3-5d70-43b8-992a-6bd5c433d24c'
09:19:55.054 | INFO    | prefect.infrastructure.process - Opening process 'blond-rabbit'...
09:19:55.130 | INFO    | prefect.agent - Completed submission of flow run '9e981ee3-5d70-43b8-992a-6bd5c433d24c'
09:19:57.018 | INFO    | Flow run 'blond-rabbit' - Created task run 'shell_run_command-7398b6ba-0' for task 'shell_run_command'
09:19:57.019 | INFO    | Flow run 'blond-rabbit' - Executing 'shell_run_command-7398b6ba-0' immediately...
09:19:57.251 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Soda Core 3.0.6

09:19:58.517 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to acquire lock 281472472552592 on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:19:58.518 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Lock 281472472552592 acquired on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:19:58.518 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to acquire lock 281472472682848 on /home/vscode/.cache/snowflake/ocsp_cache.lock
Timeout on acquiring lock 281472472682848 on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:19:58.519 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to release lock 281472472552592 on /home/vscode/.cache/snowflake/ocsp_cache.lock
Lock 281472472552592 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.123 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Scan summary:
1/1 check FAILED: 
    v_stg_transactions in we_liberate_data
      row_count between 10 and 1000 [FAILED]

09:20:00.125 | INFO    | Task run 'shell_run_command-7398b6ba-0' -         check_value: 6209
Oops! 1 failures. 0 warnings. 0 errors. 0 pass.

09:20:00.125 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Sending results to Soda Cloud

09:20:00.484 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Soda Cloud Trace: 3791758754192003542

09:20:00.878 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to acquire lock 281472486055312 on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.878 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Lock 281472486055312 acquired on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.879 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to release lock 281472486055312 on /home/vscode/.cache/snowflake/ocsp_cache.lock
Lock 281472486055312 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.930 | ERROR   | Task run 'shell_run_command-7398b6ba-0' - Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 1185, in orchestrate_task_run
    result = await task.fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect_shell/commands.py", line 90, in shell_run_command
    raise RuntimeError(msg)
RuntimeError: Command failed with exit code 2:
Lock 281472486055312 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.952 | ERROR   | Task run 'shell_run_command-7398b6ba-0' - Finished in state Failed('Task run encountered an exception.')
09:20:00.952 | ERROR   | Flow run 'blond-rabbit' - Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 588, in orchestrate_flow_run
    result = await flow_call()
  File "flows/test/test_soda_transactions.py", line 11, in test_transactions
    command2 = await shell_run_command(
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 863, in get_task_call_return_value
    return await future._result()
  File "/usr/local/lib/python3.8/site-packages/prefect/futures.py", line 236, in _result
    return final_state.result(raise_on_failure=raise_on_failure)
  File "/usr/local/lib/python3.8/site-packages/prefect/orion/schemas/states.py", line 143, in result
    raise data
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 1185, in orchestrate_task_run
    result = await task.fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect_shell/commands.py", line 90, in shell_run_command
    raise RuntimeError(msg)
RuntimeError: Command failed with exit code 2:
Lock 281472486055312 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.966 | ERROR   | Flow run 'blond-rabbit' - Finished in state Failed('Flow run encountered an exception.')
09:20:01.240 | INFO    | prefect.infrastructure.process - Process 'blond-rabbit' exited cleanly.

Amol Shirke

09/13/2022, 9:33 AM

Hello, I need suggestion: In deployment with prefect.flow_runner import subprocess deprecated. How can I set virtualvenv as like in SubprocessFlowRunner(virtualenv=path_to_interpreter) What is possible way I can gracefully upgrade this line of code?

✅ 1

Gintautas Jankus

09/13/2022, 11:13 AM

Hi, I'm having some trouble to understand how to better restart subflow tasks, thought to ask for some help :) Here is situation: I have flow which contains subflow tasks. Subflow D has upstream of two subflows A and B. They all have wait_for_flow tasks. If for example subflow A fails, I want in prefect UI restart all subflow D downstream from where it failed. But now if I try to restart in UI it will start to restart subflow A wait_for_flow task without running subflow A task itself and it will be stuck waiting. So my question how to better restart subflow D downstream(including subflow A) without restarting subflow C and B?

✅ 1

Sander

09/13/2022, 11:59 AM

Hi, trying out some simple map reduce stuff on latest prefect on a dask cluster. Should I use task.submit() or can I also use task.map()?

✅ 1