AKS 上 HDInsight 上的 Apache Flink® 叢集中的 Hive 方言

發行項
09/20/2024

重要

AKS 上的 Azure HDInsight 於 2025 年 1 月 31 日淘汰。透過此公告深入瞭解。

您必須將工作負載移轉至 Microsoft Fabric 或對等 Azure 產品，以避免突然終止工作負載。

重要

這項功能目前為預覽狀態。 Microsoft Azure 預覽版的補充使用規定包含適用於 Beta 版、預覽版或尚未正式發行之 Azure 功能的更合法條款。如需此特定預覽的相關資訊，請參閱 AKS 上的 Azure HDInsight 預覽資訊。如需問題或功能建議，請提交要求 AskHDInsight，並關注我們以獲得 Azure HDInsight 社群的更多更新。

在本文中，瞭解如何在 AKS 上的 HDInsight 上的 Apache Flink 叢集中使用 Hive 方言。

介紹

用戶無法將預設 flink 方言變更為 Hive 方言，以在 AKS 叢集上的 HDInsight 使用。所有 SQL 作業在變更為 Hive 方言之後都會失敗，並出現下列錯誤。


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

此問題的原因是因為 Hive Jira處於開啟狀態。目前，Hive 假設系統類別載入器是 URLClassLoader 的實例。在 Java 11中，此假設不是這樣。

如何在 Flink 中使用 Hive 方言

在 webssh中執行下列步驟：
1. 在 lib 位置移除現有的 flink-sql-connector-hive*jar
```
rm /opt/flink-webssh/lib/flink-sql-connector-hive*jar
```
2. 在 webssh Pod 中下載下列 jar，並將它新增至 /opt/flink-webssh/lib wget https://mvnrepository.com/artifact/org.apache.flink/flink-table-planner-loader/1.17.0之下。（上述 Hive Jar 已含修正 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. 在 [core-site.xml] 區段下的 [flink 組態管理] 中新增下列密鑰：
```
fs.azure.account.key.<STORAGE>.dfs.core.windows.net: <KEY>
flink.hadoop.fs.azure.account.key.<STORAGE>.dfs.core.windows.net: <KEY>
```

以下是 hive 方言查詢的概觀

在 Flink 中執行 Hive 方言而不進行分割

  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

數據會寫入Hive/warehouse目錄中所設定的相同容器中。

在 Flink 中使用分區執行 Hive 方言

  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');

參考

在 Apache Flink 中 Hive 方言
Apache、Apache Flink、Flink 和相關聯的開放原始碼專案名稱是 Apache Software Foundation（ASF）的商標。

共用方式為

AKS 上 HDInsight 上的 Apache Flink® 叢集中的 Hive 方言

介紹

如何在 Flink 中使用 Hive 方言

參考

其他資源