Instalación de un paquete privado como requisito en el trabajo de Apache Airflow

Nota:

El trabajo de Apache Airflow cuenta con la tecnología de Apache Airflow.
Apache Airflow es una plataforma de código abierto que se usa para crear, programar y supervisar flujos de trabajo de datos complejos mediante programación. Permite definir un conjunto de tareas, denominadas operadores, que se pueden combinar en grafos acíclicos dirigidos (DAG) para representar canalizaciones de datos.

Un paquete de Python es una manera de organizar los módulos de Python relacionados en una sola jerarquía de directorios. Normalmente, un paquete se representa como un directorio que contiene un archivo especial denominado init.py. Dentro de un directorio de paquetes, se pueden tener varios archivos de módulos de Python (archivos.py) que definen funciones, clases y variables. En el contexto del trabajo de Apache Airflow, puede desarrollar paquetes privados para agregar operadores personalizados de Apache Airflow, enlaces, sensores, complementos, etc.

En este tutorial, creará un operador personalizado sencillo como paquete de Python, lo agregará como requisito en el entorno de trabajo de Apache Airflow e importará el paquete privado como módulo dentro del archivo DAG.

Desarrollo de un operador personalizado y prueba con un dag de Apache Airflow

1. Cree un archivo sample_operator.py y conviértalo a paquete privado. Consulte la guía: Creación de un paquete en Python

   from airflow.models.baseoperator import BaseOperator
   
   
   class SampleOperator(BaseOperator):
       def __init__(self, name: str, **kwargs) -> None:
           super().__init__(**kwargs)
           self.name = name
   
       def execute(self, context):
           message = f"Hello {self.name}"
           return message
   
   

2. Cree el archivo sample_dag.py DAG de Apache Airflow para probar el operador definido en el Paso 1.

   from datetime import datetime
   from airflow import DAG
   
    # Import from private package
   from airflow_operator.sample_operator import SampleOperator
   
   
   with DAG(
   "test-custom-package",
   tags=["example"]
   description="A simple tutorial DAG",
   schedule_interval=None,
   start_date=datetime(2021, 1, 1),
   ) as dag:
       task = SampleOperator(task_id="sample-task", name="foo_bar")
   
       task
   

3. Cree un repositorio de GitHub que contenga el sample_dag.py en la carpeta Dags y el archivo de paquete privado. Los formatos de archivo más comunes son zip, .whl y tar.gz. Coloque el archivo en la carpeta Dags o Plugins, según corresponda. Sincronizar el repositorio de Git con el trabajo de Apache Airflow o puede usar el repositoriopreconfigurado Install-Private-Package

Agregar el paquete privado como requisito

Agregue el paquete como requisito en Airflow requirements. Utilice el formato /opt/airflow/git/<repoName>.git/<pathToPrivatePackage>

Por ejemplo, si el paquete privado está ubicado en /dags/test/private.whl en un repositorio de GitHub, agregue el requisito /opt/airflow/git/<repoName>.git/dags/test/private.whl al entorno Airflow.

Contenido relacionado

Inicio rápido: Creación de un trabajo de Apache Airflow