Realtime Compute for Apache Flink:Desarrollo de trabajos en Python

Elección entre Table API/SQL y DataStream API

PyFlink admite dos enfoques de desarrollo: Table API/SQL y DataStream API. Se recomienda Table API/SQL por las siguientes razones:

• Mejor rendimiento: El plan de ejecución optimizado de Table API/SQL se ejecuta completamente dentro de la JVM. En cambio, DataStream API requiere serialización y deserialización fila por fila entre la JVM y el proceso de Python, lo que introduce una sobrecarga de rendimiento significativa.

• Funcionalidades más completas: Table API/SQL ofrece un soporte más integral para conectores, formatos de datos y funciones de ventana, además de compartir el mismo ecosistema de conectores que los trabajos SQL.

• Recomendación de la comunidad: La comunidad de Apache Flink prioriza Table API/SQL como la dirección principal de desarrollo para PyFlink.

Utilice DataStream API únicamente para escenarios con lógica personalizada compleja que no pueda expresarse mediante SQL.

Referencias de desarrollo

Puede consultar los siguientes documentos para desarrollar código de negocio de Flink localmente. Una vez finalizado el desarrollo, cargue el código en la consola de desarrollo de Flink y despliegue el trabajo.

• Para el desarrollo de código de negocio en Apache Flink V1.20, consulte la Guía de desarrollo de la API de Python de Flink.

• Para problemas encontrados durante la codificación en Apache Flink y sus soluciones, consulte las Preguntas frecuentes.

Estructura del proyecto

La estructura recomendada para proyectos de Python es la siguiente:

my-flink-python-project/
├── my_job.py                # Main job file.
├── udfs.py                  # User-defined functions (optional)
├── requirements.txt         # Third-party Python dependencies (optional)
└── config.properties        # Configuration file (optional)

Gestión de dependencias

Para obtener información sobre cómo usar entornos virtuales de Python personalizados, paquetes de Python de terceros, archivos JAR y archivos de datos en trabajos de Python, consulte Use Python dependencies.

Funciones definidas por el usuario (UDF)

A continuación se presenta un ejemplo de desarrollo de una UDSF de Python que enmascara datos de cadenas sensibles:

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())def mask_phone(phone: str):"""Phone number masking: retains the first 3 and last 4 digits, replacing the middle with ****"""if phone is None or len(phone) != 11:return phone.
    return phone[:3] + '****' + phone[7:]

Utilice esta UDF en un trabajo SQL:

CREATE TEMPORARY FUNCTION mask_phone AS 'udfs.mask_phone' LANGUAGE PYTHON;INSERT INTO sink_table
SELECT name, mask_phone(phone) AS masked_phone
FROM source_table;

Para obtener información sobre cómo registrar, actualizar y eliminar UDF, consulte Manage UDFs.

Uso de conectores

Para consultar la lista de conectores compatibles con Flink, vea Supported connectors. Para usar un conector, realice los siguientes pasos:

1. Inicie sesión en la consola de Realtime Compute.

2. Haga clic en Console en la columna Actions del espacio de trabajo de destino.

3. En el panel de navegación izquierdo, haga clic en Artifacts.

4. Haga clic en Upload Artifact y seleccione el paquete de Python del conector de destino para cargarlo.

   Puede cargar un conector desarrollado por usted o uno proporcionado por Flink. Para descargar los paquetes oficiales de Python de los conectores proporcionados por Flink, consulte la Lista de conectores.

5. En la página O&M > Deployments, haga clic en Create Deployment > Python Deployment. Seleccione el paquete de Python del conector de destino en el campo Additional Dependencies, configure los demás parámetros y despliegue el trabajo.

6. Haga clic en el nombre del trabajo desplegado. En la pestaña Configuration, dentro de la sección Parameters, haga clic en Edit. En Other Configuration, agregue la información de ubicación de los paquetes de Python del conector.

   Si su trabajo depende de varios paquetes de Python de conectores, por ejemplo, dos paquetes llamados connector-1.jar y connector-2.jar, la configuración es la siguiente.

   pipeline.classpaths: 'file:///flink/usrlib/connector-1.jar;file:///flink/usrlib/connector-2.jar'

7. Para usar conectores integrados, formatos de datos y catálogos (solo en VVR 11.2 y posteriores), agregue la configuración en la sección Running Parameters bajo Additional Configuration del trabajo. Ejemplo:

   ## Multiple connectors used.
   pipeline.used-builtin-connectors: kafka;sls
   ## Multiple data formats for data transmission.
   pipeline.used-builtin-formats: avro;parquet
   ## Multiple previously created catalogs used.
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

Para más detalles sobre cómo usar conectores, consulte Complete sample code.

Depuración de trabajos

import logging

@udf(result_type=DataTypes.BIGINT())
def add(i, j):
  logging.info("hello world")
  return i + j

Depuración local

Dado que Realtime Compute for Apache Flink no tiene acceso a Internet de forma predeterminada, es posible que su código no pueda conectarse directamente a fuentes de datos en línea para realizar pruebas locales. Se recomiendan los siguientes enfoques para la depuración local:

• Pruebas unitarias: Realice pruebas unitarias independientes en las UDF para asegurar que la lógica de la función sea correcta.

• Ejecución local: Utilice fuentes de datos locales (como archivos o datos en memoria) para simular entradas y ejecute el trabajo localmente con el fin de verificar la lógica de procesamiento. Ejemplo:

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()# Use a local data source for testing.
  ds = env.from_collection([('Alice', 1), ('Bob', 2), ('Alice', 3)])
  ds.key_by(lambda x: x[0]).sum(1).print()
  env.execute("local_test")

• Depuración remota: Para depurar conectándose a fuentes de datos en línea, consulte Run and debug jobs with connectors locally.

Despliegue de trabajos

Una vez finalizado el desarrollo del trabajo de Python, cárguelo en la consola de Realtime Compute para su despliegue. Realice los siguientes pasos:

1. Inicie sesión en la consola de Realtime Compute y vaya al espacio de trabajo de destino.

2. En el panel de navegación izquierdo, haga clic en Artifacts y cargue el archivo del trabajo de Python (.py o .zip). Si existen dependencias de terceros o archivos de configuración, cárguelos también.

3. En la página O&M > Deployments, haga clic en Create Deployment > Python Deployment y complete la información de despliegue.

   Parámetro	Descripción
   Python File Path	Seleccione el archivo del trabajo de Python cargado.
   Entry Module	Si el archivo del trabajo es un archivo .py, este campo no es obligatorio. Si el archivo del trabajo es un archivo .zip, especifique el nombre del módulo de entrada, por ejemplo, my_job.
   Additional Dependencies	Seleccione archivos JAR de conectores o archivos de configuración si corresponde.
   Python Libraries	Seleccione paquetes de Python de terceros (.whl o .zip) si corresponde.
   Python Archives	Seleccione un entorno virtual de Python personalizado (.zip) si corresponde.

4. Haga clic en Deploy.

Para obtener más información sobre los parámetros de despliegue, consulte Deploy a job.

import logging
import sys

from pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

logging.basicConfig(stream=sys.stdout, level=logging.INFO)

def kafka_to_mysql():
    # Create the execution environment.
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)

    # Create the Kafka source table.
    t_env.execute_sql("""
        CREATE TABLE kafka_source (
            `id` INT,
            `name` STRING,
            `score` INT,
            `event_time` TIMESTAMP(3),
            WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
        ) WITH (
            'connector' = 'kafka',
            'topic' = 'student_topic',
            'properties.bootstrap.servers' = 'your-kafka-broker:9092',
            'properties.group.id' = 'my-group',
            'scan.startup.mode' = 'latest-offset',
            'format' = 'json'
        )
    """)

    # Create the MySQL sink table.
    t_env.execute_sql("""
        CREATE TABLE mysql_sink (
            `id` INT,
            `name` STRING,
            `score` INT,
            PRIMARY KEY (id) NOT ENFORCED
        ) WITH (
            'connector' = 'jdbc',
            'url' = 'jdbc:mysql://your-mysql-host:3306/my_database',
            'table-name' = 'student',
            'username' = 'your_username',
            'password' = 'your_password'
        )
    """)

    # Filter records with scores greater than 60 and write them to MySQL.
    t_env.execute_sql("""
        INSERT INTO mysql_sink
        SELECT id, name, score
        FROM kafka_source
        WHERE score >= 60
    """)

if __name__ == '__main__':
    kafka_to_mysql()