Teilen über

Hive-Dialekt in Apache Flink-Clustern® in HDInsight auf AKS

  • Artikel

Wichtig

Diese Funktion steht derzeit als Vorschau zur Verfügung. Die zusätzlichen Nutzungsbedingungen für Microsoft Azure-Vorschauen enthalten weitere rechtliche Bestimmungen, die für Azure-Features in Betaversionen, in Vorschauversionen oder anderen Versionen gelten, die noch nicht allgemein verfügbar gemacht wurden. Informationen zu dieser spezifischen Vorschau finden Sie unter Informationen zur Vorschau von Azure HDInsight on AKS. Bei Fragen oder Funktionsvorschlägen senden Sie eine Anfrage an AskHDInsight mit den entsprechenden Details, und folgen Sie uns für weitere Updates in der Azure HDInsight-Community.

In diesem Artikel erfahren Sie, wie Sie den Hive-Dialekt in Apache Flink-Clustern in HDInsight auf AKS verwenden.

Einführung

Benutzer können den flink-Standarddialekt nicht in den Hive-Dialekt für die Verwendung in HDInsight auf AKS-Clustern ändern. Alle SQL-Vorgänge schlagen einmal fehl, wenn die Änderung zu Hive-Dialekt erfolgt ist, wobei der folgende Fehler auftritt.


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

Der Grund für dieses Problem liegt darin, dass Hive Jira geöffnet ist. Derzeit geht Hive davon aus, dass das Systemklassen-Ladeprogramm eine Instanz von URLClassLoader ist. In Java 11 trifft diese Annahme nicht zu.

  • Führen Sie die folgenden Schritte in webssh aus:

    1. Entfernen der vorhandenen flink-sql-connector-hive*jar im Speicherort „lib“ 
      rm /opt/flink-webssh/lib/flink-sql-connector-hive*jar
    2. Laden Sie die folgende JAR-Datei in den Pod webssh herunter, und fügen Sie sie unter „/opt/flink-webssh/lib wget https://aka.ms/hdiflinkhivejdk11jar“ hinzu. (Die oben genannte Hive-JAR-Datei hat den Fix 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. Fügen Sie die folgenden Schlüssel in der flink-Konfigurationsverwaltung im Abschnitt „core-site.xml“ hinzu: 
      fs.azure.account.key.<STORAGE>.dfs.core.windows.net: <KEY>
flink.hadoop.fs.azure.account.key.<STORAGE>.dfs.core.windows.net: <KEY>

  • Hier finden Sie eine Übersicht über Hive-Dialektabfragen

    • Ausführen des Hive-Dialekts in Flink ohne Partitionierung
      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

    Die Daten werden im gleichen Container geschrieben, der im Verzeichnis „hive/warehouse“ konfiguriert ist.

    Screenshot: Containertabelle 1

    • Ausführen des Hive-Dialekts in Flink mit Partitionierung
  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');

Screenshot: Containertabelle 2

Screenshot: Containertabelle 3

Verweis