Realtime Compute for Apache Flink:Desenvolver jobs Python

Escolha entre Table API/SQL e DataStream API

O PyFlink oferece duas abordagens de desenvolvimento: Table API/SQL e DataStream API. Recomendamos a Table API/SQL pelos motivos a seguir:

• Melhor desempenho: O plano de execução otimizado da Table API/SQL roda inteiramente dentro da JVM. Já a DataStream API exige serialização e desserialização linha a linha entre a JVM e o processo Python, o que gera sobrecarga significativa de desempenho.

• Recursos mais abrangentes: A Table API/SQL oferece suporte mais completo a conectores, formatos de dados e funções de janela, compartilhando o mesmo ecossistema de conectores dos jobs SQL.

• Recomendação da comunidade: A comunidade Apache Flink prioriza a Table API/SQL como principal direção de desenvolvimento para o PyFlink.

Use a DataStream API apenas em cenários com lógica personalizada complexa impossível de expressar via SQL.

Referências de desenvolvimento

Consulte os documentos abaixo para desenvolver o código de negócio do Flink localmente. Depois de finalizar o desenvolvimento, carregue o código no console de desenvolvimento do Flink e implante o job.

• Para desenvolver código de negócio no Apache Flink V1.20, acesse o Guia de Desenvolvimento da API Python do Flink.

• Caso enfrente problemas durante a codificação no Apache Flink, consulte o FAQ para encontrar soluções.

Estrutura do projeto

A estrutura recomendada para projetos de jobs Python é a seguinte:

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)

Gerenciamento de dependências

Para saber como utilizar ambientes virtuais Python personalizados, pacotes Python de terceiros, arquivos JAR e arquivos de dados em jobs Python, consulte Use Python dependencies.

Funções definidas pelo usuário (UDFs)

Veja abaixo um exemplo de desenvolvimento de uma UDSF em Python que mascara dados sensíveis em strings:

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:]

Para usar essa UDF em um job 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 obter informações sobre como registrar, atualizar e excluir UDFs, consulte Manage UDFs.

Uso de conectores

Para ver a lista de conectores suportados pelo Flink, consulte Supported connectors. Siga as etapas abaixo para utilizar um conector:

1. Faça login no console do Realtime Compute.

2. Clique em Console na coluna Actions do workspace desejado.

3. No painel de navegação à esquerda, clique em Artifacts.

4. Clique em Upload Artifact e selecione o pacote Python do conector alvo para carregar.

   É possível carregar um conector desenvolvido internamente ou um fornecido pelo Flink. Para baixar os pacotes Python oficiais dos conectores do Flink, acesse a Lista de conectores.

5. Na página O&M > Deployments, clique em Create Deployment > Python Deployment. Selecione o pacote Python do conector desejado no campo Additional Dependencies, configure os demais parâmetros e implante o job.

6. Clique em nome do job implantado. Na aba Configuration, na seção Parameters, clique em Edit. Em Other Configuration, adicione as informações de localização dos pacotes Python dos conectores.

   Se o seu job depender de vários pacotes Python de conectores, por exemplo, dois pacotes chamados connector-1.jar e connector-2.jar, use a seguinte configuração:

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

7. Para utilizar conectores integrados, formatos de dados e catálogos (apenas na VVR 11.2 e posteriores), adicione a configuração na seção Running Parameters, dentro de Additional Configuration do job. Exemplo:

   ## 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 mais detalhes sobre o uso de conectores, consulte Complete sample code.

Depuração de jobs

import logging

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

Depuração local

Como o Realtime Compute for Apache Flink não tem acesso à Internet por padrão, seu código pode não conseguir se conectar diretamente a fontes de dados online para testes locais. Recomendamos as seguintes abordagens para depuração local:

• Testes unitários: Execute testes unitários independentes nas UDFs para garantir a correção da lógica da função.

• Execução local: Utilize fontes de dados locais (como arquivos ou dados em memória) para simular entradas e execute o job localmente, validando a lógica de processamento. Exemplo:

  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")

• Depuração remota: Para depurar conectando-se a fontes de dados online, consulte Run and debug jobs with connectors locally.

Implantação de jobs

Após concluir o desenvolvimento do job Python, carregue-o no console do Realtime Compute para implantação. Siga estas etapas:

1. Faça login no console do Realtime Compute e acesse o workspace desejado.

2. No painel de navegação à esquerda, clique em Artifacts e carregue o arquivo do job Python (.py ou .zip). Se houver dependências de terceiros ou arquivos de configuração, carregue-os também.

3. Na página O&M > Deployments, clique em Create Deployment > Python Deployment e preencha as informações da implantação.

   Parâmetro	Descrição
   Python File Path	Selecione o arquivo do job Python carregado.
   Entry Module	Campo opcional para arquivos .py. Para arquivos .zip, especifique o nome do módulo de entrada, por exemplo, my_job.
   Additional Dependencies	Selecione arquivos JAR de conectores ou arquivos de configuração, se aplicável.
   Python Libraries	Escolha pacotes Python de terceiros (.whl ou .zip), quando necessário.
   Python Archives	Selecione um ambiente virtual Python personalizado (.zip), se for o caso.

4. Clique em Deploy.

Para mais informações sobre os parâmetros de implantação, 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()