لهجة Hive في مجموعات Apache Flink® على HDInsight على AKS

مقالة
04/17/2024

هام

هذه الميزة في وضع المعاينة حاليًا. تتضمن شروط الاستخدام التكميلية لمعاينات Microsoft Azure المزيد من الشروط القانونية التي تنطبق على ميزات Azure الموجودة في الإصدار التجريبي أو قيد المعاينة أو التي لم يتم إصدارها بعد في التوفر العام. للحصول على معلومات حول هذه المعاينة المحددة، راجع معلومات معاينة Azure HDInsight على AKS. للأسئلة أو اقتراحات الميزات، يرجى إرسال طلب على AskHDInsight مع التفاصيل ومتابعتنا لمزيد من التحديثات على مجتمع Azure HDInsight.

في هذه المقالة، تعرف على كيفية استخدام لهجة Hive في مجموعات Apache Flink على HDInsight على AKS.

مقدمة

لا يمكن للمستخدم تغيير اللهجة الافتراضية flink إلى لهجة الخلية لاستخدامه على HDInsight على مجموعات AKS. تفشل جميع عمليات SQL بمجرد تغييرها إلى لهجة الخلية مع الخطأ التالي.


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

ينشأ سبب هذه المشكلة بسبب Hive Jira مفتوح. حاليا، تفترض Apache Hive أن محمل فئة النظام هو مثيل URLClassLoader. في Java 11، هذا الافتراض ليس هو الحال.

كيفية استخدام لهجة Hive في Flink

تنفيذ الخطوات التالية في webssh:
1. إزالة flink-sql-connector-hive*jar الموجود في موقع lib
```
rm /opt/flink-webssh/lib/flink-sql-connector-hive*jar
```
2. قم بتنزيل jar التالي في webssh pod وأضفه ضمن /opt/flink-webssh/lib wget https://aka.ms/hdiflinkhivejdk11jar. (يحتوي جرة الخلية أعلاه على الإصلاح https://issues.apache.org/jira/browse/HIVE-27508)
3. ```
 mv /opt/flink-webssh/lib/flink-table-planner-loader-1.17.0-*.*.*.*.jar /opt/flink-webssh/opt/
```
4. ```
mv /opt/flink-webssh/opt/flink-table-planner_2.12-1.17.0-*.*.*.*.jar /opt/flink-webssh/lib/
```
5. أضف المفاتيح التالية في flink إدارة التكوين ضمن قسم 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>
```

فيما يلي نظرة عامة على استعلامات لهجة الخلية

تنفيذ لهجة Hive في Flink دون تقسيم

  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.

تنفيذ لهجة Hive في Flink مع الأقسام

  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
Apache وApache Flink وFlink وأسماء مشاريع مصدر مفتوح المقترنة هي علامات تجارية ل Apache Software Foundation (ASF).

Share via

لهجة Hive في مجموعات Apache Flink® على HDInsight على AKS

مقدمة

كيفية استخدام لهجة Hive في Flink

المرجع

الموارد الإضافية