Dialetto Hive nei cluster Apache Flink® su HDInsight su Azure Kubernetes Service (AKS)

Articolo
09/20/2024

Importante

Azure HDInsight su AKS è stato ritirato il 31 gennaio 2025. Scopri di più con questo annuncio.

È necessario eseguire la migrazione dei carichi di lavoro a Microsoft Fabric o a un prodotto Azure equivalente per evitare la chiusura brusca dei carichi di lavoro.

Importante

Questa funzionalità è attualmente in anteprima. Le condizioni supplementari per l'utilizzo per le anteprime di Microsoft Azure includono termini legali più validi applicabili alle funzionalità di Azure in versione beta, in anteprima o altrimenti non ancora rilasciate nella disponibilità generale. Per informazioni su questa anteprima specifica, vedere informazioni sull'anteprima di Azure HDInsight su Azure Kubernetes Service (AKS). Per domande o suggerimenti sulle funzionalità, inviare una richiesta in AskHDInsight con i dettagli e seguirci per altri aggiornamenti su Azure HDInsight Community.

Questo articolo illustra come usare il dialetto Hive nei cluster Apache Flink in HDInsight su AKS.

Introduzione

L'utente non può modificare il dialetto predefinito denominato 'flink' in dialetto hive per l'utilizzo in HDInsight su cluster AKS. Tutte le operazioni SQL hanno esito negativo dopo la modifica del dialetto hive con l'errore seguente.


*java.lang.ClassCastException: class jdk.internal.loader.ClassLoaders$AppClassLoader can't be cast to class java.net.URLClassLoader*

Il motivo di questo problema è dovuto a un'istanza aperta su Hive Jira. Hive presuppone attualmente che il caricatore della classe di sistema sia un'istanza di URLClassLoader. In Java 11, questo presupposto non è valido.

Come usare il dialetto Hive in Flink

Eseguire i passaggi seguenti in webssh:
1. Rimuovere il file flink-sql-connector-hive*jar esistente nel percorso lib
```
rm /opt/flink-webssh/lib/flink-sql-connector-hive*jar
```
2. Scarica il JAR sottoelencato nel pod webssh e aggiungilo in /opt/flink-webssh/lib utilizzando wget https://mvnrepository.com/artifact/org.apache.flink/flink-table-planner-loader/1.17.0. Il file JAR hive precedente ha la correzione https://issues.apache.org/jira/browse/HIVE-27508)
```
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/
```
1. Aggiungere le chiavi seguenti nella gestione della configurazione, sezione flink, sotto la sezione core-site.xml:
```
fs.azure.account.key.<STORAGE>.dfs.core.windows.net: <KEY>
flink.hadoop.fs.azure.account.key.<STORAGE>.dfs.core.windows.net: <KEY>
```

Ecco una panoramica delle query nel dialetto hive

Esecuzione del linguaggio Hive con Flink senza partizionamento

  root [ ~ ]# ./bin/sql-client.sh
  Flink SQL>
  Flink SQL> create catalog myhive with ('type' = 'hive', 'hive-conf-dir' = '/opt/hive-conf');
  [INFO] Execute statement succeed.

  Flink SQL> use catalog myhive;
  [INFO] Execute statement succeed.

  Flink SQL> load module hive;
  [INFO] Execute statement succeed.

  Flink SQL> use modules hive,core;
  [INFO] Execute statement succeed.

  Flink SQL> set table.sql-dialect=hive;
  [INFO] Session property has been set.

  Flink SQL> set sql-client.execution.result-mode=tableau;
  [INFO] Session property has been set.

  Flink SQL> select explode(array(1,2,3));Hive Session ID = 6ba45be2-360e-4bee-8842-2765c91581c8


> [!WARNING]
> An illegal reflective access operation has occurred

> [!WARNING]
> Illegal reflective access by org.apache.hadoop.hive.common.StringInternUtils (file:/opt/flink-webssh/lib/flink-sql-connector-hive-3.1.2_2.12-1.16-SNAPSHOT.jar) to field java.net.URI.string

> [!WARNING]
> Please consider reporting this to the maintainers of org.apache.hadoop.hive.common.StringInternUtils

> [!WARNING]
> `Use --illegal-access=warn` to enable warnings of further illegal reflective access operations

> [!WARNING]
>  All illegal access operations will be denied in a future release
select explode(array(1,2,3));


+----+-------------+
| op |         col |
+----+-------------+
| +I |           1 |
| +I |           2 |
| +I |           3 |
+----+-------------+

Received a total of 3 rows

Flink SQL> create table tttestHive Session ID = fb8b652a-8dad-4781-8384-0694dc16e837

[INFO] Execute statement succeed.

Flink SQL> insert into table tttestHive Session ID = f239dc6f-4b58-49f9-ad02-4c73673737d8),(3,'c'),(4,'d');

[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: d0542da4c4252f9494298666ff4e9f8e

Flink SQL> set execution.runtime-mode=batch;
[INFO] Session property has been set.

Flink SQL> select * from tttestHive Session ID = 61b6eb3b-90a6-499c-aced-0598366c5b31

+-----+-------+
| key | value |
+-----+-------+
|   1 |     a |
|   1 |     a |
|   2 |     b |
|   3 |     c |
|   3 |     c |
|   3 |     c |
|   4 |     d |
|   5 |     e |
+-----+-------+
8 rows in set

Flink SQL> QUIT;Hive Session ID = 2dadad92-436e-426e-a88c-66eafd740d98

[INFO] Exiting Flink SQL CLI Client...

Shutting down the session...
done.
root [ ~ ]# exit

I dati sono scritti nello stesso contenitore configurato nella directory hive/warehouse.

Esecuzione del dialetto Hive in Flink con partizioni

  create table tblpart2 (key int, value string) PARTITIONED by ( part string ) tblproperties ('sink.partition-commit.delay'='1 s', 'sink.partition-commit.policy.kind'='metastore,success-file');

  insert into table tblpart2 Hive Session ID = 78fae85f-a451-4110-bea6-4aa1c172e282),(2,'b','d'),(3,'c','d'),(3,'c','a'),(4,'d','e');

Riferimento

Dialetto Hive in Apache Flink
Apache, Apache Flink, Flink e i nomi dei progetti open source associati sono marchi della Apache Software Foundation (ASF).

Condividi tramite

Dialetto Hive nei cluster Apache Flink® su HDInsight su Azure Kubernetes Service (AKS)

Introduzione

Come usare il dialetto Hive in Flink

Riferimento

Risorse aggiuntive