https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I have a Prefect process worker on a...
# ask-marvin
s
@Marvin I have a Prefect process worker on a 
DaskTaskRunner
running flow run via deployments. Is it okay to run a task outside the decorated flow function, something like:
Copy code
@task
def validate_env_vars(var_names: list[str]) -> dict[str, Optional[str]]:
    """Validate that all required environment variables are set."""
    missing_vars = [var for var in var_names if not os.getenv(var)]
    if missing_vars:
        raise ValueError(f"The following environment variables are not set: {', '.join(missing_vars)}")
    return {var: os.getenv(var) for var in var_names}

# run this task
required_vars = [
        "ENV_STATE",
        "DB_USER",
        "PREFECT_PYTHON",
       ]
vars = validate_env_vars(required_vars)

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,  # type: ignore[arg-type]
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS}
    ),
    log_prints=True,
    persist_result=False,
)
def sentinel2_satproc(
    import_file: str,
    working_dir: Path = Path.cwd()
) -> list[str]:
    ....
m
✅ thought for 98.1 seconds
Yes — it will run, but probably not in the way you want. If you do this:
Copy code
python
@task
def validate_env_vars(...):
    ...

vars = validate_env_vars(required_vars)
outside any 
@flow
, Prefect just executes that task directly and synchronously in the current Python process. It is not submitted to your flow’s 
DaskTaskRunner
, and it is not part of the deployed flow run graph. So in your case: - it runs before the flow starts - it runs in the worker process that imports/executes your code - it does not use Dask - it does not appear as a task inside the flow run in the UI - if it raises, the flow may never start at all That means it’s valid Python/Prefect usage, but usually not the best pattern if you want this check to be tracked as part of the deployment run. Recommended options 1. If this is just startup validation, make it a plain function This is the cleanest option if you only want to fail fast before the flow runs.
Copy code
python
def validate_env_vars(var_names: list[str]) -> dict[str, str]:
    missing_vars = [var for var in var_names if not os.getenv(var)]
    if missing_vars:
        raise ValueError(
            f"The following environment variables are not set: {', '.join(missing_vars)}"
        )
    return {var: os.environ[var] for var in var_names}

required_vars = [
    "ENV_STATE",
    "DB_USER",
    "PREFECT_PYTHON",
]

vars = validate_env_vars(required_vars)
2. If you want it visible/orchestrated by Prefect, call it inside the flow Then it becomes an actual Prefect task run and can use the task runner when submitted.
Copy code
python
@task
def validate_env_vars(var_names: list[str]) -> dict[str, str]:
    missing_vars = [var for var in var_names if not os.getenv(var)]
    if missing_vars:
        raise ValueError(
            f"The following environment variables are not set: {', '.join(missing_vars)}"
        )
    return {var: os.environ[var] for var in var_names}

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=PBSCluster,
        cluster_kwargs=pbs_cluster_kwargs,
        adapt_kwargs={"maximum": MAX_WORKERS},
    ),
    log_prints=True,
    persist_result=False,
)
def sentinel2_satproc(import_file: str, working_dir: Path = Path.cwd()) -> list[str]:
    required_vars = [
        "ENV_STATE",
        "DB_USER",
        "PREFECT_PYTHON",
    ]
    env_vars = validate_env_vars(required_vars)
    ...
Or, if you want it explicitly submitted through Dask:
Copy code
python
env_vars_future = validate_env_vars.submit(required_vars)
env_vars = env_vars_future.result()
Which should you choose? - Use a plain function if this is just configuration/bootstrap validation. - Use a Prefect task inside the flow if you want: - visibility in the UI - retries - logging/state tracking - dependency orchestration One important caveat If that top-level 
validate_env_vars(required_vars)
runs at import time, it can cause deployment/runtime issues because the module may be imported in contexts where those env vars are not yet set. That’s why top-level side effects are usually risky in deployed flows. Best practice - avoid executing work at module import time - do env validation either: - in a plain helper called near flow entry, or - as the first task inside the flow Docs / references: - Prefect tasks - Prefect flows - Workers and execution If you want, I can also show the safest pattern for validating env vars in a deployed flow using your 
DaskTaskRunner
+ 
PBSCluster
setup.
s
@Marvin I have this simple deployment yaml:
Copy code
deployments:
- name: s2_satproc_dev
  ...
  work_pool:
    name: my_pool
    work_queue_name: null
    job_variables:
      env:
          PIPELINE_SHARED_DIR: '"{{ $working_dir }}"
How can access the value in the 
working_dir
field of 
my_pool
work pool?
m
✅ thought for 133.9 seconds
You generally cannot access a work pool’s existing 
working_dir
field from deployment YAML as 
{{ $working_dir }}
. A few important details: - 
{{ $... }}
means OS environment variable at deploy time, not a work pool field - work pool settings are not exposed as deploy-time template variables - if you want to reuse 
working_dir
, you should set it explicitly in 
job_variables
and reference that value So this:
Copy code
yaml
env:
  PIPELINE_SHARED_DIR: "{{ $working_dir }}"
won’t do what you want unless 
working_dir
is literally an environment variable in the shell where you run 
prefect deploy
. What to do instead Set 
working_dir
as a deployment job variable, then reference it from `env`:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      work_queue_name: null
      job_variables:
        working_dir: /path/to/shared/dir
        env:
          PIPELINE_SHARED_DIR: "{{ working_dir }}"
This is the safest pattern because both values come from the same 
job_variables
context. Important caveat Whether 
working_dir
is a valid job variable depends on your worker/work pool type. For a process work pool, 
working_dir
is valid. If you’re using another worker type, the available job variables may differ. If you want to inspect/edit the work pool’s base job template, see: - Work pools - Deployments If your goal is “use the pool’s configured working_dir automatically” Prefect doesn’t provide a built-in template like “inherit the pool’s current working_dir into env”. You need to either: - duplicate the value in deployment 
job_variables
, or - store it in an environment variable / Prefect variable and reference that in both places For example with an env var:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ $PIPELINE_SHARED_DIR }}"
        env:
          PIPELINE_SHARED_DIR: "{{ $PIPELINE_SHARED_DIR }}"
Or with a Prefect variable:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ prefect.variables.pipeline_shared_dir }}"
        env:
          PIPELINE_SHARED_DIR: "{{ prefect.variables.pipeline_shared_dir }}"
If you want, I can help you rewrite your full 
prefect.yaml
for a process worker + Dask setup.
s
@Marvin Using client and self-hosted server 3.6, this didn't not work:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      work_queue_name: null
      job_variables:
        working_dir: /path/to/shared/dir
        env:
          PIPELINE_SHARED_DIR: "{{ working_dir }}"
Getting this warning:
Copy code
❯ prefect deploy --prefect-file deployments/my_deployment.yaml -n s2_satproc_dev

? Would you like to configure schedules for this deployment? [y/n] (y): n
10:42:51.689 | WARNING | prefect.utilities.templating - Value for placeholder 'working_dir' not found in provided values. Please ensure that the placeholder is spelled correctly and that the corresponding value is provided.
m
✅ thought for 100.1 seconds
Yep — that warning means Prefect 3.6 is not resolving sibling references inside 
job_variables
. So this pattern does not work in 3.6:
Copy code
yaml
job_variables:
  working_dir: /path/to/shared/dir
  env:
    PIPELINE_SHARED_DIR: "{{ working_dir }}"
Because during 
prefect deploy
, 
{{ working_dir }}
is not looked up from neighboring keys in the same YAML mapping. Prefect only resolves things like: - 
{{ $ENV_VAR }}
- 
{{ prefect.variables.some_name }}
- 
{{ prefect.blocks... }}
- step outputs like 
{{ some-step.output }}
It does not support “reference another key in `job_variables`”. Working alternatives 1. Use the same source twice If the path is static, just duplicate it:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: /path/to/shared/dir
        env:
          PIPELINE_SHARED_DIR: /path/to/shared/dir
2. Use an environment variable at deploy time This is usually the cleanest:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ $PIPELINE_SHARED_DIR }}"
        env:
          PIPELINE_SHARED_DIR: "{{ $PIPELINE_SHARED_DIR }}"
Then run:
Copy code
bash
export PIPELINE_SHARED_DIR=/path/to/shared/dir
prefect deploy --prefect-file deployments/my_deployment.yaml -n s2_satproc_dev
3. Use a Prefect variable If you want the value managed in Prefect:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ prefect.variables.pipeline_shared_dir }}"
        env:
          PIPELINE_SHARED_DIR: "{{ prefect.variables.pipeline_shared_dir }}"
What’s happening under the hood In Prefect 3.6, deployment templating happens at deploy time, and 
job_variables
is not used as a self-referential template context. So 
working_dir
is just another field, not a variable available to 
env
. So the short answer is: You cannot derive 
env.PIPELINE_SHARED_DIR
from 
job_variables.working_dir
with 
{{ working_dir }}
in Prefect 3.6. You need to: - duplicate the value, or - source both from the same external variable If you want, I can show you the best pattern for making this work cleanly with different dev/prod deployment YAMLs without repeating paths everywhere.
s
@Marvin Sure:
Copy code
f you want, I can show you the best pattern for making this work cleanly with different dev/prod deployment YAMLs without repeating paths everywhere.
m
✅ thought for 8.0 seconds
A clean pattern is to make one external source of truth for the path, then reference it in both 
working_dir
and 
env
. For dev/prod, the two nicest options are: Option 1: Prefect Variables Good if you want config managed in Prefect. Set variables per environment/workspace, for example: - 
pipeline_shared_dir_dev
- 
pipeline_shared_dir_prod
Then in YAML:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ prefect.variables.pipeline_shared_dir_dev }}"
        env:
          PIPELINE_SHARED_DIR: "{{ prefect.variables.pipeline_shared_dir_dev }}"

  - name: s2_satproc_prod
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ prefect.variables.pipeline_shared_dir_prod }}"
        env:
          PIPELINE_SHARED_DIR: "{{ prefect.variables.pipeline_shared_dir_prod }}"
This avoids hardcoding actual paths in the file. Option 2: Deploy-time environment variables Good if you deploy from CI/CD or different shells.
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ $PIPELINE_SHARED_DIR_DEV }}"
        env:
          PIPELINE_SHARED_DIR: "{{ $PIPELINE_SHARED_DIR_DEV }}"

  - name: s2_satproc_prod
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ $PIPELINE_SHARED_DIR_PROD }}"
        env:
          PIPELINE_SHARED_DIR: "{{ $PIPELINE_SHARED_DIR_PROD }}"
Then set:
Copy code
bash
export PIPELINE_SHARED_DIR_DEV=/mnt/dev/shared
export PIPELINE_SHARED_DIR_PROD=/mnt/prod/shared
My recommended pattern If you have separate dev/prod deployments in one file, I’d usually use Prefect Variables because they’re explicit and live with your Prefect environment config. Example:
Copy code
yaml
deployments:
  - name: s2_satproc_dev
    entrypoint: sentinel2_satproc.py:sentinel2_satproc
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ prefect.variables.s2_shared_dir_dev }}"
        env:
          PIPELINE_SHARED_DIR: "{{ prefect.variables.s2_shared_dir_dev }}"
          ENV_STATE: dev

  - name: s2_satproc_prod
    entrypoint: sentinel2_satproc.py:sentinel2_satproc
    ...
    work_pool:
      name: my_pool
      job_variables:
        working_dir: "{{ prefect.variables.s2_shared_dir_prod }}"
        env:
          PIPELINE_SHARED_DIR: "{{ prefect.variables.s2_shared_dir_prod }}"
          ENV_STATE: prod
Then your flow just reads:
Copy code
python
shared_dir = os.environ["PIPELINE_SHARED_DIR"]
env_state = os.environ["ENV_STATE"]
If you want less repetition still You can also use YAML anchors if your setup supports them well:
Copy code
yaml
definitions:
  dev_env: &dev_env
    working_dir: "{{ prefect.variables.s2_shared_dir_dev }}"
    env:
      PIPELINE_SHARED_DIR: "{{ prefect.variables.s2_shared_dir_dev }}"
      ENV_STATE: dev

  prod_env: &prod_env
    working_dir: "{{ prefect.variables.s2_shared_dir_prod }}"
    env:
      PIPELINE_SHARED_DIR: "{{ prefect.variables.s2_shared_dir_prod }}"
      ENV_STATE: prod

deployments:
  - name: s2_satproc_dev
    ...
    work_pool:
      name: my_pool
      job_variables:
        <<: *dev_env

  - name: s2_satproc_prod
    ...
    work_pool:
      name: my_pool
      job_variables:
        <<: *prod_env
Bottom line Because Prefect 3.6 won’t let 
env
reference 
working_dir
directly, the cleanest solution is: - define the path once outside 
job_variables
- use that same source for both fields Useful docs: - Deployments - Work pools - Prefect Variables If you want, I can turn your current deployment file into a full dev/prod example using either Prefect Variables or env vars.
✅ thought for 7.4 seconds
Happy to help. If you paste your current 
prefect.yaml
/ deployment YAML, I can rewrite it into a clean dev/prod version.
s
@Marvin I am trying to start a process worker with a YAML defined work pool, how can I fix this error:
Copy code
Discovered type 'process' for work pool 'athena_dev'.
Traceback (most recent call last):
  File ".../.venv/lib/python3.12/site-packages/prefect/cli/_utilities.py", line 37, in async_wrapper
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../.venv/lib/python3.12/site-packages/prefect/cli/worker.py", line 220, in start
    await worker.start(
  File ".../.venv/lib/python3.12/site-packages/prefect/workers/process.py", line 168, in start
    async with self as worker:
               ^^^^
  File ".../.venv/lib/python3.12/site-packages/prefect/workers/process.py", line 333, in __aenter__
    self._runner = await self._exit_stack.enter_async_context(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "$HOME/.local/share/uv/python/cpython-3.12.11-linux-x86_64-gnu/lib/python3.12/contextlib.py", line 659, in enter_async_context
    result = await _enter(cm)
             ^^^^^^^^^^^^^^^^
  File ".../.venv/lib/python3.12/site-packages/prefect/runner/runner.py", line 1392, in __aenter__
    self._tmp_dir.mkdir(parents=True, exist_ok=True)
  File "$HOME/.local/share/uv/python/cpython-3.12.11-linux-x86_64-gnu/lib/python3.12/pathlib.py", line 1311, in mkdir
    os.mkdir(self, mode)
PermissionError: [Errno 13] Permission denied: '/tmp/runner_storage/41e52b07-ab53-4e1f-a776-ce44816f7fa0'
m
✅ thought for 153.7 seconds
This error is not really about your YAML-defined work pool itself — it means the process worker cannot create its local temp working directory. Prefect is trying to create:
Copy code
text
/tmp/runner_storage/41e52b07-ab53-4e1f-a776-ce44816f7fa0
and the OS is denying it:
Copy code
text
PermissionError: [Errno 13] Permission denied
Why this happens For a process worker, Prefect creates a temporary runner directory under Python’s temp dir, which on Linux is usually 
/tmp
. In your environment, one of these is likely true: - 
/tmp
is not writable by the user running the worker - 
/tmp/runner_storage
already exists with bad ownership/permissions - 
TMPDIR
points somewhere unwritable - you’re on a shared/HPC/system-managed node with restricted temp dirs Fastest fix Start the worker with a writable temp directory by setting `TMPDIR`:
Copy code
bash
mkdir -p "$HOME/prefect_tmp"
TMPDIR="$HOME/prefect_tmp" prefect worker start --pool athena_dev --type process
If you’re using `uv`:
Copy code
bash
mkdir -p "$HOME/prefect_tmp"
TMPDIR="$HOME/prefect_tmp" uv run prefect worker start --pool athena_dev --type process
Recommended permanent fix Create a dedicated writable temp location and export 
TMPDIR
before starting the worker:
Copy code
bash
mkdir -p "$HOME/prefect_tmp"
chmod 700 "$HOME/prefect_tmp"
export TMPDIR="$HOME/prefect_tmp"

prefect worker start --pool athena_dev --type process
You can verify what Python will use as temp dir with:
Copy code
bash
python -c "import tempfile; print(tempfile.gettempdir())"
Also check 
/tmp/runner_storage
 If you want to keep using 
/tmp
, inspect permissions:
Copy code
bash
ls -ld /tmp
ls -ld /tmp/runner_storage
On a normal Linux system, 
/tmp
should usually look like:
Copy code
text
drwxrwxrwt
If 
/tmp/runner_storage
exists and is owned by another user, that could be the direct problem. Possible fix if you control that machine
Copy code
bash
rm -rf /tmp/runner_storage
or
Copy code
bash
sudo chown -R your_user:your_group /tmp/runner_storage
Only do that if you know it’s safe in your environment. Important note about work pool YAML This is not something you currently fix via work pool 
job_variables
or deployment YAML. The failing directory is created when the worker process starts, before flow runs are launched. So changing this in deployment YAML like 
working_dir
will not solve this particular error. Best practice for self-hosted/HPC/shared systems Use a user-owned temp directory explicitly:
Copy code
bash
export TMPDIR="$HOME/.cache/prefect/tmp"
mkdir -p "$TMPDIR"
prefect worker start --pool athena_dev --type process
Verified worker CLI The correct worker command is:
Copy code
bash
prefect worker start --pool athena_dev --type process
Docs: - Workers - Work pools If you want, paste how you launch the worker (
shell
, 
systemd
, 
slurm
, 
pbs
, etc.) and I can show the exact 
TMPDIR
fix for that setup.
Open in Slack
PreviousNext
Powered by