Диалект Hive в кластерах Apache Flink® в HDInsight в AKS

Статья
04/17/2024

Внимание

Эта функция в настоящее время доступна для предварительного ознакомления. Дополнительные условия использования для предварительных версий Microsoft Azure включают более юридические термины, применимые к функциям Azure, которые находятся в бета-версии, в предварительной версии или в противном случае еще не выпущены в общую доступность. Сведения об этой конкретной предварительной версии см. в статье Azure HDInsight в предварительной версии AKS. Для вопросов или предложений функций отправьте запрос на AskHDInsight с подробными сведениями и следуйте за нами для получения дополнительных обновлений в сообществе Azure HDInsight.

В этой статье описано, как использовать диалект Hive в кластерах Apache Flink в HDInsight в AKS.

Введение

Пользователь не может изменить диалект по умолчанию flink на диалект hive для использования в HDInsight в кластерах AKS. Все операции 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этом предположении не так.

Использование диалекта Hive в Flink

Выполните следующие действия в Webssh:
1. Удаление существующего flink-sql-connector-hive*jar в расположении lib
```
rm /opt/flink-webssh/lib/flink-sql-connector-hive*jar
```
2. Скачайте следующий jar-файл в webssh pod и добавьте его в wget https://aka.ms/hdiflinkhivejdk11jar/opt/flink-webssh/lib. (Указанный выше jar-файл hive имеет исправление 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-диалекта

Выполнение диалекта 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');

Справочные материалы

Диалект Hive в Apache Flink
Apache, Apache Flink, Flink и связанные открытый код имена проектов являются товарными знаками Apache Software Foundation (ASF).

Share via

Диалект Hive в кластерах Apache Flink® в HDInsight в AKS

Введение

Использование диалекта Hive в Flink

Справочные материалы

Дополнительные ресурсы