Come usare il catalogo Hive con Apache Flink® in HDInsight nel servizio Azure Kubernetes
Importante
Questa funzionalità è attualmente disponibile solo in anteprima. Le condizioni per l'utilizzo supplementari per le anteprime di Microsoft Azure includono termini legali più validi applicabili alle funzionalità di Azure disponibili in versione beta, in anteprima o non ancora rilasciate nella disponibilità generale. Per informazioni su questa anteprima specifica, vedere Informazioni sull'anteprima di Azure HDInsight nel servizio Azure Kubernetes. Per domande o suggerimenti sulle funzionalità, inviare una richiesta in AskHDInsight con i dettagli e seguire microsoft per altri aggiornamenti nella community di Azure HDInsight.
Questo esempio usa il Metastore di Hive come catalogo permanente con il catalogo Hive di Apache Flink. Questa funzionalità viene usata per archiviare i metadati della tabella Kafka e della tabella MySQL in Flink tra le sessioni. Flink usa la tabella Kafka registrata nel catalogo Hive come origine, esegue alcune ricerche e risultati sink nel database MySQL
Prerequisiti
- Cluster Apache Flink in HDInsight nel servizio Azure Kubernetes con Metastore Hive 3.1.2
- Cluster Apache Kafka in HDInsight
- È necessario assicurarsi che le impostazioni di rete siano complete come descritto in Uso di Kafka, per assicurarsi che HDInsight nel servizio Azure Kubernetes e i cluster HDInsight si trovino nella stessa rete virtuale
- MySQL 8.0.33
Apache Hive in Apache Flink
Flink offre un'integrazione bidirezionale con Hive.
- Il primo passaggio consiste nell'usare Hive Metastore (HMS) come catalogo permanente con HiveCatalog di Flink per archiviare metadati specifici Flink tra le sessioni.
- Ad esempio, gli utenti possono archiviare le tabelle Kafka o ElasticSearch in Metastore Hive usando HiveCatalog e riutilizzarle in un secondo momento nelle query SQL.
- Il secondo consiste nell'offrire Flink come motore alternativo per la lettura e la scrittura di tabelle Hive.
- HiveCatalog è progettato per essere compatibile con le installazioni Hive esistenti. Non è necessario modificare il metastore Hive esistente o modificare il posizionamento o il partizionamento dei dati delle tabelle.
Per altre informazioni, vedere Apache Hive
Preparazione dell'ambiente
Creare un cluster Apache Flink con HMS
Consente di creare un cluster Apache Flink con HMS in portale di Azure, è possibile fare riferimento alle istruzioni dettagliate sulla creazione del cluster Flink.
Dopo la creazione del cluster, verificare che HMS sia in esecuzione o meno sul lato servizio Azure Kubernetes.
Preparare l'argomento Kafka sui dati delle transazioni degli ordini utente in HDInsight
Scaricare il file JAR del client kafka usando il comando seguente:
wget https://archive.apache.org/dist/kafka/3.2.0/kafka_2.12-3.2.0.tgz
Annullare il valore del file tar con
tar -xvf kafka_2.12-3.2.0.tgz
Produrre i messaggi all'argomento Kafka.
Altri comandi:
Nota
È necessario sostituire bootstrap-server con il proprio nome host o IP dei broker 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
Preparare i dati master degli ordini utente in MySQL in Azure
Database di test:
Preparare la tabella degli ordini:
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)
Uso del download SSH necessario per il connettore Kafka e my database SQL jar
Nota
Scaricare il file JAR della versione corretta in base alla versione kafka di HDInsight e alla versione di 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
Spostamento del file JAR del pianificatore
Spostare il file jar flink-table-planner_2.12-1.17.0-.... jar che si trova nel pod webssh /opt to /lib e sposta il file jar flink-table-planner-loader1.17.0-.... jar /opt/flink-webssh/opt/ da /lib. Per altri dettagli, vedere il problema . Per spostare il file JAR di Planner, seguire questa procedura.
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
Lo spostamento di un file JAR aggiuntivo di Planner è necessario solo quando si usa il dialetto Hive o l'endpoint HiveServer2. Tuttavia, si tratta della configurazione consigliata per l'integrazione di Hive.
Convalida
Usare bin/sql-client.sh per connettersi a 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
Creare il catalogo Hive e connettersi al catalogo hive in Flink SQL
Nota
Poiché si usa già il cluster Flink con Il metastore Hive, non è necessario eseguire configurazioni aggiuntive.
CREATE CATALOG myhive WITH (
'type' = 'hive'
);
USE CATALOG myhive;
Creare una tabella Kafka in 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;
Creare una tabella MySQL in 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;
Controllare le tabelle registrate in precedenza nel catalogo Hive in Flink SQL
Sink user transaction order info into master order table in MySQL on 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;
Controllare se i dati dell'ordine delle transazioni utente in Kafka sono stati aggiunti nell'ordine delle tabelle master in MySQL in Azure Cloud Shell
Creazione di altri tre ordini utente in 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}
Controllare i dati della tabella Kafka in Flink SQL
Flink SQL> select * from kafka_user_orders;
Inserire product_id=104 nella tabella orders in MySQL in 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;
Controllare che product_id = 104 il record venga aggiunto nella tabella degli ordini in MySQL in Azure Cloud Shell
Riferimento
- Apache Hive
- Apache, Apache Hive, Hive, Apache Flink, Flink e i nomi dei progetti open source associati sono marchi di Apache Software Foundation (ASF).
Commenti e suggerimenti
Presto disponibile: Nel corso del 2024 verranno gradualmente disattivati i problemi di GitHub come meccanismo di feedback per il contenuto e ciò verrà sostituito con un nuovo sistema di feedback. Per altre informazioni, vedere https://aka.ms/ContentUserFeedback.
Invia e visualizza il feedback per