Realtime Compute for Apache Flink:Job Flink SQL

Pré-requisitos

• O usuário RAM ou a função RAM em uso deve ter as permissões necessárias para o console Realtime Compute. Para mais informações, consulte permissões.

• Um workspace Flink criado. Para mais informações, consulte Ativar o Realtime Compute for Apache Flink.

Etapa 1: Crie um rascunho SQL

1. Acesse a página de criação de rascunho SQL.

   1. Faça login no console Realtime Compute.

   2. Localize o workspace Flink desejado e clique em Console na coluna Actions.

   3. No painel de navegação à esquerda, clique em Development > ETL.

2. Clique no ícone e, em seguida, clique em New Blank Stream Draft. Insira um File Name e selecione uma Engine Version.

   O Realtime Compute for Apache Flink oferece diversos modelos de código e de sincronização de dados. Cada modelo inclui descrições de casos de uso, exemplos de código e instruções. Clique em um modelo para conhecer rapidamente os recursos do produto e a sintaxe necessária para implementar sua lógica de negócios. Para mais informações, consulte Modelos de código e Modelos de sincronização de dados.

   Parâmetro	Descrição	Exemplo
   File Name	Nome do rascunho SQL. Nota O nome deve ser único no projeto atual.	flink-test
   Engine Version	Versão do mecanismo Flink para o rascunho SQL. Recomendamos o uso de versões com a tag Recommend ou Stable para garantir maior confiabilidade e desempenho. Para mais detalhes sobre as versões do mecanismo, consulte Notas de versão e Versões do mecanismo.	vvr-8.0.8-flink-1.17

3. Clique em Create.

Etapa 2: Escrever SQL e visualize configurações do rascunho

1. Escreva o código SQL.

   Copie o SQL abaixo para o editor. Este exemplo usa um conector Datagen para gerar um fluxo de dados aleatório e um conector Print para gravar a saída nos logs do console. Para mais informações sobre conectores suportados, consulte Conectores suportados.

   -- Create a temporary source table named datagen_source.
   CREATE TEMPORARY TABLE datagen_source(
     randstr VARCHAR
   ) WITH (
     'connector' = 'datagen' -- Use the Datagen connector.
   );
   
   -- Create a temporary sink table named print_table.
   CREATE TEMPORARY TABLE print_table(
     randstr  VARCHAR
   ) WITH (
     'connector' = 'print',   -- Use the Print connector.
     'logger' = 'true'        -- Write the output to logs.
   );
   
   -- Select a substring from the randstr field and insert it into the sink table.
   INSERT INTO print_table
   SELECT SUBSTRING(randstr,0,8) from datagen_source;

   Nota

   • Este exemplo usa uma instrução INSERT INTO para gravar dados em uma única tabela de destino. Também é possível usar a instrução INSERT INTO para gravar em várias tabelas de destino. Para mais informações, consulte Instrução INSERT INTO.

   • Em produção, use tabelas registradas em Catálogos em vez de tabelas temporárias. Para mais informações, consulte Catálogos.

2. Visualize as configurações do rascunho.

   À direita do editor SQL, visualize ou defina configurações nas abas disponíveis.

   Aba	Descrição
   More configurations	Engine version: Para mais informações, consulte Versões do mecanismo e Políticas de ciclo de vida. Recomendamos o uso de uma versão recomendada ou estável. As tags de versão são: Recommend: A versão secundária mais recente da versão principal atual. Stable: A versão secundária mais recente de uma versão principal dentro do período de suporte, com defeitos conhecidos corrigidos. Normal: Outras versões secundárias ainda dentro do período de suporte. Deprecated: Versões que ultrapassaram a data de fim de vida (EOL). Additional dependencies: Dependências adicionais necessárias para o job, como funções temporárias. Autenticação Kerberos: Ative a autenticação Kerberos e configure um cluster Kerberos registrado e as informações do principal. Caso não tenha registrado um cluster Kerberos, consulte Registrar um cluster Hive Kerberos.
   Code structure	Data flow: Use o diagrama de fluxo de dados para visualizar rapidamente a linhagem dos dados. Tree structure: Use o diagrama de estrutura em árvore para visualizar rapidamente as fontes de dados.
   Version information	Visualize aqui o histórico de versões do rascunho SQL. Para mais informações sobre as funções na coluna Actions, consulte Gerencie versões de rascunho.

(Opcional) Etapa 3: Validar e depurar o rascunho SQL

1. Valide o rascunho SQL.

   A validação verifica a semântica SQL, a conectividade de rede e os metadados das tabelas usadas no rascunho. Após a validação, clique em SQL Optimization na área de resultados para visualizar possíveis riscos e sugestões de otimização do SQL.

   1. No canto superior direito do editor SQL, clique em Deep Check.

   2. Na caixa de diálogo Deep Check, clique em Confirm.

   Nota

   Se ocorrer um erro de tempo limite, a seguinte mensagem poderá ser exibida: The RPC times out maybe because the SQL parsing is too complicated. Please consider enlarging the flink.sqlserver.rpc.execution.timeout option in flink-configuration, which by default is 120 s.

   Solução: Adicione o seguinte parâmetro de configuração no início do editor SQL.

   SET 'flink.sqlserver.rpc.execution.timeout' = '600s';

2. Depure o rascunho SQL.

   O recurso de depuração permite simular a execução de um job para verificar a saída e validar a lógica SELECT ou INSERT. Esse recurso aumenta a eficiência do desenvolvimento e reduz riscos relacionados à qualidade dos dados.

   Nota

   O recurso de depuração não grava dados na tabela de destino.

   1. No canto superior direito do editor SQL, clique em Debug.

   2. Na caixa de diálogo Debug, selecione um cluster de depuração e clique em Next.

      Se nenhum cluster de sessão estiver disponível, crie um. O cluster de sessão deve usar a mesma versão do mecanismo do rascunho SQL e estar em execução. Para mais informações, consulte Etapa 1: Criar um cluster de sessão.

   3. Configure os dados de depuração e clique em OK.

      Para mais detalhes sobre a configuração, consulte Etapa 2: Depurar um job.

Etapa 4: Implantar o rascunho SQL

No canto superior direito do editor SQL, clique em Deploy. Na caixa de diálogo Deploy New Version, configure os parâmetros conforme necessário e clique em OK.

Ao implantar o rascunho, selecione uma queue ou um session cluster como Deployment target. A tabela a seguir compara essas duas opções.

Etapa 5: Iniciar o job e visualize resultados

1. No painel de navegação à esquerda, clique em O&M > Deployments.

2. Localize o job desejado e clique em Start na coluna Actions.

   Selecione stateless start e clique em Start. O job estará em execução quando o status mudar para Running. Para mais informações sobre parâmetros de inicialização, consulte Iniciar um job.

3. Na página de detalhes da Deployment, visualize os resultados do job Flink.

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

   2. Na aba Logs, clique na subaba Running TaskManagers. Na coluna Path, ID, clique em um TaskManager.

   3. Clique na aba Logs e pesquise logs relacionados ao PrintSinkOutputWriter.

      Se encontrar entradas de log contendo a cadeia Source: datagen_source → Calc → Sink: print_table, o fluxo de dados está sendo processado corretamente.

(Opcional) Etapa 6: Parar o job

Para aplicar alterações a um job (como modificações no código, atualizações de parâmetros WITH ou mudanças de versão), reimplemente-o, pare-o e reinicie-o. A reinicialização também é necessária para uma inicialização sem estado ou para aplicar alterações de configuração não dinâmicas. Para mais informações sobre como parar um job, consulte Parar um job.

1. Na página O&M > Deployments, localize o job desejado e clique em Cancel na coluna Actions.

2. Clique em OK.

Documentos relacionados

• Perguntas frequentes sobre desenvolvimento e O&M de jobs

  • Desenvolvimento de jobs

  • O&M de jobs

• Configure informações do job

  • Configure recursos antes de iniciar um job ou modifique-os para uma implantação em execução. Dois modos de configuração de recursos são suportados: básico (granularidade grossa) e especialista (granularidade fina). Para mais informações, consulte Configurar recursos do job.

  • Defina níveis de log e especifique saídas diferentes para cada nível. Para mais informações, consulte Configurar saídas de log do job.

• Fluxos de trabalho de desenvolvimento para outros tipos de job

  • Job Flink JAR

  • Job Flink Python

  • Ingestão de banco de dados em tempo real

• Melhores práticas para Flink

  • Criar um data warehouse em tempo real com Hologres

  • Criar um lakehouse streaming com Paimon e StarRocks