Aprende Orchestration Patterns in Python | Advanced Pipeline Patterns and Orchestration

Understanding orchestration is essential when building robust data pipelines. Orchestration refers to the coordination and management of tasks within a pipeline, ensuring that each step executes in the correct order and only when its dependencies are satisfied. In modern data engineering, orchestration tools like Apache Airflow use the concept of a directed acyclic graph (DAG) to model task dependencies. A DAG is a collection of tasks connected in such a way that there are no cycles, and each task runs only after its prerequisites have completed. This structure helps you manage complex workflows, prevent circular dependencies, and recover gracefully from failures.

When orchestrating ETL or ELT pipelines in Python, you often need to define the order in which extraction, transformation, and loading steps run. Each step may depend on the successful completion of previous steps. By modeling these dependencies explicitly, you can automate the execution flow, improve reliability, and make your pipelines easier to maintain or extend.

To illustrate these ideas, you can implement a simplified workflow manager in Python that mimics some orchestration patterns found in tools like Airflow. This manager will allow you to define individual ETL tasks, specify their dependencies, and execute them in the required order. By simulating a DAG-based approach, you gain hands-on experience with the building blocks of orchestration, even before adopting a full-featured workflow tool.


              1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
            
from collections import defaultdict, deque

# Define a simple Task class
class Task:
    def __init__(self, name, func, dependencies=None):
        self.name = name
        self.func = func
        self.dependencies = dependencies or []

    def run(self):
        print(f"Running task: {self.name}")
        self.func()

# Basic workflow manager to handle task orchestration
class WorkflowManager:
    def __init__(self):
        self.tasks = {}
        self.dependency_graph = defaultdict(list)
        self.in_degree = defaultdict(int)

    def add_task(self, task):
        self.tasks[task.name] = task
        for dep in task.dependencies:
            self.dependency_graph[dep].append(task.name)
            self.in_degree[task.name] += 1
        if task.name not in self.in_degree:
            self.in_degree[task.name] = 0

    def run(self):
        # Topological sort (Kahn's algorithm)
        queue = deque([name for name, deg in self.in_degree.items() if deg == 0])
        executed = set()
        while queue:
            name = queue.popleft()
            task = self.tasks[name]
            task.run()
            executed.add(name)
            for neighbor in self.dependency_graph[name]:
                self.in_degree[neighbor] -= 1
                if self.in_degree[neighbor] == 0:
                    queue.append(neighbor)
        if len(executed) != len(self.tasks):
            print("Cycle detected or missing dependencies!")

# Example ETL step functions
def extract():
    print("Extracting data...")

def transform():
    print("Transforming data...")

def load():
    print("Loading data...")

# Define tasks with dependencies
extract_task = Task("extract", extract)
transform_task = Task("transform", transform, dependencies=["extract"])
load_task = Task("load", load, dependencies=["transform"])

# Build and run the workflow
manager = WorkflowManager()
manager.add_task(extract_task)
manager.add_task(transform_task)
manager.add_task(load_task)
manager.run()

¿Todo estuvo claro?

¡Gracias por tus comentarios!

Sección 4. Capítulo 4

Pregunte a AI

Pregunte lo que quiera o pruebe una de las preguntas sugeridas para comenzar nuestra charla

Desliza para mostrar el menú


              1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
            
from collections import defaultdict, deque

# Define a simple Task class
class Task:
    def __init__(self, name, func, dependencies=None):
        self.name = name
        self.func = func
        self.dependencies = dependencies or []

    def run(self):
        print(f"Running task: {self.name}")
        self.func()

# Basic workflow manager to handle task orchestration
class WorkflowManager:
    def __init__(self):
        self.tasks = {}
        self.dependency_graph = defaultdict(list)
        self.in_degree = defaultdict(int)

    def add_task(self, task):
        self.tasks[task.name] = task
        for dep in task.dependencies:
            self.dependency_graph[dep].append(task.name)
            self.in_degree[task.name] += 1
        if task.name not in self.in_degree:
            self.in_degree[task.name] = 0

    def run(self):
        # Topological sort (Kahn's algorithm)
        queue = deque([name for name, deg in self.in_degree.items() if deg == 0])
        executed = set()
        while queue:
            name = queue.popleft()
            task = self.tasks[name]
            task.run()
            executed.add(name)
            for neighbor in self.dependency_graph[name]:
                self.in_degree[neighbor] -= 1
                if self.in_degree[neighbor] == 0:
                    queue.append(neighbor)
        if len(executed) != len(self.tasks):
            print("Cycle detected or missing dependencies!")

# Example ETL step functions
def extract():
    print("Extracting data...")

def transform():
    print("Transforming data...")

def load():
    print("Loading data...")

# Define tasks with dependencies
extract_task = Task("extract", extract)
transform_task = Task("transform", transform, dependencies=["extract"])
load_task = Task("load", load, dependencies=["transform"])

# Build and run the workflow
manager = WorkflowManager()
manager.add_task(extract_task)
manager.add_task(transform_task)
manager.add_task(load_task)
manager.run()

¿Todo estuvo claro?

¡Gracias por tus comentarios!

Sección 4. Capítulo 4