Develop Flink Python Jobs - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Este tópico descreve as informações de contexto, limites, métodos de desenvolvimento, métodos de depuração e o uso de conectores para jobs da API Python do Flink.

Informações de contexto

Você deve desenvolver jobs em Python do Flink localmente. Após concluir o desenvolvimento, implante e inicie os jobs no console de desenvolvimento do Flink para visualizar os resultados do negócio. Para obter mais informações sobre o procedimento geral, consulte {{XREF_0}}.

Requisitos do ambiente de desenvolvimento

As versões do mecanismo do Realtime Compute for Apache Flink anteriores à VVR 8.0.11 vêm com o Python 3.7.9 pré-instalado. A VVR 8.0.11 e versões posteriores vêm com o Python 3.9.21 pré-instalado. A VVR 11.7 e versões posteriores vêm com o Python 3.9.21, 3.10.19 e 3.11.14 pré-instalados.

Nota

Recomendamos que a versão do Python no seu ambiente de desenvolvimento local corresponda à versão do Python pré-instalada no mecanismo VVR de destino.

Nota

A VVR 11.7 e versões posteriores usam o Python 3.9 por padrão. Se você quiser usar o Python 3.10 ou 3.11 pré-instalado, configure o job da seguinte forma:

Na página O&M > Deployments, clique no nome do job de destino.

Na guia Configuration, na seção Parameters, clique em Edit no lado direito e adicione a seguinte configuração no campo Other Configuration (usando o Python 3.11 como exemplo):

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

O PyFlink está instalado e sua versão corresponde à versão do mecanismo VVR de destino. Por exemplo, se o mecanismo selecionado na página de implantação for vvr-11.7.0-jdk11-flink-1.20, você precisará instalar:

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

Nota

A VVR 11.5 e versões anteriores não fornecem um pacote PyFlink dedicado. Você precisa instalar a versão correspondente do PyFlink de código aberto. Por exemplo, se o mecanismo selecionado na página de implantação for vvr-8.0.9-flink-1.17, você precisará instalar apache-flink==1.17.*.

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

Uma IDE deve estar instalada. Recomendamos o PyCharm ou o VS Code.
Você deve desenvolver os jobs em Python localmente e, em seguida, implantá-los e executá-los no console do Realtime Compute for Apache Flink.

Limites

Devido ao impacto dos ambientes de implantação e de rede no Flink, observe os seguintes limites ao desenvolver jobs em Python:

Há suporte apenas para o Flink de código aberto V1.13 e versões posteriores.
O workspace do Flink vem com um ambiente Python pré-instalado, que inclui bibliotecas Python comumente usadas, como Pandas, NumPy e PyArrow. Para obter detalhes, consulte {{XREF_1}} no final deste tópico.
O ambiente de execução do Flink oferece suporte apenas ao JDK 8 e ao JDK 11. Se o seu job em Python depender de pacotes JAR de terceiros, certifique-se de que os pacotes JAR sejam compatíveis.
A VVR 4.x oferece suporte apenas ao Scala de código aberto V2.11. A VVR 6.x e versões posteriores oferecem suporte apenas ao Scala de código aberto V2.12. Se o seu job em Python depender de pacotes JAR de terceiros, certifique-se de que as dependências do pacote JAR correspondam à versão do Scala aplicável.

Desenvolvimento do job

Escolha entre Table API/SQL e DataStream API

O PyFlink oferece suporte a duas abordagens de desenvolvimento: Table API/SQL e DataStream API. Recomenda-se a Table API/SQL pelos seguintes motivos:

Melhor desempenho: O plano de execução otimizado da Table API/SQL é executado inteiramente na JVM. Em contrapartida, a DataStream API exige serialização e desserialização linha a linha entre a JVM e o processo Python, o que gera uma sobrecarga de desempenho significativa.
Recursos mais abrangentes: A Table API/SQL fornece um suporte mais completo para conectores, formatos de dados e funções de janela, além de compartilhar o mesmo ecossistema de conectores dos jobs SQL.
Recomendação da comunidade: A comunidade Apache Flink prioriza a Table API/SQL como a principal direção de desenvolvimento para o PyFlink.

Use a DataStream API apenas para cenários de lógica personalizada complexa que não possam ser expressos em SQL.

Referências de desenvolvimento

Você pode consultar os seguintes documentos para desenvolver o código de negócio do Flink localmente. Após concluir o desenvolvimento, faça o upload do código para o console de desenvolvimento do Flink e implante o job.

Para o desenvolvimento de código de negócio do Apache Flink V1.20, consulte o Guia de desenvolvimento da API Python do Flink.
Para problemas encontrados durante a codificação do Apache Flink e suas respectivas soluções, consulte o FAQ.

Estrutura do projeto

A estrutura de projeto recomendada para jobs em Python é a seguinte:

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

Gerenciamento de dependências

Para obter informações sobre como usar ambientes virtuais Python personalizados, pacotes Python de terceiros, arquivos JAR e arquivos de dados em jobs Python, consulte {{XREF_2}}.

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

A seguir, um exemplo de desenvolvimento de uma UDSF em Python que mascara dados de string confidenciais:

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

Use esta UDF em um job SQL:

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

Para obter informações sobre como registrar, atualizar e excluir UDFs, consulte {{XREF_3}}.

Uso de conectores

Para ver a lista de conectores compatíveis com o Flink, consulte {{XREF_4}}. Para usar um conector, execute as seguintes etapas:

Faça logon no console do Realtime Compute.
Clique em Console na coluna Actions do workspace de destino.
No painel de navegação à esquerda, clique em Artifacts.
Clique em Upload Artifact e selecione o pacote Python do conector de destino para fazer o upload.

Você pode fazer o upload de um conector desenvolvido internamente ou de um conector fornecido pelo Flink. Para baixar os pacotes Python oficiais dos conectores fornecidos pelo Flink, consulte a Lista de conectores.
Na página O&M > Deployments, clique em Create Deployment > Python Deployment. Selecione o pacote Python do conector de destino no campo Additional Dependencies, configure os outros parâmetros e implante o job.

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

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

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

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

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

Para obter detalhes sobre como usar conectores, consulte {{XREF_5}}.

Depuração de jobs

Você pode usar logging na implementação de código de UDFs Python para gerar informações de log para solução de problemas. Exemplo:

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

Após a geração do log, você poderá visualizar os logs nos arquivos de log do TaskManager.

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 de unidade: Execute testes de unidade independentes nas UDFs para garantir que a lógica da função esteja correta.

Execução local: Use fontes de dados locais (como arquivos ou dados na memória) para simular entradas e execute o job localmente para verificar a lógica de processamento. Exemplo:

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

Depuração remota: Para depurar conectando-se a fontes de dados online, consulte {{XREF_6}}.

Implantação do job

Após concluir o desenvolvimento do job em Python, faça o upload do job para o console do Realtime Compute para implantação. Execute as seguintes etapas:

Faça logon no console do Realtime Compute e acesse o workspace de destino.
No painel de navegação à esquerda, clique em Artifacts e faça o upload do arquivo de job Python (.py ou .zip). Se houver dependências de terceiros ou arquivos de configuração, faça o upload deles também.

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

Parâmetro	Descrição
Python File Path	Selecione o arquivo de job Python carregado.
Entry Module	Se o arquivo de job for um arquivo .py, este campo não é obrigatório. Se o arquivo de job for um arquivo .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	Selecione pacotes Python de terceiros (.whl ou .zip), se aplicável.
Python Archives	Selecione um ambiente virtual Python personalizado (.zip), se aplicável.

Clique em Deploy.

Para obter mais informações sobre os parâmetros de implantação, consulte {{XREF_7}}.

Código de exemplo completo

Este exemplo demonstra um job de streaming em Python que lê dados do Kafka, executa um processamento simples e grava os resultados no MySQL. Isso serve apenas como referência.

Nota

Este exemplo não inclui a configuração de parâmetros de execução, como checkpoints e estratégias de reinicialização. Você pode personalizar essas configurações na página Deployment Details após implantar o job. Para obter mais informações, consulte {{XREF_8}}.

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

Pacotes pré-instalados

VVR-11

Os seguintes pacotes estão pré-instalados no workspace do Flink.

Pacote	Versão
apache-beam	2.48.0
avro-python3	1.10.2
brotlipy	0.7.0
certifi	2022.12.7
cffi	1.15.1
charset-normalizer	2.0.4
cloudpickle	2.2.1
conda	22.11.1
conda-content-trust	0.1.3
conda-package-handling	1.9.0
crcmod	1.7
cryptography	38.0.1
Cython	3.0.12
dill	0.3.1.1
dnspython	2.7.0
docopt	0.6.2
exceptiongroup	1.3.0
fastavro	1.12.1
fasteners	0.20
find_libpython	0.5.0
grpcio	1.56.2
grpcio-tools	1.56.2
hdfs	2.7.3
httplib2	0.22.0
idna	3.4
importlib_metadata	8.7.0
iniconfig	2.1.0
isort	6.1.0
numpy	1.24.4
objsize	0.6.1
orjson	3.9.15
packaging	25.0
pandas	2.3.3
pemja	0.5.5
pip	22.3.1
pluggy	1.0.0
proto-plus	1.26.1
protobuf	4.25.8
py-spy	0.4.0
py4j	0.10.9.7
pyarrow	11.0.0
pyarrow-hotfix	0.6
pycodestyle	2.14.0
pycosat	0.6.4
pycparser	2.21
pydot	1.4.2
pymongo	4.15.4
pyOpenSSL	22.0.0
pyparsing	3.2.5
PySocks	1.7.1
pytest	7.4.4
python-dateutil	2.9.0
pytz	2025.2
regex	2025.11.3
requests	2.32.5
ruamel.yaml	0.18.16
ruamel.yaml.clib	0.2.14
setuptools	70.0.0
six	1.16.0
tomli	2.3.0
toolz	0.12.0
tqdm	4.64.1
typing_extensions	4.15.0
tzdata	2025.2
urllib3	1.26.13
wheel	0.38.4
zipp	3.23.0
zstandard	0.25.0
torch	2.5.1
torchvision	0.20.1
transformers	4.57.6
opencv-python-headless	4.10.0.84
pillow	11.3.0
ultralytics	8.4.66
easyocr	1.7.2
open-clip-torch	2.32.0
rembg	2.0.61
onnxruntime	1.16.3
av	14.2.0
librosa	0.11.0
soundfile	0.13.1
imagehash	4.3.2
safetensors	0.7.0
huggingface-hub	0.36.2
tokenizers	0.22.2
timm	1.0.27
scikit-learn	1.6.1
scikit-image	0.24.0
scipy	1.13.1
numba	0.60.0
llvmlite	0.43.0
matplotlib	3.9.4
pywavelets	1.6.0
tifffile	2024.8.30
shapely	2.0.7
pymatting	1.1.15
audioread	3.1.0
soxr	0.5.0.post1
joblib	1.5.3
threadpoolctl	3.6.0
sympy	1.13.1
mpmath	1.3.0
python-bidi	0.6.10
filelock	3.19.1
pyyaml	6.0.3
decorator	5.3.1
msgpack	1.1.2
ftfy	6.3.1
wcwidth	0.8.1
contourpy	1.3.0
cycler	0.12.1
fonttools	4.60.2
importlib-resources	5.4.0
kiwisolver	1.4.7
lazy-loader	0.5
attrs	26.1.0
jsonschema	4.25.1
jsonschema-specifications	2025.9.1
referencing	0.36.2
rpds-py	0.27.1
pooch	1.9.0
platformdirs	4.4.0

VVR-8

Os seguintes pacotes estão pré-instalados no workspace do Flink.

Pacote	Versão
apache-beam	2.43.0
avro-python3	1.9.2.1
certifi	2025.7.9
charset-normalizer	3.4.2
cloudpickle	2.2.0
crcmod	1.7
Cython	0.29.24
dill	0.3.1.1
docopt	0.6.2
fastavro	1.4.7
fasteners	0.19
find_libpython	0.4.1
grpcio	1.46.3
grpcio-tools	1.46.3
hdfs	2.7.3
httplib2	0.20.4
idna	3.10
isort	6.0.1
numpy	1.21.6
objsize	0.5.2
orjson	3.10.18
pandas	1.3.5
pemja	0.3.2
pip	22.3.1
proto-plus	1.26.1
protobuf	3.20.3
py4j	0.10.9.7
pyarrow	8.0.0
pycodestyle	2.14.0
pydot	1.4.2
pymongo	3.13.0
pyparsing	3.2.3
python-dateutil	2.9.0
pytz	2025.2
regex	2024.11.6
requests	2.32.4
setuptools	58.1.0
six	1.17.0
typing_extensions	4.14.1
urllib3	2.5.0
wheel	0.33.4
zstandard	0.23.0

VVR-6

Os seguintes pacotes estão pré-instalados no workspace do Flink.

Pacote	Versão
apache-beam	2.27.0
avro-python3	1.9.2.1
certifi	2024.8.30
charset-normalizer	3.3.2
cloudpickle	1.2.2
crcmod	1.7
Cython	0.29.16
dill	0.3.1.1
docopt	0.6.2
fastavro	0.23.6
future	0.18.3
grpcio	1.29.0
hdfs	2.7.3
httplib2	0.17.4
idna	3.8
importlib-metadata	6.7.0
isort	5.11.5
jsonpickle	2.0.0
mock	2.0.0
numpy	1.19.5
oauth2client	4.1.3
pandas	1.1.5
pbr	6.1.0
pemja	0.1.4
pip	20.1.1
protobuf	3.17.3
py4j	0.10.9.3
pyarrow	2.0.0
pyasn1	0.5.1
pyasn1-modules	0.3.0
pycodestyle	2.10.0
pydot	1.4.2
pymongo	3.13.0
pyparsing	3.1.4
python-dateutil	2.8.0
pytz	2024.1
requests	2.31.0
rsa	4.9
setuptools	47.1.0
six	1.16.0
typing-extensions	3.7.4.3
urllib3	2.0.7
wheel	0.42.0
zipp	3.15.0

Referências

Para um exemplo completo de fluxo de trabalho de desenvolvimento de jobs Python do Flink, consulte {{XREF_9}}.
Para obter informações sobre como usar ambientes virtuais Python personalizados, pacotes Python de terceiros, arquivos JAR e arquivos de dados em jobs Python do Flink, consulte {{XREF_10}}.
O Realtime Compute for Apache Flink também oferece suporte a jobs SQL e DataStream. Para obter mais informações, consulte {{XREF_11}} e {{XREF_12}}.