Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

I'm struggling to understand how to use a k8s custom job template for a kubernetes-job - I found some great examples from <@U02H1A95XDW> but I'm not sure how to use/implement it:  <https://github.com/anna-geller/prefect-deployment-patterns/tree/main/blocks/infrastructure_blocks/kubernetes-job/kubernetes_job_custom_job_template>

here's my `log_flow.py` example:
```import sys
import prefect
from prefect import flow, task, get_run_logger
from prefect.infrastructure import KubernetesJob
from utilities import AN_IMPORTED_MESSAGE


k8s_job = KubernetesJob(
    namespace="prefect-orion",
    image="prefecthq/prefect:2.7.11-python3.10",
    env={"EXTRA_PIP_PACKAGES": "adlfs"},
    job=KubernetesJob.job_from_file("../../../k8s/prefect-orion-job-template.yaml"),
)
k8s_job.save("prod", overwrite=True)

@task
def log_task(name):
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello %s!", name)
    <http://logger.info|logger.info>("Prefect Version = %s 🚀", prefect.__version__)
    logger.debug(AN_IMPORTED_MESSAGE)


@flow(name="log_flow")
def log_flow(name: str):
    log_task(name)


if __name__ == "__main__":
    name = sys.argv[1]
    log_flow(name)```

then to build it:  `prefect deployment build ./log_flow.py:log_flow -n log-simple -q default -i kubernetes-job -sb azure/prefect-orion`

then back in the Orion UI, the flow fails with:
```Downloading flow code from storage at ''
04:15:50 PM

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/fsspec/registry.py", line 211, in get_filesystem_class
    register_implementation(protocol, _import_class(bit["class"]))
  File "/usr/local/lib/python3.10/site-packages/fsspec/registry.py", line 234, in _import_class
    mod = importlib.import_module(mod)
  File "/usr/local/lib/python3.10/importlib/__init__.py", line 126, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "&lt;frozen importlib._bootstrap&gt;", line 1050, in _gcd_import
  File "&lt;frozen importlib._bootstrap&gt;", line 1027, in _find_and_load
  File "&lt;frozen importlib._bootstrap&gt;", line 1004, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'adlfs'
...
...
RuntimeError: File system created with scheme 'az' from base path '<az://prefect-orion>' could not be created. You are likely missing a Python module required to use the given storage protocol.```

hey <@U04PD8J0XDE> - looking at your `build` command, I think instead of the prefect default k8s infra
```-i kubernetes-job```
you'd want to use your infrastructure block you created
```-ib kubernetes-job/prod```
since that provides the filesystem protocol in your `EXTRA_PIP_PACKAGES`

these <https://docs.prefect.io/concepts/infrastructure/#using-infrastructure|docs> may be useful