Setting Up and Deploying Prefect Server: A Comprehensive Guide

1.1. Creating the Prefect Server

To create the Prefect Server, we need a simple docker container with prefect installed. While there are no official images for a docker server, we can use the official one meant for workers.

The setup process involves providing the correct configuration for the database:

/Dockerfile.server

# Prefect version must match Dockerfile.worker and pyproject.toml
FROM prefecthq/prefect:2.16-python3.10 as base

# Set connection (format: `postgresql+asyncpg://{user}:{password}@{url}:{port}/prefect`)
ARG DB_URL
RUN prefect config set PREFECT_API_DATABASE_CONNECTION_URL="${DB_URL}"
RUN prefect config set PREFECT_SERVER_API_HOST=0.0.0.0

CMD ["prefect", "server", "start"]

To create the docker image, execute the following command, replacing XXX with the database URL. For a postgres database, the format should be like postgresql+asyncpg://{user}:{password}@{url}:{port}/prefect:

docker build --build-arg DB_URL=XXX -f Dockerfile.server -t northius/prefect/server .

Once the Prefect Server is set up, you can access the web UI. If it’s running locally, you can find it at localhost:4200. However, the server won’t be able to run any flow without a worker.

To deploy the server somewhere, like AWS ECS running as a Service with Fargate, ensure to provision a public URL so that other components can connect to it.

1.2. Creating a Prefect Worker

To create a Prefect Worker, we first need to create a Work Pool so that the Worker we create can join it.

You can do this using the Prefect UI:

Once you have the Work Pool, we will create a Worker with Docker, following a similar approach to the Server:

/Dockerfile.worker

# Prefect version must match Dockerfile.server and pyproject.toml
FROM prefecthq/prefect:2.16-python3.10 as base

# Connect to Prefect
ARG PREFECT_API_URL
ENV PREFECT_API_URL=${PREFECT_API_URL}

# Allow to config pool_name
ARG POOL_NAME=subprocess-pool
ENV POOL_NAME=${POOL_NAME}

ENTRYPOINT prefect worker start --pool ${POOL_NAME}

Similarly to before, we can build the image with:

docker build --build-arg PREFECT_API_URL=XXX -f Dockerfile.worker -t northius/prefect/worker .

As before, you will need to run this docker container somewhere. In my case, I’ll also run it as another ECS service using Fargate.

2.1. Getting the code

The first thing you will need to do is to create a Prefect Block (more info at Prefect Essentials | Using prefect blocks) to store the GitHub credentials if the repo is not public.

Once done, we can define how to get the code and the dependencies with:

/prefect.yaml

pull:
- prefect.deployments.steps.git_clone:
    id: clone-step
    repository: https://github.com/github_user/repo.git
    branch: main
    credentials: "{{ prefect.blocks.github.github }}"
- prefect.deployments.steps.pip_install_requirements:
    requirements_file: requirements.txt
    directory: "{{ clone-step.directory }}"
    stream_output: False

2.2. Creating the first Deployment

Now that we’ve defined how to fetch the code, we will create the first Deployment. Add the following code to the prefect.yaml you just created.

/prefect.yaml

- name: lambdas.athena_history
  description: Export Athena history with a lambda
  entrypoint: src/lambdas/lambdas.py:athena_history
  version: 0.1.0
  work_pool:
    name: subprocess-pool
  tags: ["type:prefect", "group:lambda", "job:athena_history"]

Here is what we are setting:

Now you can deploy this to the prefect server with:

set PREFECT_API_URL=XXX
prefect deploy --all

This will deploy all Deployments in prefect.yaml to the server.

Now you can head to the server and trigger a run of the Deployment from there.

2.3. Advanced Usage of Deployments

What we’ve created works fine, but once you start adding multiple Deployments, you will probably encounter a lot of repetition. To avoid this, you can define some defaults that will be later used in each Deployment with:

/prefect.yaml

definitions:
  work_pools:
    subprocess-pool: &subprocess-pool
      name: subprocess-pool

  version: &version
    version: 0.1.0

  # Define defaults (that can be overwritten)
  defaults: &defaults
    <<: *version
    work_pool: *subprocess-pool
    tags: ["type:prefect"]

You can read more about YAML alias nodes.

With that, you can reuse the defaults like so:

/prefect.yaml

- name: lambdas.athena_history
  <<: *defaults
  description: Export Athena history with a lambda
  entrypoint: src/lambdas/lambdas.py:athena_history
  tags: ["type:prefect", "group:lambda", "job:athena_history"]

- name: lambdas.emr_history
  <<: *defaults
  description: Export EMR history with a lambda
  entrypoint: src/lambdas/lambdas.py:emr_history
  tags: ["type:prefect", "group:lambda", "job:emr_history"]
  schedules:
    - cron: "0 3 * * *"

2.4. Deploying deployments

We’ve seen how to manually deploy the Deployments, but it’s better to do it with Continuous Deployment (CD).

prefect --no-prompt deploy --all

/prefect.yaml

prefect.deployments.steps.git_clone:
    id: clone-step
    repository: https://github.com/github_user/repo.git
-    branch: main
+    branch: your_current_branch
    credentials: "{{ prefect.blocks.github.github }}"

prefect --no-prompt deploy --name lambdas.athena_history

3.1. Connect Docker

For anything that runs in a Docker container (such as DBT in my case), you can follow the same approach we did in the Prefect worker. We need to set an argument for the API URL and then store it as an environment variable with:

ARG PREFECT_API_URL
ENV PREFECT_API_URL=${PREFECT_API_URL}

And then when we build it, we need to add the following argument:

--build-arg PREFECT_API_URL=XXX

3.2. Connect EMR

In EMR, we need to set the environment variable at the driver. We can do that by adding the following configuration:

--conf spark.emr-serverless.driverEnv.PREFECT_API_URL=XXX

5.1. Updating Tags Dynamically

One really cool thing you can do with Prefect is to dynamically edit the tags of each flow at runtime. In my case, it’s very useful for tagging the environment where the flow is running (such as env:pro or env:pre).

You can do it with:

/src/common/prefect_utils.py

import asyncio

from prefect import get_run_logger
from prefect.client import get_client
from prefect.context import get_run_context


def update_tags(tags):
    logger = get_run_logger()
    if not tags:
        logger.warning("No tags passed to 'update_tags', nothing to do")
        return True

    logger.info(f"Adding {tags=} to current flow_run")

    # Read current flow
    flow_run = get_run_context().flow_run
    tags += flow_run.tags

    client = get_client()
    asyncio.run(client.update_flow_run(flow_run.id, tags=set(tags)))
    return True

5.2. Running a Lambda

Here you have the simplest example I have for running external jobs from Prefect. This calls any lambda you might want:

/src/lambdas/base.py

from prefect import get_run_logger
from prefect import task

from src.common.session import get_session


@task(name="prefect.lambdas.run")
def _run_lambda(client, name):
    logger = get_run_logger()

    response = client.invoke(
        FunctionName=name,
        InvocationType="RequestResponse",
        LogType="None",
    )

    if (status_code := response["StatusCode"]) == 200:
        logger.info(f"Lambda {name=} completed successfully")
        return True

    logger.error(f"Lambda {name=} failed with {status_code=}")


def run_lambda(env, name):
    logger = get_run_logger()
    logger.info(f"Running lambda {name=} in {env=}")

    client = get_session(env).client("lambda")
    _run_lambda(client, name)

5.3. Defining a Flow for Each Possible Lambda

With the base.py code for running lambdas, you can define a flow for each lambda you have. As an example, we could have:

/src/lambdas/lambdas.py

from typing import Literal

from prefect import flow

from prefect_northius.lambdas.base import run_lambda


@flow(name="prefect.lambdas.athena_history")
def athena_history(env: Literal["snd", "pro"]):
    run_lambda(env, "nt-lambda-function-athena-history")


@flow(name="prefect.lambdas.emr_history")
def emr_history(env: Literal["snd", "pro"]):
    run_lambda(env, "nt-lambda-function-emr-history")