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

重要

此功能目前為預覽功能。 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. 在 Pod 中 webssh 下載下列 jar，並在 /opt/flink-webssh/lib wget https://aka.ms/hdiflinkhivejdk11jar底下新增它。 (上述 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-dialect查詢的概 觀

  • 在 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) 的 商標。