Использование коннектора PXF Iceberg для чтения и записи данных в формате Iceberg между Greengage DB и каталогами Hive Metastore и REST

Подключитесь к REST-каталогу Iceberg и создайте тестовую базу данных:

CREATE DATABASE IF NOT EXISTS customers;

Создайте тестовую таблицу:

CREATE TABLE customers.data (
    id INT,
    first_name VARCHAR(20),
    last_name VARCHAR(20)
);

Заполните созданную таблицу тестовыми данными:

INSERT INTO customers.data 
    VALUES (1, 'John', 'Doe'),
           (2, 'Jane', 'Doe');

Выполните запрос к созданной таблице:

SELECT * FROM customers.data;

1       John    Doe
2       Jane    Doe
Time taken: 0.26 seconds, Fetched 2 row(s)

На мастер-хосте Greengage DB создайте внешнюю таблицу, ссылающуюся на ранее созданную таблицу Iceberg customers.data. В выражении LOCATION укажите PXF-профиль iceberg и конфигурацию сервера. В выражении FORMAT укажите pxfwritable_import — встроенную кастомную функцию форматирования для операций чтения:

CREATE EXTERNAL TABLE customers_r_rest (
    id INT, 
    first_name TEXT, 
    last_name TEXT
    )
    LOCATION ('pxf://customers.data?PROFILE=iceberg&SERVER=iceberg')
    FORMAT 'CUSTOM' (FORMATTER='pxfwritable_import');

Выполните запрос к созданной внешней таблице:

SELECT * FROM customers_r_rest;

 id | first_name | last_name
----+------------+-----------
  1 | John       | Doe
  2 | Jane       | Doe
(2 rows)

На мастер-хосте Greengage DB создайте пишущую внешнюю таблицу, ссылающуюся на ранее созданную таблицу Iceberg customers.data. В выражении LOCATION укажите PXF-профиль iceberg и конфигурацию сервера. В выражении FORMAT укажите pxfwritable_export — встроенную кастомную функцию форматирования для операций записи:

CREATE WRITABLE EXTERNAL TABLE customers_w_rest (
    id INT, 
    first_name TEXT, 
    last_name TEXT
    )
    LOCATION ('pxf://customers.data?PROFILE=iceberg&SERVER=iceberg')
    FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

Вставьте данные в созданную таблицу:

INSERT INTO customers_w_rest (
    id,
    first_name,
    last_name
    )
    VALUES (3, 'Bob', 'Brown'),
           (4, 'Jane', 'Smith');

Подключитесь к REST-каталогу Iceberg и выполните запрос к таблице customers.data:

SELECT * FROM customers.data;

1       John    Doe
2       Jane    Doe
3       Bob     Brown
4       Jane    Smith
Time taken: 0.168 seconds, Fetched 4 row(s)

DELETE FROM customers.data WHERE id IN (3,4);

На мастер-хосте Greengage DB создайте сторонний сервер, использующий обертку сторонних данных iceberg_pxf_fdw и ссылающийся на REST-каталог Iceberg:

CREATE SERVER iceberg_rest_server
    FOREIGN DATA WRAPPER iceberg_pxf_fdw
    OPTIONS (uri 'http://iceberg-rest:8181');

Создайте сопоставление пользователя, которое будет обрабатывать аутентификацию на REST-сервере Iceberg. В выражении SERVER укажите ранее созданный сервер iceberg_rest_server; в выражении OPTIONS укажите параметры аутентификации: идентификатор ключа доступа (s3_access_key_id) и ключ доступа (s3_secret_access_key):

CREATE USER MAPPING FOR CURRENT_USER 
    SERVER iceberg_rest_server
    OPTIONS (s3_access_key_id 'admin', 
             s3_secret_access_key 'password');

Создайте стороннюю таблицу, чья структура совпадает со структурой таблицы customers.data, изначально созданной в REST-каталоге Iceberg. В выражении SERVER укажите ранее созданный сервер iceberg_rest_server; в выражении OPTIONS укажите имя ранее созданной таблицы customers.data:

CREATE FOREIGN TABLE customers_f_rest (
    id INT,
    first_name TEXT,
    last_name TEXT
    )
    SERVER iceberg_rest_server
    OPTIONS (resource 'customers.data');

Выполните запрос к созданной сторонней таблице:

SELECT * FROM customers_f_rest;

 id | first_name | last_name
----+------------+-----------
  1 | John       | Doe
  2 | Jane       | Doe
(2 rows)

Вставьте данные в созданную стороннюю таблицу:

INSERT INTO customers_f_rest (
    id,
    first_name,
    last_name
    )
    VALUES (3, 'Bob', 'Brown'),
           (4, 'Jane', 'Smith');

Выполните повторный запрос к созданной сторонней таблице:

SELECT * FROM customers_f_rest;

 id | first_name | last_name
----+------------+-----------
  1 | John       | Doe
  2 | Jane       | Doe
  3 | Bob        | Brown
  4 | Jane       | Smith
(4 rows)

Подключитесь к REST-каталогу Iceberg и выполните запрос к таблице customers.data:

SELECT * FROM customers.data;

1       John    Doe
2       Jane    Doe
3       Bob     Brown
4       Jane    Smith
Time taken: 0.194 seconds, Fetched 4 row(s)

DELETE FROM customers.data WHERE id IN (3,4);

На хосте HiveClient запустите командную строку Hive:

$ hive

Создайте таблицу Hive customers в базе данных default для хранения тестового набора данных. Выражение STORED BY ICEBERG указывает, что таблица создается в формате Iceberg:

CREATE TABLE customers (
    id INT,
    first_name STRING,
    last_name STRING 
    )
    STORED BY ICEBERG;

Заполните созданную таблицу тестовыми данными:

INSERT INTO customers (
    id,
    first_name,
    last_name
    )
    VALUES (1, 'John', 'Doe'),
           (2, 'Jane', 'Doe');

Выполните запрос к созданной таблице, чтобы убедиться, что данные загружены:

SELECT * FROM customers;

+---------------+-----------------------+----------------------+
| customers.id  | customers.first_name  | customers.last_name  |
+---------------+-----------------------+----------------------+
| 1             | John                  | Doe                  |
| 2             | Jane                  | Doe                  |
+---------------+-----------------------+----------------------+
2 rows selected (0.171 seconds)

На мастер-хосте Greengage DB создайте внешнюю таблицу, ссылающуюся на ранее созданную таблицу Iceberg default.customers. В выражении LOCATION укажите PXF-профиль iceberg и конфигурацию сервера. В выражении FORMAT укажите pxfwritable_import — встроенную кастомную функцию форматирования для операций чтения:

CREATE EXTERNAL TABLE customers_r_hive (
    id INT, 
    first_name TEXT, 
    last_name TEXT
    )
    LOCATION ('pxf://default.customers?PROFILE=iceberg&SERVER=iceberg')
    FORMAT 'CUSTOM' (FORMATTER='pxfwritable_import');

Выполните запрос к созданной внешней таблице:

SELECT * FROM customers_r_hive;

 id | first_name | last_name
----+------------+-----------
  1 | John       | Doe
  2 | Jane       | Doe
(2 rows)

На мастер-хосте Greengage DB создайте пишущую внешнюю таблицу, ссылающуюся на ранее созданную таблицу Iceberg default.customers. В выражении LOCATION укажите PXF-профиль iceberg и конфигурацию сервера. В выражении FORMAT укажите pxfwritable_export — встроенную кастомную функцию форматирования для операций записи:

CREATE WRITABLE EXTERNAL TABLE customers_w_hive (
    id INT, 
    first_name TEXT, 
    last_name TEXT
    )
    LOCATION ('pxf://default.customers?PROFILE=iceberg&SERVER=iceberg')
    FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

Вставьте данные в созданную таблицу:

INSERT INTO customers_w_hive (
    id,
    first_name,
    last_name
    )
    VALUES (3, 'Bob', 'Brown'),
           (4, 'Jane', 'Smith');

На хосте HiveClient выполните запрос к таблице Iceberg customers:

SELECT * FROM customers;

+---------------+-----------------------+----------------------+
| customers.id  | customers.first_name  | customers.last_name  |
+---------------+-----------------------+----------------------+
| 1             | John                  | Doe                  |
| 2             | Jane                  | Doe                  |
| 3             | Bob                   | Brown                |
| 4             | Jane                  | Smith                |
+---------------+-----------------------+----------------------+
4 rows selected (0.371 seconds)

DELETE FROM customers WHERE id IN (3,4);

На мастер-хосте Greengage DB создайте сторонний сервер, использующий обертку сторонних данных iceberg_pxf_fdw и ссылающийся на Hive Metastore-каталог Iceberg. В выражении OPTIONS присвойте параметру catalog_type значение HIVE_METASTORE и укажите URI сервера Hive Metastore и имя Iceberg-каталога Hive Metastore:

CREATE SERVER iceberg_hive_server
    FOREIGN DATA WRAPPER iceberg_pxf_fdw
    OPTIONS (catalog_type 'HIVE_METASTORE', 
             uri 'thrift://hive-metastore:9083', 
             catalog_name 'iceberg_hive');

Создайте сопоставление пользователя, которое будет обрабатывать аутентификацию на сервере Iceberg Hive Metastore. В выражении SERVER укажите ранее созданный сервер iceberg_hive_server:

CREATE USER MAPPING FOR CURRENT_USER
    SERVER iceberg_hive_server;

Создайте стороннюю таблицу, чья структура совпадает со структурой таблицы default.customers, изначально созданной в Hive Metastore-каталоге Iceberg. В выражении SERVER укажите ранее созданный сервер iceberg_hive_server; в выражении OPTIONS укажите имя ранее созданной таблицы default.customers:

CREATE FOREIGN TABLE customers_f_hive (
    id INT, 
    first_name TEXT, 
    last_name TEXT
    )
    SERVER iceberg_hive_server
    OPTIONS (resource 'default.customers');

Выполните запрос к созданной сторонней таблице:

SELECT * FROM customers_f_hive;

 id | first_name | last_name
----+------------+-----------
  1 | John       | Doe
  2 | Jane       | Doe
(2 rows)

Вставьте данные в созданную стороннюю таблицу:

INSERT INTO customers_f_hive (
    id,
    first_name,
    last_name
    )
    VALUES (3, 'Bob', 'Brown'),
           (4, 'Jane', 'Smith');

На хосте HiveClient выполните запрос к таблице customers:

SELECT * FROM customers;

+---------------+-----------------------+----------------------+
| customers.id  | customers.first_name  | customers.last_name  |
+---------------+-----------------------+----------------------+
| 1             | John                  | Doe                  |
| 2             | Jane                  | Doe                  |
| 3             | Bob                   | Brown                |
| 4             | Jane                  | Smith                |
+---------------+-----------------------+----------------------+
4 rows selected (0.132 seconds)

DELETE FROM customers WHERE id IN (3,4);