Flow deployment patterns

[zanieb]

This is a home for discussion of patterns users are using for CI/CD of their flows.

We are working on developing an opinion on what patterns are best-practice and in the interim want to enable our community to share their thoughts on this important part of developing at scale with Prefect.

Please abide by some simple posting rules so this can stay organized:

• Post new patterns as a top-level comment
• Comment on existing patterns in-thread
• Do not ask general questions here e.g. "How do I do CI/CD?"
• Use the template below as a guideline for describing your pattern

---

(this template may be updated after a few examples are in here)

## Overview
*A bullet point list describing the process of flow development -> deployment -> run*

## Project structure
*An outline of your project file structure and brief discussion*

## Developing flows
*Writing a flow, testing a flow*

### Writing flows
*Where are flows written? Do they share common patterns?*

### Testing flows
*How flows are tested*

## Deploying flows
*Discovering flows in source, building storage, registering serialized flow with the backend*

### Registering flows
*How flows are registered. How are the flows discovered? How are options determined?*

### Storing flows
*What flow storage do you use? Why?*

## Running flows
*Configuring a flow run, deploying an agent, managing flow environments*

### Run config
*What run config do you use for your flows? How do you set it?*

### Agent
*How are you managing your agent?*

### Environments
*How are you ensuring your flow has its dependencies? How are you setting variables?

## Limitations
*Downsides to your approach*

## Benefits
*Benefits to your approach*

[zanieb]

Here's an example -- I haven't actually used this approach yet but want to provide some tangible ideas for how the template would be filled.

Overview

• Write flows in a specified folder
• Have shared items in a python module installed into a Docker image
• CI builds and tags the Dockerfile
• CI registers flows
  • Sets and builds remote storage for the flow
  • Sets DockerRun using the image tag from above
• Flows are now ready to run with a docker agent that has access to the remote storage and image registry

Project structure

flows/                      -- Holds all flows for the project
    example_a.py            -- a file per flow
    example_b.py
python/
    utilities_module/
        shared_tasks.py 
        utilities.py
Dockerfile                  -- Uses prefect's image as base, installs our python package
requirements.txt            -- Requirements for the shared utilities
setup.py                    -- installs `utilities_module`

Developing flows

Writing flows

Flows are written in their own files within the flows/ folder. Each flow file has the testing code block at the bottom for testing

if __name__ == "__main__":
    flow.run()

Flows define their own executor but storage and run configs are set by the CI process.

Testing flows

Flows are tested locally using python flows/example_a.py

Deploying flows

Registering flows

Flows are registered in CI by iterating through every file in the flows/ folder and importing the flow variable for registration. With this, all flows are registered to the same project. To add projects, the flows/ folder could have project subfolders e.g. flows/project_a/flow_a.py and the registration script could interpret it.

Here we will set the storage and run_config to be consistent across all flows.

Storing flows

Flows are stored on a remote backend of your choice e.g. S3/GCS.

Running flows

Run config

The run config for this project is a DockerRun that uses the image tag that we've built in CI. At registration time, we set this image tag by importing it from the environment. We'll attach the run config to each flow before registering it

flow.run_config = DockerRun(image=os.environ.get("PROJECT_DOCKER_IMAGE_TAG", None))

Agent

Uses a docker agent. The agent could be run locally or on a single cloud node.

Environments

Dependencies are managed by pip and our Dockerfile. Environment variables can be set at CI time so if we have a 'staging' branch they can added to the DockerRun run config then.

Limitations

• All flows share the same base image and requirements.txt

Benefits

• Flows can share utilities
• Python requirements.txt file can be used for dependency management
• Flows are isolated from other code
• Fast flow storage build times (a new docker image doesn't need to be built for each flow)

[sti0]

Could you provide a snippet to the flow registration? How do I import the flow variable when using context manager?

Flows are registered in CI by iterating through every file in the flows/ folder and importing the flow variable for registration. With this, all flows are registered to the same project. To add projects, the flows/ folder could have project subfolders e.g. flows/project_a/flow_a.py and the registration script could interpret it.
Here we will set the storage and run_config to be consistent across all flows.

Thank you

Edit:

Find a solution by myself and like to share it with others:

import importlib
from prefect.storage import Local

STORAGE = Local("/some/path", stored_as_script=False)

module = importlib.import_module("flow_module")

module.flow.storage = STORAGE
module.flow.register(project_name="testpro")

[sti0]

Lerning never stops... better approach was to simply use the build-in utilities function extract_flow_from_file()

prefect/src/prefect/utilities/storage.py

Line 53 in f1bf35b

def extract_flow_from_file(

Somehow it's missing at the doc's page.

[aripollak]

Here's a script we're using in GitHub Actions to register all flows in .py files in the given directory path while running on the given Docker image (we're using ECR to host it), using S3 script storage and an ECS agent, so the flow itself doesn't have to specify those defaults. When built on a pull request, this is called with a test --project and --no-schedule. The Docker image contains shared tasks in an installed Python package, along with necessary Python dependencies.

Because of Docker build caching, we don't have to build & push a new image if the shared code hasn't changed, but unfortunately this script with S3 storage causes a new S3 upload & registration for each flow every time this is run. A future improvement might be to detect which files have changed in the current git commit, and only register those. Alternatively, take advantage of the reregistration by pointing the image tag to a unique version that contains the same shared code from the current commit, so the task code wouldn't get out of sync with the shared code in the Docker image.

prefect_register.py

import argparse
import pathlib
import runpy
import sys

from prefect.core.flow import Flow
from prefect.cli.build_register import build_and_register, get_project_id
from prefect.client.client import Client
from prefect.run_configs.ecs import ECSRun
from prefect.storage.s3 import S3
from prefect.utilities.context import context


def main():
    args = parse_args()
    client = Client()
    project_id = get_project_id(client, args.project)
    flows = flows_with_default_storage_and_runconfig(args)
    # Use Prefect CLI's builtin registration utility since it has some nice
    # status messages
    build_and_register(client, flows, project_id, schedule=args.schedule)


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--project', required=True)
    parser.add_argument('--schedule', action=argparse.BooleanOptionalAction,
                        default=True)
    parser.add_argument('--path',  type=pathlib.Path, required=True)
    parser.add_argument('--bucket', required=True)
    parser.add_argument('--image', required=True)
    return parser.parse_args()


def flows_with_default_storage_and_runconfig(args):
    paths = args.path.glob('*.py')
    flows = []
    for path in paths:
        # Modified from https://github.com/PrefectHQ/prefect/blob/a236c4c52/src/prefect/cli/build_register.py#L133
        # so we don't set Local storage by default
        try:
            with context({'loading_flow': True, 'local_script_path': str(path)}):
                namespace = runpy.run_path(path, run_name='<flow>')
        except Exception:
            print(f'Error loading {path!r}:', sys.stderr)
            raise

        for flow in namespace.values():
            if not isinstance(flow, Flow):
                continue
            if flow.storage is None:
                flow.storage = S3(bucket=args.bucket, stored_as_script=True,
                                  local_script_path=str(path))
            if flow.result is None:
                # Not sure why this is necessary for saving task results to S3 given it's already happening in
                # https://github.com/PrefectHQ/prefect/blob/a236c4c58/src/prefect/cli/build_register.py#L308
                flow.result = flow.storage.result
            if flow.run_config is None:
                flow.run_config = ECSRun()
            if flow.run_config.image is None:
                # This is separate from the main run_config check above so that the
                # flow can specify other custom ECSRun parameters
                flow.run_config.image = args.image

            flows.append(flow)
    return flows


if __name__ == '__main__':
    main()

[speedyturkey]

Thanks for starting this conversation, @madkinsz ! Here is my first crack at this:

Overview

I am not finished building this out, so some items are "how I intend to do it" as opposed to "what I've already done".

• Write flows to a specified folder
• Shared code installed into a base docker image
• CI adds common KubernetesRun attributes
• CI adds docker storage and calls .register()
• Flows are now ready to run on Kubernetes cluster

We're an AWS shop, and use CodePipeline/CodeBuild for CI/CD, and we're using EKS for running our agents and flows. We have some fairly consistent and opinionated practices around infra-as-code and CI/CD. In short, we design our repos/apps so that each git branch is deployable as a fully operational entity. This means our Prefect projects, flows, etc will all be defined in this context, eg project name of my-app-master or my-app-feature-branch.

I'm eager and grateful for any feedback, thoughts, or alternative ways of doing things! I'm also willing to share (lightly edited) code examples if helpful.

Project structure

Here is my basic structure. I am omitting common code and other unrelated items. The deploy script handles infrastructure, helm charts, and calls the register_flows.py file.

├── bin
│   ├── deploy
│   ├── dispatch.sh
│   └── register_flows.py
├── buildspec.yml
└── flows
    ├── job_template.yaml
    ├── flow_1.py
    └── flow_2.py

Developing flows

Flows go in the flows directory. It is my intention that the flow author has the ability to ignore essentially the entire build process and just write their flow.

Writing flows

TBD

Testing flows

TBD

Deploying flows

Discovering flows in source, building storage, registering serialized flow with the backend

Registering flows

I have a designated directory called flows, and make use of the prefect.utilities.storage.extract_flow_from_file helper function.

Philosophically, I have tried to set as many sensible defaults as possible so that flow files can be focused entirely on the logic of the flow/tasks, and flow authors can override specific configuration options as needed.

If a given flow has storage already defined, the build process respects that. If not, it adds default docker storage:

• Currently using prefecthq base image, but will soon switch to a custom base image so that private Python code can be loaded.
• Image name is assigned in the format {APP}/{flow.name}-{BRANCH}.
• Images are pushed to ECR (AWS private image repository)

If a given flow has a run configuration already defined, the build process respects that. If not, it adds default configuration:

• A custom Kubernetes job template, which includes labels/annotations for logging and monitoring
• A service account name, which is specific to the application/branch in question
• A basic set of environment variables.

Finally, registration is called:

• images are built at this stage
• a label indicating environment (dev, prod, etc) is assigned to the flow
• an idempotency key is used (flow.serialized_hash)

Storing flows

We are currently using docker storage. This is a natural fit, as we were already using docker images and are comfortable with the tooling. I'd be open to other storage (such as S3) if there was some particular use case where docker was not well-suited.

Running flows

I am still building this out, but I am planning to create a wrapper around the prefect CLI which will make running flows easier, given the opinionated nature of our infra and CI/CD processes. I anticipate all flows will run on Kubernetes.

Run config

So far, exclusively using KubernetesRun.

Agent

I have a dedicated repo for our Prefect agents. We have one agent per environment (dev, prod) which are being run as kubernetes deployments, via helm.

Environments

I anticipate passing a list of Python dependencies to be installed in the docker image at build/registration time, but have not focused on this area yet. Generally, we only use a few basic environment variables, and these are set on the run configuration. Other, more specific variables can be looked up from the SSM Parameter Store (AWS service) if needed.

Limitations

• Not yet able to store multiple flows in a single Python file. Currently thinking of writing a helper extract_all_flows_from_file.
• As more flows are added, I expect build times to grow linearly. I'd like to come up with a good strategy to leverage caching and to build only flows which may have changed.
• I have not yet spent time on event-driven flows, which will represent a non-trivial number of our flows. I plan to follow the "use lambda to do graphql mutations" example I've seen on the blog.

Benefits

• Fits in well with our existing philosophy and tooling.
• Separation of concerns - agents, storage, flow config, flow logic
• Almost everything is automated and I am expecting it to be very flexible
• If I need anything fundamentally different, it's easy to create a new repo for different kinds of flows

[josh-gree]

Does using the idempotency key lead to better docker caching? Regardless of the changes I make the docker build always starts from the very beginning and it really slows the dev process down...

[josh-gree]

I wonder if this may lead towards not actually using docker for storage and perhaps using block storage instead (s3, gcs,...) along with a custom docker image!?

[josh-gree]

Although not completely sure how that would fit together...

[speedyturkey]

@josh-gree Idempotency key simply prevents prefect from incrementing the flow version if no changes have been made to that specific flow. Unfortunately doesn't seem to have any direct effect on docker caching, although I'm no expert on this topic.
I think most of your build time is going to come from installing individual sets of Python dependencies on top of your base image, so I'm not sure that storing flows on S3/GCS would make a huge difference here - if I'm wrong I'd love to hear your experience or others'.

[zanieb]

@josh-gree If you use a DockerRun and file-based storage then you don't have to rebuild your docker image each time your flow changes which can lead to some significant speedups here. The docker build should be caching layers before the step that adds the flow though so I'm a bit confused that builds are very slow for you.

[alexifm]

Overview

This is the approach used at infima developed by @marwan116 and myself. A first pass with a few iterations on it has been built out and is in active use. Can provide more detailed code of certain parts if that would be of interest.

• Flows are built in a module within the internal package we develop
• Gitlab CI pipeline auto-registers flows to prefect cloud using the package CLI tool
• Flows are versioned with our package version and run against a docker image with the library
• A FlowBuilder class which handles a lot of the boilerplate build of a Flow, like dask-k8s environment setup
• The dev’s job is to implement the method that contains the actual flow logic, define the flow parameters, and specify environment configs values.

Project structure

Flows are a module within the broader package:

src/infima/flows/
    cli.py                  -- CLI for handling flows
    core.py                 -- FlowBuilder class lives here
    example_a/example_a.py  -- Creates a FlowBuilder object

Developing & Writing flows

The dev’s job is to implement the method that contains the actual flow logic, define the flow parameters, and specify environment configs values.

@attr.s(auto_attribs=True, kw_only=True)
class ExampleFlowBuilder(FlowBuilder):
    """Example FlowBuilder."""

    param: Parameter = paramattr("param", "test")  # <- a special parameter attribute

    def build_flow(self) -> Flow:
        """Build the flow."""
        with Flow(self.name) as flow:
            # Developer implements the flow here
            # Can use self.param here as a flow parameter
        return flow

flow = ExampleFlowBuilder(
    name="name-of-flow",
    # Cloud
    project_name="name-of-project",
    # Storage
    bucket="s3-bucket",  # <- we're using S3 for now
    # Environment
    image="",
    k8s_secrets=[],  # <- we use K8s secrets on EKS
    min_workers=1,   # <- We programmatically build out a Dask-K8s spec and Environment
    max_workers=2,
    dask_memory_management=True,
    ng_name="",
    worker_cpu=,
    worker_memory_gb=,
    # flow build arguments
    other_attribute=
)

Testing flows

We plan on using FlowBuilder to enforce a test mode parameter. We would then run each flow in test mode in the CI test stage. We will be running the tests in our cluster so we don't have to mock as much but also test the flows closer to the real thing instead of relying on a lot of mocking.

Deploying/Registering flows

We have a manual way to indicate within the package if a flow is "active". Our CLI tool calls infima flows register and will register all active flows on version bump (we use semantic release for the package). Registration happens even if the flow doesn't change. The flow depends on other modules in the package so we just move all flows along with the package instead of a "smart" registration. We might move to a registry approach to auto activate a flow in the package.

Storing flows

We currently use S3 storage because it's easy. Also, last we checked, you could not pull a flow by name from a Git storage though it should be possible. We will likely overhaul this all when we migrate to RunConfig

Running flows

We have an agent running in EKS. Our FlowBuilder is tailored to running Dask-K8s. Schedules can also be set in the builder. Otherwise one can infima flows run <name> ... with some args or go to the cloud UI.

Future Plans

We are planning to migrate to a staging/prod split where the flows will still be automatically registered but the auto-register to prod will only occur when we create a prod release through a tag. This means there will be duplication of some of the flows that require it.

Limitations

• Because we completely ignore the version on the cloud UI and it's all tied to package version, the versioning is opaque on the Prefect Cloud.
• There's no "smart" registration: a flow and anything it relies on could be completely untouched but it'll keep getting registered because it needs to point to a different version of our docker image
• We're restricted to whatever environment the FlowBuilder allows. It's opinionated to dask-k8s on AWS EKS.
• We think we have run into race conditions where we register a flow and conflict with it’s scheduled operation but we try to be robust with the construction of the flows.
• The CI pipeline is purely additive: we cannot delete flows that get removed/renamed

Benefits

• We got dask-k8s working well and so the benefit for a dev is it's ridiculously easy to get a flow going because of the FlowBuilder.
• It's also trivial to have the flow registered since it's part of the CI/CD process on Gitlab.
• No dependency hell: We initially had a separate repo for flows but moved all the flows into our main package because we couldn’t keep the versions synced.

[zanieb]

Thanks for contributing! This is a lot like what I sketched out but I used a decorator so that my (theoretical) data scientists would not have to write a class. Building a CLI into someone's repo seemed a little heavy for something provided by Prefect so we dropped it. There are some thoughts in #3927 as well about defining flows in a function, but it hasn't gotten traction :)

[alexifm]

This is a lot like what I sketched out but I used a decorator so that my (theoretical) data scientists would not have to write a class

We're on like iteration 1b. We got it up and running and just haven't moved it ahead much. If it aint broke, don't fix it, right? But yeah, I could totally see improved patterns since we're still duplicating a fair amount and potentially excessively verbose.

Building a CLI into someone's repo seemed a little heavy for something provided by Prefect so we dropped it.

I think what we did was definitely a mix of things that could be good for the community and things that are very opinionated. For one, so much of it is completely tied into a formal Python package structure, semantic release, and all that.

[mousetree]

Overview

Our approach is as follows:

• We have a single repository for all flows (we call it sable-batch)
• Developers create a new flow and place the .py file in the flows/ folder
• Any supporting code is added to the lib/ or utils/ folders
• We use setup.py to describe all dependencies
• Developers don't have to worry about specifying deploying parameters. We use a custom Dockerfile that uses pip install to install the entire sable-batch project.
• Our CI process calls a build.py script that imports each flow file, adds it to a single Docker container, pushes the container and registers each flow.
• We still use the legacy Docker storage and KubernetesJobEnvironment but are planning on changing this (more on this below)

Project structure

.
├── .env
├── Dockerfile
├── docs
├── readme.md
├── sable_batch
│   ├── build.py
│   ├── constants.py
│   ├── flows
│   │   ├── flow1.py
│   │   ├── flow2.py
│   │   ├── flow3.py
│   │   ├── files/ # files that are needed by flows
│   │   ├── schemas # bigquery schemas
│   │   │   └── schema1.py
│   │   ├── sql # sql scripts used by flows - we have some custom tasks that use the filename as input and run on BigQuery
│   │   │   ├── script1.sql
│   │   │   ├── script2.sql
│   ├── job_spec_default.yaml
│   ├── lib
│   │   ├── lib1.py
│   │   └── lib2.py
│   ├── tasks
│   │   ├── custom_task1.py
│   │   └── custom_task2.py
│   ├── test_build.py
│   └── utils # deprecate in favour of lib folder
│       ├── util1.py
│       └── util2.py
├── setup.cfg
├── setup.py
└── tests
    ├── test_flow1.py
    ├── test_flow2.py
    └── test_lib1.py

Developing flows

Writing a flow, testing a flow

We currently keep flows very simple. Here is a very minimal example:

import prefect
from dotenv import find_dotenv, load_dotenv
from prefect import Flow, task
from prefect.tasks.secrets import EnvVarSecret

from sable_batch.lib.stuff import do_stuff
from sable_batch.tasks import SimpleSQL

@task
def do_something(some_data, some_secret: str):
    logger = prefect.context.get("logger")
    logger.info("Something")
    do_stuff() # this is a method in our shared lib 


with Flow("hello world") as flow:
    some_secret = EnvVarSecret("SOME_SECRET", raise_if_missing=True)
    some_data = SimpleSQL("file1.sql")
    do_something(some_data, some_secret)

if __name__ == "__main__":
    load_dotenv(find_dotenv(raise_error_if_not_found=True), override=True)
    flow.run()

We use EnvVarSecret a lot and use dotenv for managing all variables locally. The variables are set on the cluster using the job_spec.yaml file (described later).

We store tests in the /tests folder and use pytest to run them on every PR. Here is a sample test file:

from unittest.mock import MagicMock, patch
import prefect
import pandas as pd

from sable_batch.flows.flow1 import (
    extract_names
)


class MockedLogger:
    def info(msg):
        print(msg)

    def warning(msg):
        print(msg)


@patch.object(MockedLogger, "info")
def test_extract_names(logger):
    given_value = pd.DataFrame(
        [{"first_name": "Philip Joe", "middle_name": "", "last_name": ""}]
    )
    expected_value = pd.DataFrame(
        [{"first_name": "Philip", "middle_name": "", "last_name": "Joe"}]
    )
    assert extract_names(given_value, logger).equals(expected_value) is True

Writing flows

Flows are all stored in the sable_batch/flows directly. The only common pattern they share is that they all have with Flow("name") as flow: at the root so that the build script can import them and access the flow object (more on this below).

Testing flows

See example above. On each PR we run: pytest --disable-warnings. Additionally we have a small utility script that tests if the flows can be pickled. Rather than waiting for the prefect's validation process to run when building a Docker, we manually check this earlier:

from prefect import Flow

# this is a variable in the sibling build.py file (discussed below)that lists all flows
from build import FLOWS 

print("Testing pickling...")
for flowfile in FLOWS:
    pickled_path = flowfile.flow.save()
    unpickled_flow = Flow.load(pickled_path)
    print(f"✔ {unpickled_flow.name}")

Deploying flows

Our CI process executes a build.py file that looks something like:

import uuid
from os import environ, path

import docker
from prefect.environments import KubernetesJobEnvironment
from prefect.environments.storage import Docker

# We manually import the flows here but should move to some type of auto discovery
from sable_batch.flows import (
    flow1,
    flow2,
    flow3
)

# We iterate through this later in the script
FLOWS = [
    flow1,
    flow2,
    flow3
]


def main():
    here = path.dirname(path.realpath(__file__))
    registry_url = "gcr.io/something/sable_batch"
    image_tag = uuid.uuid4().hex
    tls_config = None
    base_url = None
    dockerfile = path.join(here, "..", "Dockerfile")

    is_ci = environ.get("CI", default=False)

   # You can ignore this, it's needed just to make Circle's Docker-in-Docker build work. Thought it might be useful for others who use Circle
    if is_ci:
        tls_config = docker.tls.TLSConfig(
            client_cert=(
                path.join(environ.get("DOCKER_CERT_PATH", ""), "cert.pem"),
                path.join(environ.get("DOCKER_CERT_PATH", ""), "key.pem"),
            ),
            verify=False,
        )
        base_url = environ.get("DOCKER_HOST")

    storage = Docker(
        image_name="sable_batch",
        registry_url=registry_url,
        image_tag=image_tag,
        base_url=base_url,  # required for CircleCI
        tls_config=tls_config,  # required for CircleCI
        dockerfile=dockerfile,
    )
    
    for flow_file in FLOWS:
        flow = flow_file.flow # every flow must expose a `flow` object

        job_spec = "job_spec_default.yaml" # described below
        agent_labels = []

       # we add these floating constants to flow files to mark custom things (would prefer to somehow do it on the flow object itself)
        if "JOB_SPEC" in dir(flow_file):
            job_spec = flow_file.JOB_SPEC
        if "AGENT_LABELS" in dir(flow_file):
            agent_labels = flow_file.AGENT_LABELS

        print(
            f'Adding "{flow.name}" using job spec {job_spec} and labels {agent_labels}'
        )
        
        # We still use the legacy environment
        flow.environment = KubernetesJobEnvironment(
            labels=agent_labels,
            job_spec_file=path.join(here, job_spec),
            unique_job_name=True,
        )
        
        # We add all flows to a single Docker and push it. We want to move away from this (more later)
        flow.storage = storage
        storage.add_flow(flow)

    storage.build(push=is_ci)

    # Only register flows once the Docker image has been uploaded
    if is_ci:
        for flow_file in FLOWS:
            flow = flow_file.flow
            flow.register(
                project_name="default",
                build=False,
                idempotency_key=flow.serialized_hash(), # Thank you prefect!
            )


if __name__ == "__main__":
    main()

Some of the other elements referenced above are the Dockerfile:

FROM prefecthq/prefect:0.13.18-python3.8

RUN apt-get update && \
    apt-get install -y gnupg

WORKDIR /opt/sable_batch

COPY . .
RUN pip install -e .
RUN prefect version

And our setup.py file:

#!/usr/bin/env python

"""The setup script."""

from setuptools import find_packages, setup

requirements = (
    [
        "prefect[gcp,kubernetes,postgres,viz]==0.13.18",
        "pandas",
        "requests",
        "analytics-python",
        "beautifulsoup4",
        "cloudpickle",
        "python-dotenv",
        "sqlalchemy",
        "google-cloud-bigquery-storage<2.0.0",
        "pyarrow",
        "flatten-dict",
        "us",
        "pysftp",
        "pytest",
        "python-gnupg",
        "xlsxwriter",
        "chardet<4.0.0",
    ],
)

setup_requirements = [
    "pytest-runner",
]

test_requirements = [
    "pytest>=3",
]

extras = {"dev": ["black", "flake8", "mypy", "pre-commit"]}

setup(
    author="Someone",
    author_email="someone@someone.com",
    python_requires=">=3.5",
    description="Sable Batch",
    install_requires=requirements,
    extras_require=extras,
    long_description="",
    include_package_data=True,
    keywords="sable_batch",
    name="sable_batch",
    packages=find_packages(include=["sable_batch"]),
    setup_requires=setup_requirements,
    test_suite="tests",
    tests_require=test_requirements,
    url="https://github.com/sable-money/sable-batch",
    version="0.1.0",
)

And if anyone is using CircleCI, this is what our .circle/config.yml looks like:

version: 2.1
orbs:
  gcp-gcr: circleci/gcp-gcr@0.7.1
jobs:
  build:
    docker:
      - image: circleci/python:3.8
        environment:
          PIPENV_VENV_IN_PROJECT: true
    steps:
      - checkout
      - run: sudo chown -R circleci:circleci /usr/local/bin
      - run:
          name: Install Dependencies # this is not ideal as we dont need everything
          command: |
            pip install -e .
      - setup_remote_docker:
          version: 19.03.12
          docker_layer_caching: true
      - gcp-gcr/gcr-auth
      - run:
          name: Authenticate with Prefect Cloud
          command: |
            prefect backend cloud
            prefect auth login -t ${PREFECT_CLOUD_TOKEN}
      - run:
          name: Build container
          command: |
            python sable_batch/test_build.py
            python sable_batch/build.py

workflows:
  version: 2
  build-workflow:
    jobs:
      - build:
          context: SABLE_PRODUCTION
          filters:
            branches:
              only:
                - master

Registering flows

See above

Storing flows

During the CI process we add each flow to a Docker container. We then build and push a monster Docker container that contains all flows and all dependencies. This is not ideal as a new container is built every time the CI runs. We want to move to a process where we have a base Docker image that contains all dependencies and Python library code (that only gets updated if setup.py or lib/ changes) and flows that are stored in S3.

Running flows

During build.py each flow is set to use this custom job spec:

apiVersion: batch/v1
kind: Job
metadata:
  name: prefect-job
  labels:
    identifier: ""
spec:
  template:
    metadata:
      labels:
        identifier: ""
    spec:
      restartPolicy: Never
      containers:
        - name: flow-container
          image: ""
          command: []
          args: []
          envFrom:
            - secretRef:
                name: prefect-globals
            - configMapRef:
                name: prefect-globals
          resources:
            limits:
              cpu: "1"
              memory: 1G
            requests:
              cpu: "0.5"
              memory: 300M

We run the Prefect agent using a standard k8s deployment.yaml file. The only notable thing is that we use ConfigMap and Secret for storing all environment variables. These are stored outside of the sable_batch repo.

Run config

See above

Agent

Standard Kubernetes Deployment

Environments

See above

Limitations

• It's very inefficient to rebuild the mono Docker container every time a flow changes. We will soon move to a stable base Docker image containing the dependencies and library code + flows stored in S3.
• We have no real method for setting flow-specific deployment/runtime options. All flows share the same config (with a few messy exceptions that use constants exposed in the module). We will hopefully soon allow individual flows to be configured using some convention
• We're still using the legacy KubernetesJobEnvironment.
• All flows are publishing to the same "default" project. In future, we want to allow flows to describe the project they belong to, or be placed in a folder per project.
• We do not have a test environment for testing flows against real data. Instead developers run them on their machine and set the environment variables to the development settings. This is easily fixable during the CI process and using labels for development agents I guess.
• Agent versions can drift from the runtime image version as they are set in different repos (not exactly sure if the agent needs to be the same as the flow's image?)

Benefits

• Dependencies are easily managed in setup.py
• Developers can easily get things setup locally by running pip install -e . and adding .env file. The development setup will mirror the production setup as the Dockerfile does the same.
• We have a few conventional folders for specific things: flows/, tasks/ for custom Prefect Tasks, lib/ for shared code etc. This should allow us to easily improve the build process (only rebuild Docker if lib/ or setup.py changes; auto import everything from flows/
• All environment variables can be set locally using a .env file and we use our existing ConfigMap and Secrets to provide them in production

Future plans

We have lots of improvements we'd like to make:

• Move to RunConfig. We would also ideally have default RunConfig's applied to all flows that don't have one specified already (as most of the run configs should be more or less the same)
• Use a slowly changing base Docker image that contains dependencies and shared library code + flows stored in S3. The CI process will only run what needs to run
• Support flows being registered to different projects (and probably having folder per project)
• Figure out how Dask works
• Improve test coverage and implement some form of E2E testing (maybe Great Expectations or similar)
• Start using DBT Prefect tasks as we currently have a lot of SQL files in addition to Pandas-style Python

[mousetree]

We've now made a lot of new improvements:

Change 1: Use folders for Prefect projects.

Our tree now looks like:

- sable_batch
     - flows
        └── operations
            └── flow1.py
            └── flow2.py
        └── reporting
            └── flow3.py
     - build_projects.py
     - build_config.py
        

Change 2: Auto detect flows and register

The following script walks the directory /sable_batch/flows and uses the folders it finds as Prefect projects and the files under each project as flows (if they can import)

# sable_batch/build_projects.py

import argparse
import logging
import pathlib
from dataclasses import dataclass
from glob import glob
from os import listdir, path, walk
from typing import NamedTuple

from build_config import (
    get_default_executor,
    get_default_run_config,
    get_default_storage,
)
from prefect.core.flow import Flow
from prefect.environments import LocalEnvironment
from prefect.utilities.storage import extract_flow_from_file

IGNORE_PATHS = ["__pycache__", "schemas", "files", "sql", "flows"]


class FlowBuild(NamedTuple):
    project: str
    flow: Flow


def get_flows(filter_file=None):
    logging.info("Discovering flows...")
    flows = []
    # Default flows directory is /sable_batch/flows
    flows_dir = path.join(path.dirname(path.realpath(__file__)), "flows")
    for dirpath, dirs, files in walk(flows_dir):
        # Ignore legacy files that will later be moved
        if path.basename(dirpath) not in IGNORE_PATHS:
            # Prefect project name is the folder name
            project = path.basename(dirpath)
            for file in files:
                # Only use Python files
                if filter_file and filter_file not in file:
                    continue
                if path.splitext(file)[1] == ".py":
                    full_path = path.join(dirpath, file)
                    logging.info(f"Found {project} >> {file}")
                    try:
                        flow = extract_flow_from_file(full_path)
                        flows.append(FlowBuild(project, flow))
                    except Exception as e:
                        logging.error(e)
    return flows


def build_flows(flow_builds: [FlowBuild], test_mode=False):
    for flow_build in flow_builds:
        flow = flow_build.flow
        logging.debug(flow.name)
        flow.validate()
        if not flow.run_config:
            flow.run_config = get_default_run_config()
        if not flow.storage:
            flow.storage = get_default_storage()
        if not flow.executor:
            flow.executor = get_default_executor()
        if not test_mode:
            flow.register(
                project_name=flow_build.project,
                idempotency_key=flow.serialized_hash(),
            )


def main(args):
    test_mode = args.test
    filter_file = args.filename
    flow_builds = get_flows(filter_file=filter_file)
    build_flows(flow_builds, test_mode=test_mode)


parser = argparse.ArgumentParser(description="Builds and deploys all prefect flows")
parser.add_argument(
    "--test", help="Run tests only. Skips registration", action="store_true"
)
parser.add_argument("--debug", help="Include debug logs", action="store_true")
parser.add_argument(
    "filename", nargs="?", default=None, help="Optional filename to filter by"
)


if __name__ == "__main__":
    args = parser.parse_args()
    log_level = logging.DEBUG if args.debug else logging.INFO
    logging.basicConfig(level=log_level)
    logging.getLogger("google.cloud").setLevel(level=logging.ERROR)
    main(args)

Change 3: Move to RunConfig and have sane defaults

As per the above script, flows are automatically configured with defaults. Currently the following works well for us:

• We primarily use KubernetesRun that creates new Jobs for each flow run. We find this works well enough for most of our flows. For flows that need higher task concurrency we also have the option of using Dask (we have a long-living Dask cluster setup).
• We have a single base Docker image that contains all shared code and dependencies. This base image is used for the Prefect jobs (using a job config YAML similar to my original post above) and it is also used to run the Dask workers.
• Flows are stored in Google Cloud Storage (equivalent to S3).

The following file provides the defaults:

# sable_batch/build_projects.py

from os import path

from prefect.executors import DaskExecutor, LocalExecutor
from prefect.run_configs import KubernetesRun
from prefect.storage.gcs import GCS


def get_default_executor(dask=False):
    if dask:
        return DaskExecutor(address="tcp://dask-scheduler:8786")
    else:
        return LocalExecutor()


def get_default_run_config(
    labels: [str] = None, job_spec_filename="job_spec_default.yaml"
):
    job_template_path = path.join(
        path.dirname(path.realpath(__file__)), job_spec_filename
    )
    return KubernetesRun(
        job_template_path=job_template_path,
        image="gcr.io/xxx/sable_batch/prefect-base:latest",
        labels=labels,
    )


def get_default_storage():
    return GCS(bucket="xxx")

For flows that want to deviate from the defaults, developers can specify flow.run_config = ... or alternatively use the above methods:

from sable_batch.build_configs import get_default_run_config

with Flow("test flow") as flow:
  flow.run_config = get_default_run_config(labels=["dev"])
  flow.executor = get_default_executor(dask=True)

[zanieb]

Sweet! This seems to be approaching ideal :)

[ghost]

Really interesting! When I did a few tests trying to set the Executor outside the main .py of the flow, that Executor wasn't used for some reason. Could this be because of the storage? (I used GitHub)

[zanieb]

The executor isn't bundled with the serialized flow so I presume when you set it outside the flow file it's not set during execution because the flow loaded from storage does not know about your other file modifying the attribute. (I'm pretty sure)

[wrightjonathan]

We are seeing the same with the Executor. That is, if it is not included in the flow file, in storage, it does not take affect.

Setting the storage in this way (just before registration) does work for us. Can anyone confirm if this is also the case for the run_config. Thank you.

Update...

I can access the storage and run_config from graphQL but not the Executor. Which suggest that's the only one not persisted in the Prefect database.

[AlessandroLollo]

Overview

We want to make ETL easy for every developer, even for non-Data Engineers.
That's why we've invested time and effort in creating an internal standard for writing Prefect flows.
To support this standard, we've also built a little internal library that everyone can use to create the "skeleton" of an ETL flow based on Prefect.

On an high-level perspective, the dev -> deploy -> run flow works as follow:

• the developer runs a command-line utility to create the skeleton for the new ETL flow base on Prefect. The skeleton includes a minimal version of the flow + all other files needed for deployment to ECR/ECS
• the developer modifies the Prefect flow base on his needs/requirements.
• the developer runs a command-line utility to deploy the flow
• the developer runs the flow from Prefect Server UI

Project structure

Flows are organized into projects as described below

project
  |----flow
           |----Dockerfile
           |----src
                    |----etl.py
                    |----requirements.txt
                    |----dbt

This file structure is reflected on Prefect Server, meaning that once the flow is deployed, it will live inside project on Prefect Server.
Dockerfile contains all the steps needed to store the flow into a Docker image. This file is generated by the same command-line utility that generates the flow skeleton.
etl.py contains the definition of the Prefect flow. The last step in this file is flow.save()
requirements.txt contains all Python dependencies that are required to run the flow.
dbt contains all configs/models/sources/tests related to DBT (this is optional and can be included or not while using the command-line utility to create a new flow).

Developing flows

Developers modify etl.py to include everything they need based on their requirements.
To test a flow, the developer has to do the following:

• start a local Prefect Server.
• start a local Docker Agent.
• deploy the flow locally using a local-deploy script utility.
• run the flow using local Prefect Server.

Writing flows

Where are flows written? Do they share common patterns?
We have decided to store each flow into its own Docker images, which are hosted on ECR.
Given the project structure discussed above, all Docker images have the same structure.
The Dockerfile is defined as follow:

FROM prefect-flows-base-image:v1.548.0

# Arguments required to build the image
ARG pFlowName
ARG pImageName
ARG pImageTag
ARG pProjectName
ARG pEnvironment

ENV FLOW_NAME ${pFlowName}
ENV PROJECT_NAME ${pProjectName}
ENV IMAGE_NAME ${pImageName}
ENV IMAGE_TAG ${pImageTag}
ENV ENVIRONMENT ${pEnvironment}

# Set Prefect vars
ENV PREFECT__USER_BASE_PATH=/root/.prefect
ENV PREFECT__USER_CONFIG_PATH=${PREFECT__USER_BASE_PATH}/config.toml

# Copy Prefect config file
COPY common/config/config.toml ${PREFECT__USER_CONFIG_PATH}

# Copy register utility and Python requirements
COPY common/register /tmp/register

# Copy flow stuff
COPY flows/<project_name>/<flow_name>/src/ /tmp/flow/

# Upgrade pip
RUN pip install pip --upgrade

# Install all Python requirements
RUN pip install -r /tmp/flow/requirements.txt

# Install DBT packages
RUN cd /tmp/flow/dbt && \
    dbt deps

# Set PYTHONPATH
ENV PYTHONPATH="${PYTHONPATH}:/tmp/flow"

# Save Prefect flow to default folder
RUN python /tmp/flow/etl.py

This Dockerfile is used to build the image where the Prefect flow is stored.
prefect-flows-base-image:v1.548.0 is a custom, light-weight image we've built on top of python:3.8-slim that is used as a base image for all flows.
<project_name> is the project name (here we used just a placeholder)
<flow_name> is the flow name (here we used just a placeholder)

Testing flows

We do not have automatic tests yet.
Actually, we encourage developers to run them locally to ensure everything work as expected.

Deploying flows

Discovering flows in source, building storage, registering serialized flow with the backend
To make things easier for developers, we provide the following config.toml, which is included in every Docker image.

# We set home_dir like this to make storage build and flow.save() read/write to the same folder
home_dir = "/opt/prefect"

# Set backend to be Prefect Server
backend = "server"

# Set debug mode to false
debug = false

And in each Dockerfile we do the following:

# Set Prefect vars
ENV PREFECT__USER_BASE_PATH=/root/.prefect
ENV PREFECT__USER_CONFIG_PATH=${PREFECT__USER_BASE_PATH}/config.toml

# Copy Prefect config file
COPY common/config/config.toml ${PREFECT__USER_CONFIG_PATH}

This Dockerfile is used in our CI/CD process to create the flow storage.
We don't use Prefect built-in build() because we don't have Python installed on our CI/CD environment (Jenkins).

Registering flows

We have a little register.py script, which does the following:

# Load flow
flow = Flow.load(
    fpath=f"{prefect.context.config.home_dir}/flows/{slugify(FLOW_NAME)}.prefect"
)

# Register flow to Prefect Server
flow.register(
    project_name=PROJECT_NAME,
    build=False
)

Storing flows

We use Docker storage.
Each flow is store in its own Docker image, which is hosted on ECR.
We've decided to use Docker storage mainly because ECR is already widely used in our processes and it has been already integrated with our CI/CD process.

Running flows

We try to keep our flows as simple as possible and we encourage developers to use LocalEnvironment and/or LocalDaskExecutor.
Agents are deployed together with Prefect Server on ECS, so they are not part of the deployment process.
Developers are even not aware of where their flows run.

Run config

We are still using Prefect 0.13.13 so we don't use Run Configs yet.
However, we have plans to move to Prefect 0.14+ in the next few months.

Agent

We have a docker-compose that we use to spin up:

• Prefect Server as an ECS service
• Prefect UI as an ECS service
• Prefect Fargate Agent in EC2 mode as an ECS service
• Prefect Fargate Agent in Fargate mode as an ECS service

Environments

Flow dependencies are listed in a dedicated requirements.txt which is then used to install all dependencies during Docker image build.
Each developer decides what are the Python packages they need for their flow, and lists them accordingly in requirements.txt

Limitations

The biggest downside of this approach is that we are generating a lot of ECR repos/images.
Moreover, when we have to update the base image, we might need to rebuild all dependant images manually.

Benefits

The main benefit of this approach is that developers can focus on writing ETL, they do not need to care about deployment stuff.
The developer experience is quite good and the development process is fast.

[joshmeek]

Adding link to a post about a deployment pattern on Azure outlined in https://infinitelambda.com/post/prefect-workflow-automation-azure-devops-aks/ by @nikvakl

[zanieb]

See also, this guide on using GitHub actions: https://www.prefect.io/blog/deploying-prefect-flows-with-github-actions/