Использование каталога Hive с Apache Flink® в HDInsight в AKS

Статья
03/29/2024

Внимание

Эта функция в настоящее время доступна для предварительного ознакомления. Дополнительные условия использования для предварительных версий Microsoft Azure включают более юридические термины, применимые к функциям Azure, которые находятся в бета-версии, в предварительной версии или в противном случае еще не выпущены в общую доступность. Сведения об этой конкретной предварительной версии см. в статье Azure HDInsight в предварительной версии AKS. Для вопросов или предложений функций отправьте запрос на AskHDInsight с подробными сведениями и следуйте за нами для получения дополнительных обновлений в сообществе Azure HDInsight.

В этом примере хранилище метаданных Hive используется в качестве постоянного каталога с каталогом Hive Apache Flink. Мы используем эту функцию для хранения метаданных таблицы Kafka и таблицы MySQL в Flink между сеансами. Flink использует таблицу Kafka, зарегистрированную в каталоге Hive в качестве источника, выполните некоторые результаты поиска и приемника в базе данных MySQL

Необходимые компоненты

Кластер Apache Flink в HDInsight в AKS с хранилищем метаданных Hive 3.1.2
Кластер Apache Kafka в HDInsight
- Необходимо убедиться, что параметры сети выполнены, как описано в разделе "Использование Kafka", чтобы убедиться, что HDInsight в кластерах AKS и HDInsight находятся в одной виртуальной сети.
MySQL 8.0.33

Apache Hive в Apache Flink

Flink предлагает двухкратную интеграцию с Hive.

Первым шагом является использование хранилища метаданных Hive (HMS) в качестве постоянного каталога с помощью Flink в HiveCatalog для хранения определенных метаданных Flink в сеансах.
- Например, пользователи могут хранить свои таблицы Kafka или ElasticSearch в хранилище метаданных Hive с помощью HiveCatalog и повторно использовать их позже в запросах SQL.
Второй — предложить Flink в качестве альтернативного механизма для чтения и записи таблиц Hive.
HiveCatalog предназначен для обеспечения совместимости с существующими установками Hive. Вам не нужно изменять существующее хранилище метаданных Hive или изменять размещение данных или секционирование таблиц.

Дополнительные сведения см. в разделе Apache Hive

Подготовка среды

Создание кластера Apache Flink с помощью HMS

Позволяет создать кластер Apache Flink с помощью HMS на портал Azure, см. подробные инструкции по созданию кластера Flink.

После создания кластера проверка HMS выполняется или нет на стороне AKS.

Подготовка данных транзакций заказа пользователей в разделе Kafka в HDInsight

Скачайте jar-файл клиента kafka с помощью следующей команды:

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

Распакуйте тар-файл с помощью

tar -xvf kafka_2.12-3.2.0.tgz

Создай сообщения в разделе Kafka.

Другие команды:

Примечание.

Вам потребуется заменить bootstrap-server собственным именем узла или IP-адресом брокеров 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

Подготовка основных данных заказа пользователей в MySQL в Azure

Тестирование базы данных:

Подготовьте таблицу заказа:

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)

Скачивание SSH требуется соединитель Kafka и My База данных SQL jars

Примечание.

Скачайте правильный jar-файл версии в формате HDInsight kafka и 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

Перемещение jar-файла планировщика

Переместите jar flink-table-planner_2.12-1.17.0-.... jar, расположенный в файле pod webssh /opt to /lib, и переместить jar-flink-table-planner-loader1.17.0-.. jar /opt/flink-webssh/opt/ from /lib. Дополнительные сведения см. в статье о проблеме . Выполните следующие действия, чтобы переместить jar-файл планировщика.

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/

Примечание.

Дополнительное перемещение jar-файла планировщика необходимо только при использовании диалекта Hive или конечной точки HiveServer2. Однако это рекомендуемая настройка для интеграции Hive.

Проверка

Подключение к Flink SQL с помощью bin/sql-client.sh

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

Создание каталога Hive и подключение к каталогу hive в Flink SQL

Примечание.

Так как мы уже используем кластер Flink с хранилищем метаданных Hive, вам не нужно выполнять дополнительные конфигурации.

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

USE CATALOG myhive;

Создание таблицы Kafka в 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;

Создание таблицы MySQL в 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;

Проверка таблиц, зарегистрированных в выше каталоге Hive в Flink SQL

Сведения о заказе транзакций приемника в таблицу главного порядка в MySQL в 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;

Проверьте, добавляются ли данные заказа транзакций пользователей в Kafka в главном порядке таблицы в MySQL в Azure Cloud Shell

Создание трех дополнительных заказов пользователей на 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}

Проверка данных таблицы Kafka в Flink SQL

Flink SQL> select * from kafka_user_orders;

Вставка `product_id=104` в таблицу заказов в MySQL в 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;

Запись проверки `product_id = 104` добавлена в таблицу заказов в MySQL в Azure Cloud Shell

Справочные материалы

Apache Hive
Apache, Apache Hive, Hive, Apache Flink, Flink и связанные открытый код имена проектов являются товарными знаками Apache Software Foundation (ASF).

Share via

Использование каталога Hive с Apache Flink® в HDInsight в AKS

Необходимые компоненты

Apache Hive в Apache Flink

Подготовка среды

Создание кластера Apache Flink с помощью HMS

Подготовка данных транзакций заказа пользователей в разделе Kafka в HDInsight

Подготовка основных данных заказа пользователей в MySQL в Azure

Скачивание SSH требуется соединитель Kafka и My База данных SQL jars

Проверка

Подключение к Flink SQL с помощью bin/sql-client.sh

Создание каталога Hive и подключение к каталогу hive в Flink SQL

Создание таблицы Kafka в Apache Flink SQL

Создание таблицы MySQL в Apache Flink SQL

Проверка таблиц, зарегистрированных в выше каталоге Hive в Flink SQL

Сведения о заказе транзакций приемника в таблицу главного порядка в MySQL в Flink SQL

Проверьте, добавляются ли данные заказа транзакций пользователей в Kafka в главном порядке таблицы в MySQL в Azure Cloud Shell

Создание трех дополнительных заказов пользователей на Kafka

Проверка данных таблицы Kafka в Flink SQL

Вставка `product_id=104` в таблицу заказов в MySQL в Flink SQL

Запись проверки `product_id = 104` добавлена в таблицу заказов в MySQL в Azure Cloud Shell

Справочные материалы

Дополнительные ресурсы

Share via

Использование каталога Hive с Apache Flink® в HDInsight в AKS

Необходимые компоненты

Apache Hive в Apache Flink

Подготовка среды

Создание кластера Apache Flink с помощью HMS

Подготовка данных транзакций заказа пользователей в разделе Kafka в HDInsight

Подготовка основных данных заказа пользователей в MySQL в Azure

Скачивание SSH требуется соединитель Kafka и My База данных SQL jars

Проверка

Подключение к Flink SQL с помощью bin/sql-client.sh

Создание каталога Hive и подключение к каталогу hive в Flink SQL

Создание таблицы Kafka в Apache Flink SQL

Создание таблицы MySQL в Apache Flink SQL

Проверка таблиц, зарегистрированных в выше каталоге Hive в Flink SQL

Сведения о заказе транзакций приемника в таблицу главного порядка в MySQL в Flink SQL

Проверьте, добавляются ли данные заказа транзакций пользователей в Kafka в главном порядке таблицы в MySQL в Azure Cloud Shell

Создание трех дополнительных заказов пользователей на Kafka

Проверка данных таблицы Kafka в Flink SQL

Вставка product_id=104 в таблицу заказов в MySQL в Flink SQL

Запись проверки product_id = 104 добавлена в таблицу заказов в MySQL в Azure Cloud Shell

Справочные материалы

Дополнительные ресурсы

Вставка `product_id=104` в таблицу заказов в MySQL в Flink SQL

Запись проверки `product_id = 104` добавлена в таблицу заказов в MySQL в Azure Cloud Shell