log_flow.py

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)

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

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 "<frozen importlib._bootstrap>", line 1050, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1027, in _find_and_load
  File "<frozen importlib._bootstrap>", 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.