Apache Iceberg на практике

В этой главе рассматриваются

• Настройка окружения Apache Iceberg
• Создание таблиц Iceberg в Spark
• Чтение таблиц Iceberg в Dremio
• Построение дашборда бизнес-аналитики

Мы разобрали ключевые идеи, лежащие в основе lakehouse на Apache Iceberg: его архитектуру, компоненты и компромиссы. Эта глава переведёт нас от теории к практике. Наша цель - не просто собрать что-то работающее, а почувствовать, как lakehouse складывается воедино от начала до конца, чтобы при разборе каждого компонента далее в книге у вас была конкретная картина того, как детали стыкуются между собой.

В практическом упражнении этой главы вы развернёте работающее окружение lakehouse на своём ноутбуке. Вы начнёте с настройки окружения, а затем создадите таблицы, выполните запросы и визуализируете результаты. Попутно вы создадите таблицы Iceberg, обратитесь к ним с запросами через Dremio и изучите данные в дашборде бизнес-аналитики (BI). К концу главы у вас будет окружение, с которым можно экспериментировать и к которому можно возвращаться, пока мы разбираем отдельные компоненты в части 2.

3.1 Наш пример

В этой главе используются Dremio и Nessie, потому что с ними легко начать работу с lakehouse и они позволяют сосредоточиться на понятиях Iceberg, а не на инфраструктуре. Кроме того, я хорошо знаю эти инструменты, что поможет мне сделать разбор ясным и практичным. В реальных развёртываниях lakehouse часто строятся на более широком наборе движков и фреймворков, таких как Apache Spark, Apache Flink и Trino. Эти варианты и их место в архитектуре lakehouse мы рассмотрим в следующих главах.

В нашем сценарии мы возьмём данные о торговых транзакциях сети модной розницы - это тот тип операционных данных, который есть уже у большинства организаций. Набор данных включает сведения о товарах и категориях, расположение магазинов, демографию покупателей и маркетинговые кампании. В реальности такие данные могут поступать из нескольких источников: записываться напрямую в базу данных PostgreSQL, передаваться через Kafka и потребляться множеством систем или загружаться в аналитическое хранилище рядом с операционными данными.

Мы возьмём базу данных PostgreSQL как простой и знакомый источник и пройдём базовый рабочий процесс lakehouse (см. рис. 3.1). Мы извлечём данные из PostgreSQL и с помощью Apache Spark запишем их в таблицы Apache Iceberg. Затем мы подготовим эти таблицы в Dremio и определим виртуальные представления для анализа трендов продаж по регионам, сегментам покупателей и эффективности кампаний. Дашборды Apache Superset будут использовать эти представления и давать интерактивные инсайты командам маркетинга, продаж и мерчандайзинга.

Этот пример намеренно прост, чтобы вы могли сосредоточиться на опыте работы с lakehouse, а не на множестве возможных архитектур приёма данных. Описанные паттерны будут работать независимо от того, поступают ли данные пакетными задачами, потоковыми конвейерами или системами, управляемыми событиями; их мы рассмотрим в следующих главах. Код к этой главе можно найти по адресу https://github.com/AlexMercedCoder/architecting-an-apache-iceberg-lakehouse.

Что ж, приступим и поработаем с Apache Iceberg своими руками!

3.2 Настройка окружения Apache Iceberg

Чтобы работать с таблицами Iceberg, мы развернём локальное сквозное окружение lakehouse. С помощью Docker Compose мы поднимем набор необходимых сервисов, представляющих типичный аналитический стек:

1. Операционная база данных - PostgreSQL (в роли транзакционной системы)
2. Слой хранения data lake - MinIO (объектное хранилище, совместимое с S3)
3. Слой приёма и обработки - Apache Spark
4. Слой каталога - Nessie (отслеживает таблицы Iceberg)
5. Движок запросов - Dremio (выполняет быстрые SQL-запросы)
6. BI и визуализация - Apache Superset (дашборды)

База данных выступит нашим источником данных, а остальные перечисленные элементы - компоненты, играющие критическую роль в lakehouse на Iceberg. Развернув их локально, мы сможем наблюдать, как они работают вместе в реальных сценариях. Другие варианты для многих из этих ролей мы рассмотрим при обсуждении каждого слоя архитектуры lakehouse на Iceberg.

На рис. 3.2 показаны компоненты окружения lakehouse на Iceberg, которые мы обсуждали в главах 1 и 2.

3.2.1 Предварительное условие: установка Docker

Прежде чем начать, вам понадобится Docker Desktop, установленный на вашем компьютере. Если его ещё нет, скачайте и установите его бесплатно с docker.com. Docker запускает программы в виртуальных средах, называемых контейнерами, что позволяет использовать Nessie, Apache Spark и Dremio, не устанавливая их напрямую на компьютер.

После установки Docker выполните в терминале следующую команду, чтобы убедиться, что он работает:

docker --version

Вы должны увидеть вывод, похожий на рис. 3.3, - он подтверждает, что Docker установлен.

3.2.2 Создание файла Docker Compose

Теперь опишем наше окружение, включая сервисы, необходимые для запуска data lakehouse на хост-машине: каталог (Nessie), слой хранения (MinIO), инструмент интеграции/федерации (Dremio), инструмент приёма данных (Spark) и инструмент потребления (Superset). Нам также нужно указать, какие образы Docker использовать и какие сетевые порты сопоставить каждому контейнеру, чтобы запустить полноценное окружение lakehouse локально.

В пустом каталоге создайте файл с именем docker-compose.yml и добавьте следующую конфигурацию (полный фрагмент см. в репозитории GitHub: https://github.com/AlexMercedCoder/architecting-an-apache-iceberg-lakehouse/blob/main/ch2/):

version: "3" #1
services: #2
 nessie: #3
   image: projectnessie/nessie:latest
   container_name: nessie
   networks:
     lakehouse-net:
   ports:
     - 19120:19120

1. 1Версия используемого файла Docker Compose
2. 2Список сервисов, составляющих наше окружение
3. 3Наш первый сервис - каталог Nessie

Эта конфигурация Docker Compose описывает сервис, запускающий сервер каталога Project Nessie с бэкендом хранения в памяти. Сервис называется nessie и использует последний образ Project Nessie. Он подключён к пользовательской сети lakehouse-net, что позволяет ему обмениваться данными с другими сервисами в той же сети - вычислительными движками, Apache Spark и Dremio. Контейнер публикует порт 19120 и на хосте, и внутри контейнера, обеспечивая внешний доступ к REST API Nessie. Поскольку эта конфигурация использует хранение в памяти, она подходит для сред разработки или тестирования, где сохранность данных между перезапусками не требуется.

Следующий сервис Docker Compose, minio, описывает сервер MinIO - высокопроизводительную систему объектного хранения, совместимую с Amazon S3 API. В нём заданы переменные окружения для учётных данных root-пользователя и региона (us-east-1, локально). Контейнер работает в сети lakehouse-net, что позволяет ему взаимодействовать с другими компонентами архитектуры lakehouse. Он публикует два порта: 9000 для доступа к S3-совместимому API и 9001 для веб-консоли MinIO. Команда указывает MinIO использовать /data в качестве каталога хранения и привязывает консоль администратора к порту 9001, делая её доступной через браузер. Такая конфигурация удобна для тестирования и прототипирования слоёв объектного хранения в среде, совместимой с Iceberg:

  minio: #1
   image: minio/minio:latest
   container_name: minio
   environment:
     - MINIO_ROOT_USER=admin
     - MINIO_ROOT_PASSWORD=password
     - MINIO_REGION=us-east-1
   networks:
     lakehouse-net:
   ports:
     - 9001:9001
     - 9000:9000
   command: ["server", "/data", "--console-address", ":9001"]

1. 1Сервис хранения нашего data lake на последнем образе MinIO

Следующий сервис Docker Compose, dremio, описывает контейнер для запуска Dremio - высокопроизводительного SQL-движка запросов с открытым исходным кодом для архитектур lakehouse. Он подключён к сети lakehouse-net, что позволяет ему взаимодействовать с другими компонентами, например с MinIO для хранения или Nessie для управления каталогом. Контейнер публикует три порта: 9047 для доступа к интерфейсу и REST API Dremio, 31010 для подключений JDBC-клиентов и 32010 для внутреннего взаимодействия внутри распределённого движка Dremio. При такой настройке Dremio становится центральной платформой интерактивных запросов, способной федерировать данные из множества источников и предоставлять единый SQL-интерфейс поверх таблиц Iceberg:

  dremio: #1
   image: dremio/dremio-oss:latest
   container_name: dremio
   networks:
     lakehouse-net:
   ports:
     - 9047:9047
     - 31010:31010
     - 32010:32010

1. 1Сервис нашего аналитического движка Dremio, который будет выполнять аналитические запросы

Следующий сервис Docker Compose, spark, запускает контейнер Apache Spark, настроенный для разработки и экспериментов с data lakehouse. Он использует образ alexmerced/spark35nb:latest, включающий Spark 3.5 и Jupyter Notebook для интерактивной разработки. Контейнер с именем spark подключён к сети lakehouse-net, что позволяет ему взаимодействовать с другими сервисами, такими как MinIO и Nessie. Он публикует три порта: 8080 для веб-интерфейса мастера Spark, 7077 для порта мастера Spark (используется воркерами и драйверами) и 8888 для доступа к Jupyter Notebook. Переменные окружения задают Spark учётные данные для доступа к S3-совместимому хранилищу AWS, обеспечивая беспрепятственное чтение и запись в MinIO через S3 API. Такая настройка позволяет запускать задачи Spark или интерактивно исследовать таблицы Iceberg через Jupyter, что идеально подходит для локального тестирования и прототипирования:

  spark: #1
   image: alexmerced/spark35nb:latest
   container_name: spark
   networks:
     lakehouse-net:
   ports:
     - 8080:8080  # Master Web UI
     - 7077:7077  # Master Port
     - 8888:8888  # Jupyter Notebook
   environment:
     - AWS_ACCESS_KEY_ID=admin
     - AWS_SECRET_ACCESS_KEY=password
     - AWS_DEFAULT_REGION=us-east-1
     - AWS_REGION=us-east-1

1. 1Сервис нашего движка приёма данных с Apache Spark, который мы будем использовать для записи данных в таблицы Iceberg

Следующий сервис, postgres, настраивает контейнер PostgreSQL, играющий роль операционного источника данных в нашем примере архитектуры lakehouse. Сервис использует официальный образ PostgreSQL latest и создаёт базу данных с именем mydb с учётными данными пользователя, заданными через переменные окружения. Он работает в общей сети lakehouse-net, позволяя другим сервисам - например, Spark или инструментам приёма данных - подключаться для извлечения данных. Контейнер публикует внутренний порт 5432, сопоставленный с портом 5435 на хосте, чтобы избежать конфликтов с другими локальными экземплярами PostgreSQL. Проще говоря, он имитирует транзакционную систему данных, из которой структурированные данные можно периодически или непрерывно принимать в lakehouse для последующей аналитики.

  postgres: #1
   image: postgres:latest
   container_name: postgres
   environment:
     POSTGRES_DB: mydb
     POSTGRES_USER: myuser
     POSTGRES_PASSWORD: mypassword
   networks:
     lakehouse-net:
   ports:
     - "5435:5432"

1. 1Сервис нашей базы данных, представляющий операционный источник данных

Сервис Docker Compose superset разворачивает Apache Superset - BI-платформу с открытым исходным кодом, преднастроенную для работы с Dremio ради визуальной аналитики в окружении lakehouse. Он использует образ alexmerced/dremio-superset, который упрощает интеграцию с Dremio как SQL-движком на бэкенде и включает необходимые зависимости Dremio. Контейнер подключён к сети lakehouse-net, что обеспечивает беспрепятственную связь с такими сервисами, как Dremio и каталог Iceberg. Сервис публикует порт 8088, открывая доступ к веб-интерфейсу Superset для построения дашбордов и исследования наборов данных. Такая конфигурация позволяет обращаться к таблицам Iceberg через Dremio и визуализировать результаты в Superset, замыкая сквозной аналитический процесс:

  superset: #1
   image: alexmerced/dremio-superset
   container_name: superset
   networks:
     lakehouse-net:
   ports:
     - 8088:8088
Networks: #2
 lakehouse-net:

1. 1Сервис нашего BI-инструмента на Apache Superset для создания дашбордов по нашим данным
2. 2Объявляем общую сеть, чтобы все наши сервисы могли взаимодействовать.

3.2.3 Запуск окружения

Теперь, когда у нас есть файл Docker Compose, поднимем окружение lakehouse. В терминале перейдите в каталог, где вы сохранили docker-compose.yml, и выполните команду:

docker-compose up -d

Она сделает следующее:

• Запустит каталог Nessie для отслеживания наших таблиц Iceberg
• Поднимет MinIO - наше объектное хранилище для файлов метаданных и данных Iceberg
• Развернёт Dremio, движок запросов на основе SQL
• Настроит Apache Spark для переноса данных в таблицы Iceberg
• Создаст базу данных PostgreSQL, имитирующую операционную систему
• Инициализирует Apache Superset для построения BI-дашбордов

Проверка, что всё запущено

Когда Docker Compose завершит работу, проверьте запущенные контейнеры:

docker ps

Вы должны увидеть примерно такой вывод:

CONTAINER ID   IMAGE                         STATUS         PORTS                    
xxxxxxxxxx     projectnessie/nessie:latest   Up 5 minutes   19120/tcp                  
xxxxxxxxxx     minio/minio:latest            Up 5 minutes   9000-9001/tcp              
xxxxxxxxxx     dremio/dremio-oss:latest      Up 5 minutes   9047/tcp, 31010-
                                                           32010/tcp  
xxxxxxxxxx     alexmerced/spark35notebook    Up 5 minutes   7077/tcp, 8888/tcp        
xxxxxxxxxx     postgres:latest               Up 5 minutes   5435->5432/tcp            
xxxxxxxxxx     alexmerced/dremio-superset    Up 5 minutes   8088/tcp                  

Если все контейнеры работают, ваш lakehouse на Iceberg готов. На рис. 3.4 показана сеть контейнеров, которую создаёт Docker.

Посмотрим, как эти контейнеры работают вместе. Когда нам нужно принять новые данные, мы запускаем скрипт на Python, который отправляет задачу Spark. В нашем случае Spark читает из Postgres и пишет в наш lakehouse на Iceberg. Задача настраивает Nessie как каталог, а MinIO как слой хранения, так что Spark знает: все файлы данных и метаданных нужно записывать в заданное место в MinIO. Затем каталог обновляется и включает новые метаданные о создаваемой или обновляемой таблице. Dremio использует ту же конфигурацию Nessie и MinIO, поэтому вы можете просматривать наборы данных и создавать логические представления, которые очищают и подготавливают данные под ваш сценарий. Наконец, вы можете подключить к Dremio BI-инструмент, например Apache Superset, и строить дашборды прямо по этим данным.

3.2.4 Доступ к сервисам

Теперь убедимся, что каждый сервис работает:

• MinIO (объектное хранилище) - откройте http://localhost:9001 и войдите с admin/password.
• Dremio (движок запросов) - откройте http://localhost:9047 и создайте учётную запись администратора.
• Spark (обработка данных) - откройте http://localhost:8888, чтобы попасть в Jupyter Notebook.
• PostgreSQL (операционная БД) - подключитесь через клиент PostgreSQL по адресу localhost:5435, имя пользователя myuser, пароль mypassword.
• Superset (BI-дашборды) - откройте http://localhost:8088 и войдите с admin/admin. Superset мы инициализируем позже.

Наше окружение lakehouse поднято и работает - пора применить его к делу.

3.3 Создание таблиц Iceberg в Spark

Теперь, когда окружение lakehouse на Apache Iceberg настроено, пора поработать с реалистичными данными. На этом этапе мы сымитируем данные, которые обычно встречаются в операционных системах, например в транзакционной базе PostgreSQL. Для упражнения мы возьмём розничные торговые транзакции, включающие сведения о товарах, суммы продаж, демографию покупателей и расположение магазинов. Этот набор данных даёт достаточную глубину для последующей аналитики, позволяя исследовать тренды покупательского поведения, эффективность кампаний и результаты по регионам.

Первый шаг - извлечь данные из PostgreSQL. В реальной системе вы подключались бы к продуктивным базам данных, обслуживающим бизнес-приложения: кассовые системы, CRM или платформы электронной коммерции. Здесь мы сымитируем такую среду локально с помощью экземпляра PostgreSQL, наполненного примерами данных о продажах. По ходу дела подумайте, как часто меняются данные и что вы бы использовали: пакетную загрузку, микропакеты или потоковый приём. Эти решения повлияют на вашу архитектуру. К приёму данных мы вернёмся в одной из следующих глав.

Далее мы воспользуемся Apache Spark, чтобы преобразовать данные и загрузить их в таблицы Apache Iceberg. Iceberg поддерживает эволюцию схемы, путешествия во времени и партиционирование, что помогает управлять крупномасштабными, готовыми к аналитике наборами данных в data lakehouse.

После загрузки данных проверьте, что они корректно сохранены в бэкенде объектного хранения (MinIO) и что каталог Iceberg на Nessie зафиксировал каждую операцию. Nessie - это каталог, который отслеживает наши таблицы Iceberg и управляет ими, так что другие инструменты могут обнаруживать и использовать те же таблицы без дополнительной интеграции. Так обычно и работают команды в современных архитектурах lakehouse: они принимают данные из транзакционных источников в открытый, управляемый и масштабируемый формат, а затем используют их для аналитики, отчётности или конвейеров обучения моделей ИИ/ML.

3.3.1 Наполнение базы данных PostgreSQL

Наш контейнер PostgreSQL уже запущен благодаря команде docker-compose up. Загрузим в него примеры данных о продажах, похожие на то, что могла бы порождать операционная система. Затем перенесём эти данные в систему, удобную для аналитики, - lakehouse на Iceberg.

Выполните в терминале следующую команду, чтобы войти в оболочку Postgres:

docker exec -it postgres psql -U myuser mydb

Оказавшись внутри, создайте таблицу fashion_sales_data и наполните её тестовыми данными следующим SQL:

CREATE TABLE fashion_sales (
   id SERIAL PRIMARY KEY,
   product_name VARCHAR(255),
   category VARCHAR(50),
   sales_amount DECIMAL(10, 2),
   sales_date DATE,
   store_location VARCHAR(100),
   customer_age_group VARCHAR(50),
   campaign_name VARCHAR(100)
); #1

INSERT INTO fashion_sales (product_name, category, sales_amount, sales_date, store_location, customer_age_group, campaign_name)
VALUES
   ('Slim Fit Jeans', 'Denim', 89.99, '2024-03-01', 'New York', '18-24', 'Spring Launch'),
   ('Leather Jacket', 'Outerwear', 249.99, '2024-03-01', 'Los Angeles', '25-34', 'Spring Launch'),
   ('Graphic T-Shirt', 'Tops', 39.99, '2024-03-02', 'Chicago', '18-24', 'March Madness'),
   ('Summer Dress', 'Dresses', 129.99, '2024-03-03', 'New York', '35-44', 'March Madness'),
   ('Casual Sneakers', 'Footwear', 99.99, '2024-03-03', 'Los Angeles', '25-34', 'Spring Launch'); #2

1. 1Создаёт таблицу fashion_sales, в которой будут храниться наши данные
2. 2Добавляет несколько записей в таблицу fashion_sales

Выйдите из оболочки Postgres следующей командой:

\q

Теперь в нашей базе данных PostgreSQL есть набор данных, готовый к приёму в таблицы Iceberg.

3.3.2 Запуск Apache Spark

Теперь воспользуемся Apache Spark, чтобы забрать данные из PostgreSQL и записать их в таблицы Iceberg.

Сначала нам нужно место для хранения этих таблиц. В качестве бэкенда объектного хранения мы используем MinIO. Откройте браузер и перейдите по адресу http://localhost:9001, чтобы попасть в консоль MinIO. Войдите с именем пользователя admin и паролем password. После входа создайте новый бакет с именем warehouse. Именно туда Spark будет записывать данные таблиц Iceberg.

Spark уже работает в нашем окружении Docker и включает Jupyter Notebook для интерактивной обработки данных. Откройте браузер и перейдите по адресу http://localhost:8888. Создайте новый Python-ноутбук, чтобы выполнить наш PySpark-скрипт (код Spark на Python).

Вам понадобится IP-адрес контейнера MinIO. Выполните в терминале следующую команду, чтобы найти его:

docker network ls

Она выведет список ваших сетей. Найдите нужное имя сети - оно должно быть сочетанием имени каталога с вашим файлом Docker Compose и имени сети, объявленного в конце этого файла. Затем выполните команду:

docker network inspect <network_name>

Найдите адрес IPv4 контейнера MinIO, как показано на рис. 3.5, и запомните его для следующего шага. Это уникальный адрес контейнера в сети Docker. Обычно достаточно использовать имя контейнера (например, http://minio:9001). Но иногда служба доменных имён (DNS) Docker может неверно разрешать имена контейнеров - с MinIO в этом упражнении такое случается. Чтобы обойти проблему, подключайте PySpark-скрипт к контейнеру MinIO по IP-адресу.

3.3.3 Настройка Apache Spark для работы с Iceberg

Python - самый популярный язык для написания кода Spark, более востребованный, чем исходные API Spark на Scala и Java. В следующем коде на Python мы загрузим библиотеки Spark и зададим переменные с сетевыми адресами каталога и слоя хранения, а также с расположением в S3, где мы хотим хранить данные. Далее мы настроим сессию Spark этими параметрами каталога и хранилища и запустим сессию, чтобы можно было сказать Spark, что делать в рамках задачи.

Сначала вставьте и выполните в ноутбуке Jupyter следующий код на Python, подставив IP-адрес MinIO в указанном месте:

import pyspark 
from pyspark.sql import SparkSession #1

## DEFINE VARIABLES #2
CATALOG_URI = "http://nessie:19120/api/v1"
WAREHOUSE = "s3://warehouse/"
STORAGE_URI = "http://<minio_ip>:9000"

1. 1Импортирует библиотеку pyspark, чтобы отправлять указания Spark
2. 2Задаёт переменные, которые мы используем в конфигурации каталога Iceberg

Этот фрагмент на Python инициализирует переменные конфигурации для конвейера данных PySpark, работающего с окружением lakehouse на Apache Iceberg:

• import pyspark и from pyspark.sql import SparkSession подключают библиотеки PySpark, необходимые для создания сессий Spark и управления ими; сессия - это точка входа для операций с DataFrame в Spark.
• CATALOG_URI задаёт REST-эндпоинт сервиса каталога Nessie, который отвечает за отслеживание метаданных и версий таблиц Iceberg. URI указывает на контейнер Nessie в локальной сети Docker.
• WAREHOUSE указывает корневое расположение в объектном хранилище, где физически будут находиться таблицы Iceberg. Здесь используется S3-совместимый синтаксис пути и указан бакет с именем warehouse, который, как предполагается, размещён в локальном сервисе MinIO.
• STORAGE_URI задаёт базовый адрес для доступа к объектному хранилищу MinIO. Это позволяет Spark подключаться к MinIO по S3 API для чтения и записи данных таблиц. Здесь нам понадобится IP-адрес сервиса MinIO, который мы получили ранее.

Вместе эти переменные говорят Spark использовать Nessie для версионирования таблиц, а MinIO - как совместимый с Iceberg слой хранения.

Следующий фрагмент кода настраивает Spark на приём данных в таблицы Apache Iceberg, используя Project Nessie как версионируемый каталог и MinIO как S3-совместимый бэкенд хранения:

conf = (
   pyspark.SparkConf()
       .setAppName('Iceberg Ingestion')
       .set('spark.jars.packages', 
            'org.postgresql:postgresql:42.7.3,'
            'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0,'
            'org.projectnessie.nessie-integrations:nessie-spark-extensions-3.5_2.12:0.77.1,'
            'software.amazon.awssdk:bundle:2.24.8,'
            'software.amazon.awssdk:url-connection-client:2.24.8')
       .set('spark.sql.extensions', 
            'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,'
            'org.projectnessie.spark.extensions.NessieSparkSessionExtensions')
       .set('spark.sql.catalog.nessie', 'org.apache.iceberg.spark.SparkCatalog')
       .set('spark.sql.catalog.nessie.uri', CATALOG_URI)
       .set('spark.sql.catalog.nessie.ref', 'main')
       .set('spark.sql.catalog.nessie.authentication.type', 'NONE')
       .set('spark.sql.catalog.nessie.catalog-impl', 'org.apache.iceberg.nessie.NessieCatalog')
       .set('spark.sql.catalog.nessie.s3.endpoint', STORAGE_URI)
       .set('spark.sql.catalog.nessie.warehouse', WAREHOUSE)
       .set('spark.sql.catalog.nessie.io-impl', 'org.apache.iceberg.aws.s3.S3FileIO')
) #1

1. 1Создаёт конфигурацию Spark, настраивающую наш каталог Apache Iceberg

Этот код создаёт объект SparkConf и задаёт имя приложения «Iceberg Ingestion». Свойство spark.jars.packages загружает набор необходимых библиотек: JDBC-драйвер PostgreSQL для чтения из источника PostgreSQL, среду выполнения Iceberg для Spark для операций с таблицами, расширения Nessie для Spark для интеграции с каталогом и компоненты AWS SDK для работы с S3-совместимыми хранилищами, такими как MinIO.

Далее код регистрирует расширения сессии Spark, необходимые для Iceberg и Nessie. Затем он определяет логический каталог с именем Nessie и указывает Spark использовать класс Iceberg SparkCatalog с настройками, специфичными для Nessie. К ним относятся REST URI сервера Nessie, ветка-ссылка (main) и режим аутентификации, отключённый для простоты. Каталог настроен на использование специфичной для Nessie реализации Iceberg и на доступ к хранилищу через AWS-совместимый интерфейс S3FileIO, указывающий на заданный эндпоинт MinIO и каталог warehouse.

Вместе эти настройки позволяют задаче Spark записывать данные в таблицы Iceberg, управляемые Nessie, и хранить их в MinIO с полной поддержкой таких возможностей, как путешествия во времени, ветвление и атомарные коммиты.

Следующий фрагмент кода запускает сессию Spark с ранее заданной пользовательской конфигурацией conf, включающей все настройки, необходимые для работы с Apache Iceberg, Project Nessie и MinIO:

spark = SparkSession.builder.config(conf=conf).getOrCreate()
print("Spark Running")

Вызов SparkSession.builder.config(conf=conf).getOrCreate() создаёт новую сессию Spark, если её ещё нет, либо возвращает существующую. Эта сессия - основная точка входа для работы с API Spark, позволяющая выполнять операции чтения из таблиц Iceberg и записи в них. Строка print("Spark Running") подтверждает, что сессия Spark успешно запущена, сигнализируя о готовности окружения к обработке данных.

3.3.4 Загрузка данных из PostgreSQL в Iceberg

Теперь подключимся к PostgreSQL, прочитаем таблицу sales_data и запишем её в таблицу Iceberg в нашем data lake. Здесь MinIO выступит слоем хранения data lake, подобно распределённым файловым системам (например, HDFS) или облачным объектным хранилищам, применяемым в продуктивных средах.

Jupyter Notebook уже работает в окружении контейнеров, которое вы настроили ранее. Откройте браузер и перейдите по адресу http://localhost:8888, чтобы попасть в интерфейс Jupyter. Затем создайте новый ноутбук в каталоге проекта.

Вставьте следующий код в новую ячейку ноутбука и выполните её:

# Define the JDBC connection properties
jdbc_url = "jdbc:postgresql://postgres:5432/mydb" #1
properties = {
   "user": "myuser",
   "password": "mypassword",
   "driver": "org.postgresql.Driver"
} #2

# Read the sales_data table from Postgres into a Spark DataFrame
sales_df = spark.read.jdbc(url=jdbc_url, table="fashion_sales", properties=properties) #3

# Show the first few rows of the dataset
sales_df.show() #4

1. 1Сохраняет строку подключения к базе postgres в переменной
2. 2Свойства подключения к нашей базе данных
3. 3Использует подключение, чтобы загрузить набор данных в датафрейм
4. 4Печатает набор данных, чтобы убедиться в корректной загрузке

Вы должны увидеть вывод с данными о продажах из PostgreSQL, как показано на рис. 3.6.

Теперь запишем эти данные в таблицу Apache Iceberg, отслеживаемую Nessie. Прежде чем создавать таблицу, нужно определить пространство имён. Это логическая группировка, которую каталоги Iceberg используют для организации таблиц. Пространство имён похоже на базу данных или схему в традиционном хранилище данных либо на папку, группирующую связанные наборы данных, но существует исключительно на уровне каталога, а не как физический каталог в хранилище.

Следующие команды сначала создают в Nessie пространство имён, где будет жить наша таблица. Затем они загружают данные из PostgreSQL в DataFrame Spark и записывают их как таблицу Iceberg. В ходе этого процесса файлы данных записываются в MinIO, а метаданные таблицы регистрируются в Nessie. В результате Nessie знает о существовании таблицы, её схеме и расположении метаданных, что позволяет другим инструментам согласованно находить её и обращаться к ней с запросами:

#Create a namespace
spark.sql("CREATE NAMESPACE IF NOT EXISTS nessie.sales;") #1

# Write the DataFrame to an Iceberg table in the Nessie catalog
sales_df.writeTo("nessie.sales.fashion_sales").createOrReplace() #2
# Verify that the data was written to Iceberg by reading the table
spark.read.table("nessie.sales.fashion_sales").show() #3

1. 1Создаёт пространство имён «sales», под которым мы организуем наборы данных
2. 2Записывает данные из Postgres в новую таблицу «fashion_sales» в этом пространстве имён
3. 3Печатает новую таблицу Iceberg, чтобы убедиться в её корректном создании

Если всё прошло успешно, вы увидите те же данные о продажах, но уже сохранённые в Iceberg, как показано на рис. 3.7.

3.3.5 Проверка хранения данных в MinIO

Мы используем MinIO для объектного хранения, поэтому убедимся, что наши данные сохранены правильно:

1. Откройте http://localhost:9001.
2. Войдите с admin/password.
3. Щёлкните по бакету warehouse.
4. Перейдите в sales/sales_data_UUID/.

Вы должны увидеть файлы Parquet с нашими данными, а также файлы метаданных в форматах JSON и Avro, как показано на рис. 3.8. Файлы Parquet - это наши данные, а файлы JSON - файлы метаданных Iceberg на верхнем уровне его структуры метаданных, которую мы разбирали в главе 2. Файлы Avro - это манифесты и списки манифестов внутри этой структуры. Spark принял данные в таблицы Iceberg и сохранил их в формате Parquet в нашем объектном хранилище MinIO.

3.4 Чтение таблиц Iceberg с помощью Dremio

Теперь, когда мы приняли данные в Apache Iceberg с помощью Spark, пора обратиться к нашим таблицам Iceberg с запросами через Dremio - движок запросов для lakehouse, обеспечивающий аналитику прямо на data lake. По ходу книги мы рассмотрим и другие движки, работающие с Apache Iceberg, включая Trino и DuckDB, изучая экосистему Iceberg и подбирая инструменты для каждого слоя lakehouse.

В этом разделе мы подключим Dremio к каталогу Nessie, чтобы вы могли получить доступ к своим таблицам Apache Iceberg и исследовать их. Затем мы обратимся к набору данных о продажах, принятому ранее, с помощью стандартного SQL в интерфейсе Dremio. Вы увидите, что работа с таблицами Iceberg во многом ощущается как работа с таблицами в традиционном хранилище данных - и при этом не нужно перемещать или реплицировать данные только ради того, чтобы построить модель хранилища.

По мере выполнения запросов вы увидите, как Apache Iceberg повышает производительность за счёт таких возможностей, как отсечение по метаданным. Iceberg хранит подробные метаданные таблицы, включая информацию о партициях и статистику на уровне файлов, что позволяет Dremio пропускать ненужные данные при выполнении запроса. Запросы выполняются быстрее даже по мере роста набора данных. К концу этого раздела вы увидите, как Iceberg позволяет вести быструю аналитику прямо на data lake с производительностью уровня хранилища данных, но без сложности и затрат на дублирование данных.

3.4.1 Запуск Dremio

Dremio уже работает как часть нашей конфигурации Docker Compose. Чтобы получить к нему доступ, выполните следующие шаги:

1. Откройте браузер и перейдите по адресу http://localhost:9047.
2. Войдите под учётной записью администратора, созданной ранее.

Откроется интерфейс Dremio. В Dremio можно подключаться к разным источникам данных, выполнять SQL по их наборам данных и даже делать запросы сразу к нескольким источникам (федеративные запросы). Dremio также может подключаться к каталогам lakehouse, таким как Nessie, чтобы быстро находить таблицы Iceberg, которые отслеживает ваш каталог, и работать с ними.

3.4.2 Подключение Dremio к каталогу Nessie

Чтобы обращаться к таблицам Iceberg, Dremio должен подключиться к каталогу Iceberg - например, к Nessie, который мы здесь и используем. Для простоты упражнения мы запустим контейнер Nessie без настройки авторизации. В продуктивной среде вам следует выбрать каталог и включить нужные вам механизмы управления, чтобы контролировать доступ пользователей. Каталоги Iceberg и доступные варианты мы разберём в одной из следующих глав.

Чтобы настроить каталог Nessie, выполните следующие шаги:

1. Нажмите Add Source.
2. Выберите «Nessie» среди доступных типов источников.
3. Введите следующие общие настройки (см. рис. 3.9):
   * Source name: Nessie
   * Nessie endpoint URL: http://nessie:19120/api/v2 ` * Authentication type:None`
4. Введите следующие настройки хранения (см. рис. 3.10):
   * AWS root path: warehouse * AWS access key**:** `admin`
   * AWS secret key: password ``
   * Снимите флажок «Encrypt Connection» (поскольку мы не используем SSL)
5. Введите следующие свойства подключения (см. рис. 3.11). Эти свойства позволяют Dremio подключаться к другим S3-совместимым слоям хранения, таким как MinIO, и находить файлы:
   * fs.s3a.path.style.access = true
   * fs.s3a.endpoint = minio:9000
   * dremio.s3.compat = true
6. Нажмите Save, чтобы добавить источник Nessie.

После добавления источника Dremio автоматически обнаружит таблицы Iceberg, зарегистрированные в Nessie, как показано на рис. 3.12.

3.4.3 Запросы к таблицам Iceberg в Dremio

Теперь, когда Dremio подключён к Nessie, обратимся с запросом к нашему набору данных о продажах в Iceberg. В редакторе SQL Dremio выполните команду:

SELECT * FROM nessie.sales.fashion_sales;

Вы должны увидеть те же данные о продажах, которые мы изначально приняли из PostgreSQL и которые теперь хранятся как таблица Iceberg в MinIO.

Ключевое преимущество Iceberg - отсечение данных на основе метаданных, ускоряющее запросы за счёт пропуска ненужных сканирований. Вместо того чтобы сканировать все данные в каждом файле, движки вроде Dremio опираются на пофайловую статистику из метаданных таблицы и пропускают файлы, где нет подходящих данных. Например, если вы ищете записи, где sales_amount равно 35, движок может пропустить файл, в метаданных которого диапазон значений этого столбца указан как от 38 до 40, потому что подходящих записей там не будет.

Проверка метаданных таблицы

Чтобы просмотреть таблицы метаданных Iceberg, выполните следующее выражение (это специфичный для Dremio синтаксис - синтаксис других движков см. в приложении A):

SELECT * FROM TABLE( table_snapshot( 'nessie.sales.fashion_sales' ) )

Этот вывод перечисляет снимки (исторические версии таблицы), как показано на рис. 3.13. Идентификаторы снимков можно использовать в запросах с путешествием во времени, чтобы обращаться к более ранним версиям таблицы.

Путешествия во времени по снимкам Iceberg

Чтобы совершить путешествие во времени по идентификатору снимка, используйте выражение вида:

SELECT * FROM nessie.sales.sales_data AT SNAPSHOT <snapshot_id>;

А это выражение выполнит путешествие во времени по конкретной метке времени:

SELECT * FROM nessie.sales.sales_data AT TIMESTAMP <timestamp>;

Запросы с путешествием во времени позволяют изучать предыдущие версии таблицы, помогая локализовать ошибки. Путешествия во времени также полезны для аудита и соблюдения требований, поскольку делают исторические данные доступными для проверки.

3.5 Создание BI-дашборда на основе ваших таблиц Iceberg

Теперь, когда мы приняли данные в Apache Iceberg и обратились к ним через Dremio, сделаем шаг дальше и построим BI-дашборд для визуализации инсайтов. В этом разделе мы подключим Apache Superset к Dremio, создадим набор данных на основе нашей таблицы Iceberg и построим интерактивные графики и дашборды. По завершении у вас будет живой BI-дашборд на Dremio и Apache Iceberg, показывающий возможности аналитики на lakehouse.

3.5.1 Запуск Apache Superset

Apache Superset - BI-инструмент с открытым исходным кодом, позволяющий создавать интерактивные дашборды и визуализации на основе источников данных с SQL. Если Superset ещё не запущен, запустите его командой:

docker-compose up -d superset

Затем выполните следующую команду, чтобы инициализировать Superset - она создаст пользователей по умолчанию, таблицы метаданных и роли:

docker exec -it superset superset init

Когда инициализация завершится, откройте браузер и перейдите по адресу http://localhost:8088. Войдите со следующими учётными данными:

• Имя пользователя: admin
• Пароль: admin

Вы попадёте на главную страницу Superset, откуда можно подключиться к Dremio.

3.5.2 Подключение Superset к Dremio

Нам нужно настроить Dremio как источник данных в Superset. Выполните следующие шаги:

1. Нажмите Settings > Database Connections.
2. Нажмите + Add Database.
3. Выберите «Other» в качестве типа базы данных.
4. В поле Dremio URI введите следующее, заменив USERNAME и PASSWORD своими учётными данными Dremio:

dremio+flight://USERNAME:PASSWORD@dremio:32010/?UseEncryption=false

1. Нажмите Test Connection, чтобы убедиться, что всё работает.
2. Нажмите Save.

Dremio подключён к Superset, как показано на рис. 3.14. Этот URI использует для доступа к данным протокол Arrow Flight от Dremio (подробнее о протоколах взаимодействия - в главе 8). Для локальной установки шифрование отключено, но в продуктивной среде следует включить UseEncryption=true ради безопасных соединений.

3.5.3 Создание набора данных из таблиц Iceberg

Теперь, когда подключение готово, создадим набор данных на основе нашей таблицы Iceberg sales_data, как показано на рис. 3.15:

1. Нажмите кнопку + в правом верхнем углу → Create Dataset.
2. Выберите Dremio в качестве базы данных.
3. Перейдите к nessie.sales.fashion_sales.
4. Нажмите Create Dataset.

Теперь наша таблица Iceberg доступна для графиков и дашбордов.

3.5.4 Построение графиков и дашбордов

Теперь визуализируем данные из Iceberg с помощью интерактивных графиков. Apache Superset позволяет легко добавлять наборы данных из ваших источников, строить по ним графики и собирать всё в единый дашборд, которым можно поделиться со всеми, кому нужно видеть эти данные:

1. Нажмите кнопку + → Create Chart.
2. Выберите набор данных fashion_sales.
3. Выберите Line Chart (линейный график).
4. Задайте следующее:
   * X-Axis: sales_date
   * Metric: SUM(sales_amount)
5. Нажмите Run Query, чтобы посмотреть предварительный вид графика.
6. Нажмите Save, назовите его Sales by Day и добавьте на новый дашборд (рис. 3.16).

Создадим ещё один график:

1. Нажмите + Create Chart.
2. Выберите набор данных fashion_sales.
3. Выберите Pie Chart (круговую диаграмму).
4. Задайте следующее:
   * Dimension: category
   * Metric: SUM(sales_amount)
5. Нажмите Run Query, чтобы построить график.
6. Нажмите Save, назовите его Sales by Category и добавьте на дашборд (рис. 3.17).

Теперь, когда мы построили несколько графиков вроде того, что показан на рис. 3.17, объединим их в один дашборд:

1. Нажмите Dashboards → + Create Dashboard
2. Назовите его Sales Analytics.
3. Нажмите + Add Chart и выберите следующее:
   * Sales by Day
   * Sales by Category
4. Измените размер графиков и расположите их.
5. Нажмите Save & Publish.

Теперь у вас есть полноценный BI-дашборд на вашем data lake, похожий на тот, что показан на рис. 3.18! Возможно, вы заметили, что работа с инструментами вроде Apache Superset ничем не отличается от работы с любой таблицей в базе данных или хранилище данных - и в этом весь смысл. Эти таблицы вписываются в ваши существующие процессы в открытом формате и могут использоваться в масштабе всего предприятия без изменения того, как вы потребляете данные. А поскольку это Iceberg, вы можете подменить инструменты на любом шаге, что и демонстрирует ценность модульной экосистемы Apache Iceberg.

Когда закончите, выключите окружение, выполнив следующую команду:

docker compose down -v

Флаг -v в этой команде указывает Docker очистить тома хранения, созданные для каждого контейнера, освобождая место, занятое при их работе.

В следующей главе мы сделаем шаг назад и оценим вашу инфраструктуру данных. Мы также начнём планировать внедрение Iceberg в реальных условиях - от определения архитектурных требований до выбора подходящих компонентов для успешной реализации.

Итоги

• Локальное окружение lakehouse на Apache Iceberg можно развернуть с помощью Docker Compose, подняв такие сервисы, как MinIO, Apache Spark, Nessie, Dremio, Superset и PostgreSQL.
• MinIO выступает слоем объектного хранения для файлов данных и метаданных таблиц Iceberg, локально имитируя облачное хранилище.
• Project Nessie служит каталогом, который отслеживает таблицы Iceberg, чтобы Apache Spark и Dremio могли их обнаруживать.
• Apache Spark извлекает данные о продажах из операционной базы PostgreSQL, преобразует их и загружает в таблицы Iceberg, отслеживаемые Nessie.
• Dremio подключается к каталогу Nessie, чтобы обращаться к таблицам Iceberg напрямую в объектном хранилище и обеспечивать работу ваших аналитических нагрузок.
• Apache Superset можно подключить к Dremio, чтобы создавать интерактивные BI-дашборды, визуализирующие инсайты из окружения lakehouse на Iceberg.