Develop jobs with Flink Python Coding Skill - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Visão geral

O Flink Python Coding Skill é uma ferramenta de codificação assistida por IA fornecida pelo Realtime Compute for Apache Flink. Integrado como plug-in aos principais assistentes de codificação com IA, ele injeta o sistema de conhecimento de desenvolvimento PyFlink no agente de IA e gera código de jobs Flink em Python com base em descrições em linguagem natural.

Essa skill abrange toda a cadeia de desenvolvimento PyFlink, incluindo inicialização do ambiente, acesso a fontes de dados, seleção de operadores, processamento de dados, escrita de resultados e preparação para implantação. O foco na API DataFrame e nas capacidades de processamento multimodal de IA reduz a barreira de entrada para o desenvolvimento de jobs de computação em tempo real e pré-processamento de dados para IA.

A tabela a seguir descreve os principais recursos.

Recurso	Descrição
Desenvolvimento orientado por linguagem natural	Descreva um requisito em uma única frase. A skill seleciona APIs, configura parâmetros e gera o código automaticamente, sem necessidade de memorizar APIs complexas.
Verificação de validade da API	Uma biblioteca integrada de documentação oficial da API PyFlink e scripts de verificação de operadores conferem em tempo real se uma API existe, evitando a geração de interfaces inexistentes.
Padrões de código para produção	Configurações de checkpoint, paralelismo, tipo de retorno de UDF e provedor de modelo são concluídas automaticamente. O código gerado pode ser implantado diretamente.
Capacidades multimodais integradas	Suporta nativamente a família de operadores de imagem, `llm.predict` e `llm.ai_embed`. É possível executar compreensão de imagens, geração de texto e vetorização diretamente em jobs de streaming e batch.
Integração multiplataforma	Compatível com os principais assistentes de codificação com IA, como Claude Code, Cursor, Codex, Gemini CLI, Tongyi Lingma e Qoder. A instalação requer apenas um único comando.

O AI Intelligent Assistant no console também oferece uma skill integrada de desenvolvimento de jobs que suporta geração e validação de SQL, criação de rascunhos e implantação. Para obter mais informações, consulte Intelligent Assistant.

Arquitetura

O Flink Python Coding Skill utiliza uma arquitetura de injeção de conhecimento em múltiplas camadas. Ele injeta sistematicamente todo o sistema de conhecimento de desenvolvimento PyFlink no agente de IA, garantindo que o código gerado seja preciso, executável e pronto para implantação.

A tabela abaixo detalha a função de cada componente.

Componente	Função
Definição da skill de codificação	Define as responsabilidades principais, os limites de capacidade e o fluxo de trabalho da skill.
Guias de contexto	Documentos de referência que abrangem PyFlink DataFrame, Table, DataStream, AI Function e operadores multimodais.
Agente de seleção de API e operador	Agente inteligente que descobre, verifica e recomenda APIs e operadores.
Mecanismo de regras de seleção	Políticas de seleção baseadas em princípios como prioridade ao desempenho, prioridade ao processamento de streaming e prioridade à compatibilidade.
Biblioteca de documentação da API	Documentação completa da API PyFlink cobrindo DataFrame, operadores de imagem, LLM e AI Function. Suporta consultas em tempo real.
Scripts de verificação de operadores	Scripts executáveis que verificam a existência de uma API ou operador e recuperam a documentação detalhada.

Plataformas suportadas

O Flink Python Coding Skill está em conformidade com o padrão aberto Agent Skills e suporta os principais assistentes de codificação com IA. A tabela a seguir indica o diretório de instalação para cada plataforma.

Plataforma de codificação com IA	Diretório de instalação
Diretório padrão comum	`.agents/skills/`. Claude Code, Cursor, Codex e Gemini CLI identificam este diretório automaticamente.
Claude Code	`.claude/skills/`
Cursor	`.cursor/rules/`
Codex	`.codex/skills/`
OpenCode	`.opencode/skills/`
Gemini CLI	`.gemini/skills/`
Tongyi Lingma e Qoder	`.aone_copilot/skills/` ou `.qoder/skills/`

Pré-requisitos

Instale o Node.js 18 ou superior.

Instale a skill

Execute o seguinte comando no diretório raiz do projeto. O download e a instalação ocorrem automaticamente, sem necessidade de extração manual.

npx skills add aliyun/alibabacloud-aiops-skills \
  --skill alibabacloud-flink-python-coding \
  --global \
  --full-depth

Para mais detalhes sobre a instalação, consulte o portal de Skills da Alibaba Cloud.

Verifique a instalação

Durante a instalação, selecione o agente de codificação com IA desejado seguindo as instruções da linha de comando. Após concluir a instalação, reinicie o cliente do agente e execute o comando abaixo para conferir o resultado da instalação global.

npx skills list --global

A saída deve conter o seguinte conteúdo.

SKILL.md      # Skill definition and usage instructions
assets        # Static resources
references    # PyFlink API documentation and references

Nota

Para instalar a skill no diretório dedicado de um assistente de IA específico, como .claude/skills/, copie o diretório para esse local após a conclusão da instalação.

Uso

Após a instalação, descreva seus requisitos em linguagem natural no assistente de codificação com IA. Não é necessário seguir um modelo fixo. Para garantir que o código gerado se adeque à sua lógica de negócios, inclua as seguintes informações no prompt.

Elemento do prompt	Descrição
Versão do mecanismo VVR	Por exemplo, VVR 11.8.0. Isso assegura que as APIs geradas sejam compatíveis com o mecanismo alvo.
Conectores e schema	Os tipos de fonte e destino de dados, como Kafka, Paimon e Hologres, além das definições de campos.
Operações de processamento de dados	As transformações, filtragens, agregações, UDFs, inferências de IA ou processamentos multimodais desejados.

O exemplo a seguir ilustra um prompt.

Create a Flink Python job on VVR 11.8.0. Read robot-captured frame images from OSS-CDC (fields: image_url, episode_id, frame_id, timestamp), call qwen3-vl-max to generate scene descriptions, use text-embedding-v4 to generate vectors, and write the results to a Paimon data lake.

A skill gera uma cadeia completa de jobs, desde a leitura de dados e o processamento via DataFrame até a escrita dos resultados.

Lê dados da fonte com base no schema confirmado, nas semânticas de inicialização e no modo de streaming ou batch.
Executa transformações, processamento temporal, joins, agregações, UDFs, inferências de IA ou processamentos multimodais usando APIs PyFlink compatíveis com a versão VVR alvo.
Escreve os resultados no destino de dados confirmado, mantendo o schema acordado, as semânticas temporais e as semânticas de changelog.
Gerencia automaticamente parâmetros de conectores, configurações de modelos, dependências Python e arquivos de runtime, evitando a escrita de credenciais diretamente no código.
Gera pacotes de código, pacotes de dependências e instruções de implantação conforme necessário para o deploy do job.

Importante

A skill realiza verificações de sintaxe Python e validação de APIs localmente. Essa verificação local não cobre todos os problemas possíveis no ambiente online. Após implantar o job no workspace alvo e iniciá-lo, valide a conectividade com a fonte de dados, a configuração de recursos, as definições de checkpoint e os resultados do processamento.

Capacidades principais

Recomendação inteligente de APIs

O PyFlink oferece um sistema de API multicamadas que inclui a API DataFrame (semelhante ao pandas), a API Table declarativa e SQL, a API DataStream de baixo nível, UDFs Python e extensões multimodais como operadores de imagem, LLM e AI Function. O agente API Selector integrado à skill seleciona e valida as APIs automaticamente.

Recomendação orientada a tarefas: sugere a combinação ideal de APIs com base na descrição da tarefa e explica o motivo da escolha.
Validação de existência da API: confere em tempo real se a API existe, consultando a documentação oficial do PyFlink.
Soluções alternativas: fornece opções substitutas, incluindo fallback via UDF, quando a API preferida apresenta limitações.

A API DataFrame abrange as seguintes capacidades essenciais.

Processamento padrão de dados: from_records, from_dict, with_column, filter, select, group_by, join.
Processamento multimodal de imagens: fetch_content, image.is_valid, image.decode, image.resize, image.encode.
Chamadas nativas a LLM: llm.predict, llm.ai_embed.
Escrita de dados: write_kafka, write_hologres, write_generic.

Prevenção de problemas comuns

A tabela a seguir lista problemas frequentes em códigos PyFlink gerados por IAs de uso geral e explica como a skill os trata.

Problema comum	Como a skill lida com isso
Chamadas a APIs inexistentes	Valida as APIs em tempo real com base na documentação oficial do PyFlink e executa verificações locais.
Falta de gatilho de execução	Chama a função de escrita de dados correspondente ao formato de saída alvo para acionar a execução do job.
Versão incorreta do mecanismo	Faz a correspondência exata com a versão do mecanismo VVR especificada.
Tipo de retorno de UDF incompatível	Fornece a declaração correta do tipo de retorno conforme a documentação.
Configuração ausente de conector	Preenche automaticamente os parâmetros obrigatórios para conectores como Kafka, OSS e Paimon.
Confusão entre modos streaming e batch	Seleciona automaticamente o modo de streaming ou batch com base nos requisitos.

Exemplos

Cenário 1: Recomendação de API PyFlink e perguntas e respostas

Requisito: Ler vídeos do OSS, extrair quadros em intervalos fixos e chamar um modelo grande para analisar a cena. É necessário determinar as APIs disponíveis e o caminho de implementação.

O exemplo abaixo mostra um prompt.

In PyFlink, I need to read videos from OSS, take one frame every 5 seconds, and call a large model once to analyze the scene. Which APIs should I use?

O exemplo a seguir apresenta a saída esperada.

We recommend the DataFrame multimodal APIs of VVR 11.8 or later. The core chain is:

OSS URI → read_video_frames → IMAGE → image_encode(data_url) → llm.predict

Main APIs to use:

- pf.read_video_frames(...): the recommended high-level interface. It accepts a single OSS URI, a path list, a wildcard path, or a DataFrame that contains video URIs. sample_interval_ms=5000 samples one frame every 5 seconds along the video timeline instead of using a Flink processing-time timer.
- video_explode_frames(...): the low-level frame extraction operator. It expands the result into one row per frame. Use it directly when fine-grained control over video slicing is required.
- video_metadata(...), video_split(...): use them when a long video requires duration detection, slicing, and load redistribution.
- image_encode(..., output="data_url"): encodes an IMAGE into a data URL that vision models can consume.
- pf.set_model_provider(...), pf.OpenAICompatProvider(...): registers a built-in model, or a Model Studio or OpenAI-compatible model.
- DataFrame.llm.predict(...): performs vision understanding frame by frame and outputs scene descriptions.

For long videos, video_extract_frames is not recommended, because it puts all frames into an array in a single row. The per-frame output of read_video_frames and video_explode_frames is better suited to model calls and concurrency control.

If videos are continuously uploaded to OSS:

- Use the oss-cdc connector to subscribe to OSS/MNS object creation events.
- Call pf.read_video_frames(video_df, "url", ...) after the url column is obtained.

In production, the OSS URI plus frame_time_ms can also be used as the cache key to avoid duplicate billing caused by job recovery or model retries.

Não é necessário pesquisar e comparar documentos. Basta descrever seus requisitos em linguagem natural para obter sugestões de APIs, condições de aplicabilidade e caminhos de implementação validados contra a documentação oficial.

Cenário 2: Análise de sentimento em tempo real de avaliações com PyFlink

Requisito: Executar análise de sentimento em tempo real sobre avaliações de clientes em um tópico Kafka e gravar os rótulos de sentimento e as pontuações de confiança de volta no Kafka.

Veja um exemplo de prompt abaixo.

I want to build a customer review analysis job on VVR 11.8.
I have a Kafka cluster whose endpoint is kafka.example.com:9092.
The customer_reviews topic contains two fields: review_id and review.
Call the large model in Flink AI services to determine the sentiment of each review,
then write review_id, the sentiment label, and the confidence score to the review_sentiments topic in the same cluster.

Você não precisa escrever Flink SQL manualmente nem compreender a cadeia de operadores DataStream. Uma descrição em linguagem natural produz um job pronto para implantação.

Cenário 3: Análise de imagens de produtos de e-commerce e ingestão vetorizada em data lake

Requisito: Processar imagens de produtos no OSS em lotes. Ler, pré-processar e extrair a cor dominante das imagens, chamar um modelo grande multimodal para gerar descrições dos produtos e gravar os resultados no data lake.

Confira um exemplo de prompt a seguir.

We have a batch of product images to process in real time with PyFlink.
The input is Parquet files under oss://example-bucket/product-images/input/,
with the fields product_id and image_url.
Download, validate, and decode the images, resize them all to 256x256, and compute an image quality score.
Then use colorthief to extract the dominant color,
and use a large model to generate a product description within 100 words.
Finally, write product_id, the quality score, the dominant color, and the product description to a Paimon data lake.
The Paimon warehouse path is oss://example-bucket/warehouse/.
The VVR version is 11.8.

Não há necessidade de implantar modelos, gerenciar recursos GPU ou escrever serviços de inferência. A API DataFrame fornece operadores multimodais de imagem e funções de IA integrados, permitindo executar compreensão de imagens, geração de texto e vetorização diretamente em jobs Flink de streaming e batch.

Realtime Compute for Apache Flink:Develop jobs with Flink Python Coding Skill