Como usar o catálogo do Hive com o Apache Flink® no HDInsight no AKS

Artigo
10/23/2024

Nota

Vamos desativar o Azure HDInsight no AKS em 31 de janeiro de 2025. Antes de 31 de janeiro de 2025, você precisará migrar suas cargas de trabalho para o Microsoft Fabric ou um produto equivalente do Azure para evitar o encerramento abrupto de suas cargas de trabalho. Os clusters restantes na sua subscrição serão interrompidos e removidos do anfitrião.

Apenas o apoio básico estará disponível até à data da reforma.

Importante

Esta funcionalidade está atualmente em pré-visualização. Os Termos de Utilização Suplementares para Pré-visualizações do Microsoft Azure incluem mais termos legais que se aplicam a funcionalidades do Azure que estão em versão beta, em pré-visualização ou ainda não disponibilizadas para disponibilidade geral. Para obter informações sobre essa visualização específica, consulte Informações de visualização do Azure HDInsight no AKS. Para perguntas ou sugestões de recursos, envie uma solicitação no AskHDInsight com os detalhes e siga-nos para obter mais atualizações na Comunidade do Azure HDInsight.

Este exemplo usa o Metastore do Hive como um catálogo persistente com o Hive Catalog do Apache Flink. Usamos essa funcionalidade para armazenar metadados da tabela Kafka e da tabela MySQL no Flink entre sessões. Flink usa a tabela Kafka registrada no Hive Catalog como fonte, executar alguma pesquisa e resultado de coletor para o banco de dados MySQL

Pré-requisitos

Cluster Apache Flink no HDInsight no AKS com Hive Metastore 3.1.2
Cluster Apache Kafka no HDInsight
- Você é obrigado a garantir que as configurações de rede estejam completas, conforme descrito em Usando o Kafka, ou seja, para garantir que o HDInsight nos clusters AKS e HDInsight estejam na mesma VNet
MySQL 8.0.33

Apache Hive no Apache Flink

O Flink oferece uma dupla integração com o Hive.

A primeira etapa é usar o Hive Metastore (HMS) como um catálogo persistente com o HiveCatalog do Flink para armazenar metadados específicos do Flink em sessões.
- Por exemplo, os usuários podem armazenar suas tabelas Kafka ou ElasticSearch no Hive Metastore usando o HiveCatalog e reutilizá-las posteriormente em consultas SQL.
O segundo é oferecer o Flink como um motor alternativo para ler e escrever tabelas Hive.
O HiveCatalog foi projetado para ser "pronto para uso" compatível com as instalações existentes do Hive. Você não precisa modificar seu Hive Metastore existente ou alterar o posicionamento de dados ou particionamento de suas tabelas.

Para obter mais informações, consulte Apache Hive

Preparação do ambiente

Criar um cluster Apache Flink com HMS

Permite criar um cluster Apache Flink com HMS no portal do Azure, você pode consultar as instruções detalhadas sobre a criação do cluster Flink.

Após a criação do cluster, verifique se o HMS está em execução ou não no lado do AKS.

Preparar dados de transação de pedidos do usuário Tópico Kafka no HDInsight

Baixe o jar do cliente kafka usando o seguinte comando:

wget https://archive.apache.org/dist/kafka/3.2.0/kafka_2.12-3.2.0.tgz

Descompacte o arquivo tar com

tar -xvf kafka_2.12-3.2.0.tgz

Produza as mensagens para o tópico Kafka.

Outros comandos:

Nota

Você é obrigado a substituir o servidor de inicialização pelo seu próprio nome de host ou IP dos corretores kafka

--- delete topic
./kafka-topics.sh --delete --topic user_orders --bootstrap-server wn0-contsk:9092

--- create topic
./kafka-topics.sh --create --replication-factor 2 --partitions 3 --topic user_orders  --bootstrap-server wn0-contsk:9092

--- produce topic
./kafka-console-producer.sh --bootstrap-server wn0-contsk:9092 --topic user_orders

--- consumer topic
./kafka-console-consumer.sh --bootstrap-server wn0-contsk:9092 --topic user_orders --from-beginning

Preparar dados mestre de ordem do usuário no MySQL no Azure

Banco de dados de teste:

Prepare a tabela de pedidos:

mysql> use mydb
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

mysql> CREATE TABLE orders (
  order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  order_date DATETIME NOT NULL,
  customer_id INTEGER NOT NULL,
  customer_name VARCHAR(255) NOT NULL,
  price DECIMAL(10, 5) NOT NULL,
  product_id INTEGER NOT NULL,
  order_status BOOLEAN NOT NULL
) AUTO_INCREMENT = 10001;


mysql> INSERT INTO orders
VALUES (default, '2023-07-16 10:08:22','0001', 'Jark', 50.00, 102, false),
       (default, '2023-07-16 10:11:09','0002', 'Sally', 15.00, 105, false),
       (default, '2023-07-16 10:11:09','000', 'Sally', 25.00, 105, false),
       (default, '2023-07-16 10:11:09','0004', 'Sally', 45.00, 105, false),
       (default, '2023-07-16 10:11:09','0005', 'Sally', 35.00, 105, false),
       (default, '2023-07-16 12:00:30','0006', 'Edward', 90.00, 106, false);

mysql> select * from orders;
+----------+---------------------+-------------+---------------+----------+------------+--------------+
| order_id | order_date          | customer_id | customer_name | price    | product_id | order_status |
+----------+---------------------+-------------+---------------+----------+------------+--------------+
|    10001 | 2023-07-16 10:08:22 |           1 | Jark          | 50.00000 |        102 |            0 |
|    10002 | 2023-07-16 10:11:09 |           2 | Sally         | 15.00000 |        105 |            0 |
|    10003 | 2023-07-16 10:11:09 |           3 | Sally         | 25.00000 |        105 |            0 |
|    10004 | 2023-07-16 10:11:09 |           4 | Sally         | 45.00000 |        105 |            0 |
|    10005 | 2023-07-16 10:11:09 |           5 | Sally         | 35.00000 |        105 |            0 |
|    10006 | 2023-07-16 12:00:30 |           6 | Edward        | 90.00000 |        106 |            0 |
+----------+---------------------+-------------+---------------+----------+------------+--------------+
6 rows in set (0.22 sec)

mysql> desc orders;
+---------------+---------------+------+-----+---------+----------------+
| Field         | Type          | Null | Key | Default | Extra          |
+---------------+---------------+------+-----+---------+----------------+
| order_id      | int           | NO   | PRI | NULL    | auto_increment |
| order_date    | datetime      | NO   |     | NULL    |                |
| customer_id   | int           | NO   |     | NULL    |                |
| customer_name | varchar(255)  | NO   |     | NULL    |                |
| price         | decimal(10,5) | NO   |     | NULL    |                |
| product_id    | int           | NO   |     | NULL    |                |
| order_status  | tinyint(1)    | NO   |     | NULL    |                |
+---------------+---------------+------+-----+---------+----------------+
7 rows in set (0.22 sec)

Usando SSH download necessário conector Kafka e jars de banco de dados MySQL

Nota

Faça o download da versão correta do jar de acordo com a nossa versão kafka do HDInsight e a versão do MySQL.

wget https://repo1.maven.org/maven2/org/apache/flink/flink-connector-jdbc/3.1.0-1.17/flink-connector-jdbc-3.1.0-1.17.jar
wget https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.33/mysql-connector-j-8.0.33.jar
wget https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.2.0/kafka-clients-3.2.0.jar
wget https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka/1.17.0/flink-connector-kafka-1.17.0.jar

Movendo o frasco do planejador

Mova o jar flink-table-planner_2.12-1.17.0-.... jar localizado em webssh pod's /opt to /lib e mova para fora o jar flink-table-planner-loader1.17.0-.... jar /opt/flink-webssh/opt/ de /lib. Consulte o problema para obter mais detalhes. Execute as etapas a seguir para mover o jar do planejador.

mv /opt/flink-webssh/lib/flink-table-planner-loader-1.17.0-*.*.*.*.jar /opt/flink-webssh/opt/
mv /opt/flink-webssh/opt/flink-table-planner_2.12-1.17.0-*.*.*.*.jar /opt/flink-webssh/lib/

Nota

Um jar de planejador extra em movimento só é necessário ao usar o dialeto Hive ou o ponto de extremidade HiveServer2. No entanto, esta é a configuração recomendada para a integração do Hive.

Validação

Use bin/sql-client.sh para se conectar ao Flink SQL

bin/sql-client.sh -j flink-connector-jdbc-3.1.0-1.17.jar -j mysql-connector-j-8.0.33.jar -j kafka-clients-3.2.0.jar -j flink-connector-kafka-1.17.0.jar

Crie o catálogo do Hive e conecte-se ao catálogo do hive no Flink SQL

Nota

Como já usamos o cluster Flink com o Hive Metastore, não há necessidade de executar configurações adicionais.

CREATE CATALOG myhive WITH (
    'type' = 'hive'
);

USE CATALOG myhive;

Criar tabela Kafka no Apache Flink SQL

CREATE TABLE kafka_user_orders (
  `user_id` BIGINT,
  `user_name` STRING,
  `user_email` STRING,
  `order_date` TIMESTAMP(3) METADATA FROM 'timestamp',
  `price` DECIMAL(10,5),
  `product_id` BIGINT,
  `order_status` BOOLEAN
) WITH (
    'connector' = 'kafka',  
    'topic' = 'user_orders',  
    'scan.startup.mode' = 'latest-offset',  
    'properties.bootstrap.servers' = '10.0.0.38:9092,10.0.0.39:9092,10.0.0.40:9092', 
    'format' = 'json' 
);

select * from kafka_user_orders;

Criar tabela MySQL no Apache Flink SQL

CREATE TABLE mysql_user_orders (
  `order_id` INT,
  `order_date` TIMESTAMP,
  `customer_id` INT,
  `customer_name` STRING,
  `price` DECIMAL(10,5),
  `product_id` INT,
  `order_status` BOOLEAN
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://<servername>.mysql.database.azure.com/mydb',
  'table-name' = 'orders',
  'username' = '<username>',
  'password' = '<password>'
);

select * from mysql_user_orders;

Verifique as tabelas registradas no catálogo do Hive acima no Flink SQL

Colete informações de ordem de transação do usuário na tabela de ordem mestre no MySQL no Flink SQL

INSERT INTO mysql_user_orders (order_date, customer_id, customer_name, price, product_id, order_status)
 SELECT order_date, CAST(user_id AS INT), user_name, price, CAST(product_id AS INT), order_status
 FROM kafka_user_orders;

Verifique se os dados da ordem de transação do usuário no Kafka são adicionados na ordem da tabela mestre no MySQL no Azure Cloud Shell

Criando mais três pedidos de usuário no Kafka

sshuser@hn0-contsk:~$ /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --bootstrap-server wn0-contsk:9092 --topic user_orders
>{"user_id": null,"user_name": "Lucy","user_email": "user8@example.com","order_date": "07/17/2023 21:33:44","price": "90.00000","product_id": "102","order_status": false}
>{"user_id": "0009","user_name": "Zark","user_email": "user9@example.com","order_date": "07/17/2023 21:52:07","price": "80.00000","product_id": "103","order_status": true}
>{"user_id": "0010","user_name": "Alex","user_email": "user10@example.com","order_date": "07/17/2023 21:52:07","price": "70.00000","product_id": "104","order_status": true}

Verifique os dados da tabela Kafka no Flink SQL

Flink SQL> select * from kafka_user_orders;

Inserir `product_id=104` na tabela de pedidos no MySQL no Flink SQL

INSERT INTO mysql_user_orders (order_date, customer_id, customer_name, price, product_id, order_status)
SELECT order_date, CAST(user_id AS INT), user_name, price, CAST(product_id AS INT), order_status
FROM kafka_user_orders where product_id = 104;

Verifique `product_id = 104` se o registro foi adicionado na tabela de ordem no MySQL no Azure Cloud Shell

Referência

Apache Hive
Apache, Apache Hive, Hive, Apache Flink, Flink e nomes de projetos de código aberto associados são marcas comerciais da Apache Software Foundation (ASF).

Partilhar via

Como usar o catálogo do Hive com o Apache Flink® no HDInsight no AKS

Pré-requisitos

Apache Hive no Apache Flink

Preparação do ambiente

Criar um cluster Apache Flink com HMS

Preparar dados de transação de pedidos do usuário Tópico Kafka no HDInsight

Preparar dados mestre de ordem do usuário no MySQL no Azure

Usando SSH download necessário conector Kafka e jars de banco de dados MySQL

Validação

Use bin/sql-client.sh para se conectar ao Flink SQL

Crie o catálogo do Hive e conecte-se ao catálogo do hive no Flink SQL

Criar tabela Kafka no Apache Flink SQL

Criar tabela MySQL no Apache Flink SQL

Verifique as tabelas registradas no catálogo do Hive acima no Flink SQL

Colete informações de ordem de transação do usuário na tabela de ordem mestre no MySQL no Flink SQL

Verifique se os dados da ordem de transação do usuário no Kafka são adicionados na ordem da tabela mestre no MySQL no Azure Cloud Shell

Criando mais três pedidos de usuário no Kafka

Verifique os dados da tabela Kafka no Flink SQL

Inserir `product_id=104` na tabela de pedidos no MySQL no Flink SQL

Verifique `product_id = 104` se o registro foi adicionado na tabela de ordem no MySQL no Azure Cloud Shell

Referência

Comentários

Recursos adicionais

Partilhar via

Como usar o catálogo do Hive com o Apache Flink® no HDInsight no AKS

Pré-requisitos

Apache Hive no Apache Flink

Preparação do ambiente

Criar um cluster Apache Flink com HMS

Preparar dados de transação de pedidos do usuário Tópico Kafka no HDInsight

Preparar dados mestre de ordem do usuário no MySQL no Azure

Usando SSH download necessário conector Kafka e jars de banco de dados MySQL

Validação

Use bin/sql-client.sh para se conectar ao Flink SQL

Crie o catálogo do Hive e conecte-se ao catálogo do hive no Flink SQL

Criar tabela Kafka no Apache Flink SQL

Criar tabela MySQL no Apache Flink SQL

Verifique as tabelas registradas no catálogo do Hive acima no Flink SQL

Colete informações de ordem de transação do usuário na tabela de ordem mestre no MySQL no Flink SQL

Verifique se os dados da ordem de transação do usuário no Kafka são adicionados na ordem da tabela mestre no MySQL no Azure Cloud Shell

Criando mais três pedidos de usuário no Kafka

Verifique os dados da tabela Kafka no Flink SQL

Inserir product_id=104 na tabela de pedidos no MySQL no Flink SQL

Verifique product_id = 104 se o registro foi adicionado na tabela de ordem no MySQL no Azure Cloud Shell

Referência

Comentários

Recursos adicionais

Inserir `product_id=104` na tabela de pedidos no MySQL no Flink SQL

Verifique `product_id = 104` se o registro foi adicionado na tabela de ordem no MySQL no Azure Cloud Shell