Flow deployment patterns #4042
Replies: 7 comments 15 replies
-
|
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
Project structureDeveloping flowsWriting flowsFlows are written in their own files within the Flows define their own executor but storage and run configs are set by the CI process. Testing flowsFlows are tested locally using Deploying flowsRegistering flowsFlows are registered in CI by iterating through every file in the Here we will set the storage and run_config to be consistent across all flows. Storing flowsFlows are stored on a remote backend of your choice e.g. S3/GCS. Running flowsRun configThe run config for this project is a AgentUses a docker agent. The agent could be run locally or on a single cloud node. EnvironmentsDependencies are managed by Limitations
Benefits
|
Beta Was this translation helpful? Give feedback.
-
|
Could you provide a snippet to the flow registration? How do I import the flow variable when using context manager?
Thank you Edit: Find a solution by myself and like to share it with others: |
Beta Was this translation helpful? Give feedback.
-
|
Lerning never stops... better approach was to simply use the build-in utilities function prefect/src/prefect/utilities/storage.py Line 53 in f1bf35b Somehow it's missing at the doc's page. |
Beta Was this translation helpful? Give feedback.
-
|
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 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.pyimport 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() |
Beta Was this translation helpful? Give feedback.
-
|
Thanks for starting this conversation, @madkinsz ! Here is my first crack at this: OverviewI am not finished building this out, so some items are "how I intend to do it" as opposed to "what I've already done".
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 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 structureHere is my basic structure. I am omitting common code and other unrelated items. The Developing flowsFlows go in the Writing flowsTBD Testing flowsTBD Deploying flowsDiscovering flows in source, building storage, registering serialized flow with the backend Registering flowsI have a designated directory called 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:
If a given flow has a run configuration already defined, the build process respects that. If not, it adds default configuration:
Finally, registration is called:
Storing flowsWe 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 flowsI 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 configSo far, exclusively using AgentI 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. EnvironmentsI 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
Benefits
|
Beta Was this translation helpful? Give feedback.
-
|
Does using the |
Beta Was this translation helpful? Give feedback.
-
|
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!? |
Beta Was this translation helpful? Give feedback.
-
|
Although not completely sure how that would fit together... |
Beta Was this translation helpful? Give feedback.
-
|
@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. |
Beta Was this translation helpful? Give feedback.
-
|
@josh-gree If you use a |
Beta Was this translation helpful? Give feedback.
-
OverviewThis 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.
Project structureFlows are a module within the broader package: Developing & Writing flowsThe dev’s job is to implement the method that contains the actual flow logic, define the flow parameters, and specify environment configs values. Testing flowsWe plan on using Deploying/Registering flowsWe have a manual way to indicate within the package if a flow is "active". Our CLI tool calls Storing flowsWe 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 Running flowsWe have an agent running in EKS. Our Future PlansWe 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
Benefits
|
Beta Was this translation helpful? Give feedback.
-
|
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 :) |
Beta Was this translation helpful? Give feedback.
-
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.
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. |
Beta Was this translation helpful? Give feedback.
-
OverviewOur approach is as follows:
Project structureDeveloping flowsWriting a flow, testing a flow We currently keep flows very simple. Here is a very minimal example: We use We store tests in the Writing flowsFlows are all stored in the Testing flowsSee example above. On each PR we run: Deploying flowsOur CI process executes a Some of the other elements referenced above are the Dockerfile: And our setup.py file: And if anyone is using CircleCI, this is what our Registering flowsSee above Storing flowsDuring 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 Running flowsDuring We run the Prefect agent using a standard k8s deployment.yaml file. The only notable thing is that we use Run configSee above AgentStandard Kubernetes Deployment EnvironmentsSee above Limitations
Benefits
Future plansWe have lots of improvements we'd like to make:
|
Beta Was this translation helpful? Give feedback.
-
|
We've now made a lot of new improvements: Change 1: Use folders for Prefect projects.Our tree now looks like: Change 2: Auto detect flows and registerThe following script walks the directory # 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 defaultsAs per the above script, flows are automatically configured with defaults. Currently the following works well for us:
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 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) |
Beta Was this translation helpful? Give feedback.
-
|
Sweet! This seems to be approaching ideal :) |
Beta Was this translation helpful? Give feedback.
-
|
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) |
Beta Was this translation helpful? Give feedback.
-
|
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) |
Beta Was this translation helpful? Give feedback.
-
|
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. |
Beta Was this translation helpful? Give feedback.
-
OverviewWe want to make ETL easy for every developer, even for non-Data Engineers. On an high-level perspective, the dev -> deploy -> run flow works as follow:
Project structureFlows are organized into projects as described below This file structure is reflected on Prefect Server, meaning that once the Developing flowsDevelopers modify
Writing flowsWhere are flows written? Do they share common patterns? This Dockerfile is used to build the image where the Prefect flow is stored. Testing flowsWe do not have automatic tests yet. Deploying flowsDiscovering flows in source, building storage, registering serialized flow with the backend And in each Dockerfile we do the following: This Dockerfile is used in our CI/CD process to create the flow storage. Registering flowsWe have a little Storing flowsWe use Docker storage. Running flowsWe try to keep our flows as simple as possible and we encourage developers to use Run configWe are still using Prefect AgentWe have a
EnvironmentsFlow dependencies are listed in a dedicated LimitationsThe biggest downside of this approach is that we are generating a lot of ECR repos/images. BenefitsThe main benefit of this approach is that developers can focus on writing ETL, they do not need to care about deployment stuff. |
Beta Was this translation helpful? Give feedback.
-
|
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 |
Beta Was this translation helpful? Give feedback.
-
|
See also, this guide on using GitHub actions: https://www.prefect.io/blog/deploying-prefect-flows-with-github-actions/ |
Beta Was this translation helpful? Give feedback.
-
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:
(this template may be updated after a few examples are in here)
Beta Was this translation helpful? Give feedback.
All reactions