Realtime Compute for Apache Flink:Jobs com dialeto Hive

Pré-requisitos

• Se você acessar o console como usuário RAM, função RAM ou outra identidade, verifique se possui as permissões necessárias. Para mais informações, consulte permissões.

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

Limitações

• Somente o Ververica Runtime (VVR) 8.0.11 e versões posteriores oferecem suporte ao dialeto Hive.

• Atualmente, os jobs SQL suportam apenas a sintaxe INSERT Statements do dialeto Hive. Declare USE Catalog <yourHiveCatalog> antes das instruções INSERT Statements. Caso precise criar uma tabela, execute a operação na página Scripts.

• Não há suporte para funções definidas pelo usuário (UDFs) do Hive e do Flink.

Etapa 1: Criar um catálogo Hive

1. Configure os metadados do Hive. Para mais informações, consulte Configurar metadados do Hive.

2. Crie um catálogo Hive. Para mais informações, consulte Criar um catálogo Hive.

   Neste tutorial, o catálogo Hive chama-se hdfshive.

Etapa 2: Preparar tabelas Hive de exemplo

1. No painel de navegação à esquerda, acesse Development > Scripts. Clique em New para criar um script.

2. Execute as instruções SQL de exemplo a seguir.

   Importante

   A tabela de origem e a tabela de destino do Hive devem ser tabelas permanentes criadas com a instrução CREATE TABLE. Não é possível usar tabelas temporárias criadas com a instrução CREATE TEMPORARY TABLE.

   -- Use the Hive catalog. In this example, the catalog is named hdfshive and was created in Step 1.
   USE CATALOG hdfshive;   
   -- Create a source table with the default storage format.
   CREATE TABLE source_table (
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )WITH ('connector' = 'hive');
   -- Create a sink table with the default storage format.
   CREATE TABLE target_table (
   city STRING,
   avg_salary FLOAT,
   user_count INT
   )WITH ('connector' = 'hive');
   -- Insert sample data into the source table.
   INSERT INTO source_table VALUES
   (1, 'Alice', 25, 'New York', 5000.0),
   (2, 'Bob', 30, 'San Francisco', 6000.0),
   (3, 'Charlie', 35, 'New York', 7000.0),
   (4, 'David', 40, 'San Francisco', 8000.0),
   (5, 'Eva', 45, 'Los Angeles', 9000.0);
   -- Create a table with a specific storage format, for example, Parquet.
   -- Load the Hive module.
   load MODULE hive with ('hive-version' = '2.3.6');
   use CATALOG `hdfshive`;
   -- Required: Set the SQL dialect to 'hive' to recognize Hive DDL keywords such as 'STORED'.
   set 'table.sql-dialect' = 'hive';
   CREATE TABLE `parquet_table`(
    id INT,
    name STRING,
    age INT,
    city STRING,
    salary FLOAT
   )STORED AS PARQUET;

Etapa 3: Criar um job Hive SQL

1. No painel de navegação à esquerda, acesse Development > ETL.

2. Clique em New. Na caixa de diálogo New Draft, selecione Blank Batch Draft (BETA) e clique em Next.

3. Insira as informações do job.

   Parâmetro	Descrição	Exemplo
   Name	Nome do job. Nota O nome do job deve ser exclusivo no workspace atual.	hive-sql
   Location	Pasta onde o arquivo de código do job fica armazenado. Também é possível clicar no ícone à direita de uma pasta existente para criar uma subpasta.	Drafts
   Engine version	Versão do mecanismo Flink usada pelo job. Recomendamos selecionar uma versão com a tag RECOMMENDED. Essas versões oferecem maior confiabilidade e desempenho. Para mais informações sobre versões do mecanismo, consulte Notas de versão e Versões do mecanismo.	vvr-8.0.11-flink-1.17
   SQL dialect	Linguagem SQL para processamento de dados. Nota Este parâmetro aparece somente se você selecionar uma versão do mecanismo compatível com o dialeto Hive.	Hive SQL

4. Clique em Create.

Etapa 4: Escrever e implantar o job Hive SQL

1. Escreva as instruções SQL.

   Este exemplo calcula o número de usuários com mais de 30 anos e o salário médio por cidade. Copie o script SQL abaixo para o editor SQL.

   -- Use the Hive catalog. In this example, the catalog is named hdfshive and was created in Step 1.
   USE CATALOG hdfshive; 
   INSERT INTO TABLE target_table
   SELECT
     city,
     AVG(salary) AS avg_salary, -- Calculate the average salary
     COUNT(id) AS user_count -- Count the number of users
   FROM source_table
   WHERE age > 30 -- Filter for users older than 30
   GROUP BY city; -- Group by city

2. No canto superior direito, clique em Deploy. Na caixa de diálogo, configure os parâmetros conforme necessário (este tutorial usa as configurações padrão) e clique em OK.

(Opcional) Etapa 5: Configurar parâmetros de runtime

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

2. Na lista suspensa, selecione Batch job. Localize o job desejado e clique em Details na coluna Actions.

3. No painel de detalhes da implantação, clique em Edit na seção Runtime parameters configuration.

4. No campo Other Configuration, adicione a seguinte configuração:

   fs.oss.jindo.endpoint: <YOUR_Endpoint> 
   fs.oss.jindo.buckets: <YOUR_Buckets>
   fs.oss.jindo.accessKeyId: <YOUR_AccessKeyId>
   fs.oss.jindo.accessKeySecret: <YOUR_AccessKeySecret>

   Para mais informações sobre esses parâmetros, consulte Gravar dados no OSS-HDFS.

5. Clique em Save.

Etapa 6: Iniciar o job e visualizar os resultados

1. Na página Deployments, selecione Batch job no filtro, localize o job desejado (por exemplo, hive-sql) e clique em Start na coluna Actions.

2. Após o status do job mudar para FINISHED, visualize os resultados.

   Na página Development > Scripts, execute a seguinte instrução SQL para visualizar os dados, que incluem o número de usuários com mais de 30 anos e seus salários médios em cada cidade.

   -- Use the Hive catalog. In this example, the catalog is named hdfshive and was created in Step 1.
   USE CATALOG hdfshive; 
   select * from target_table;

   A consulta retorna três linhas da tabela target_table com as colunas city, avg_salary e user_count: Los Angeles (9000.0, 1), New York (7000.0, 1) e San Francisco (8000.0, 1).

Desenvolvimento de job JAR Hive

É possível executar jobs com dialeto Hive como jobs JAR. Isso requer a versão 11,2 ou posterior do pacote JAR "ververica-connector-hive-2.3.6". Também é necessário garantir que as configurações do Hive no seu job JAR correspondam às configurações do console.

1. Configurações do console

   1. O URI do JAR especifica o pacote JAR enviado para o job JAR.

   2. Em Additional Dependencies, envie os quatro arquivos de configuração do seu cluster Hive: core-site.xml, mapred-site.xml, hdfs-site.xml e hive-site.xml. Envie também o pacote JAR ververica-connector-hive-2.3.6.

   3. Configure os parâmetros de runtime. Com base na configuração do seu cluster Hive, se precisar gravar dados no OSS-HDFS, utilize as definições descritas em (Opcional) Etapa 5: Configurar parâmetros de runtime.

      table.sql-dialect: HIVE
      classloader.parent-first-patterns.additional: org.apache.hadoop;org.antlr.runtime
      kubernetes.application-mode.classpath.include-user-jar: true

2. Exemplo de código de job JAR:

   1. StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
      Configuration conf = new Configuration();
      conf.setString("type", "hive");
      conf.setString("default-database", "default");
      conf.setString("hive-version", "2.3.6"); 
      conf.setString("hive-conf-dir", "/flink/usrlib/" );
      conf.setString("hadoop-conf-dir", "/flink/usrlib/");
      CatalogDescriptor descriptor = CatalogDescriptor.of("hivecat", conf);
      tableEnv.createCatalog("hivecat", descriptor);
      tableEnv.loadModule("hive", new HiveModule());
      tableEnv.useModules("hive");
      tableEnv.useCatalog("hivecat");
      tableEnv.executeSql("insert into `hivecat`.`default`.`test_write` select * from `hivecat`.`default`.`test_read`;");

Documentos relacionados

• Para mais informações sobre a sintaxe da instrução INSERT para o dialeto Hive, consulte Instruções INSERT | Apache Flink.

• Para mais informações sobre processamento de dados em lote com Flink SQL, consulte Processamento em lote.