Приветствую.
Цель стать: показать на практике как проходит insert запрос в Apache Iceberg. Все шаги можно и даже нужно повторить.
Что будет в статье: Мы поднимем локально, через Docker, Apache Iceberg и сделаем запрос через Apache Spark. Так же подробнее посмотрим как выполняется запрос и что он после себя оставляет. То есть как именно записывает данные.
Автор данной статьи предполагает, что вы немного в курсе за Apche Iceberg и пришли сюда, чтобы понять эта картинка с архитектурой(об этом чуть позже) работает. Если вам нужен полноценный курс по Apache Iceberg, то моя рекомендация книга Apache Iceberg: The Definitive Guide. Эта статья "вдохновлена" главой номер 3.
Глоссарий
Прежде чем идти дальше, хочется привести краткое определение некоторых терминов, которые я буду использовать. Это будет очень краткое, но достаточное, описание терминов, чтобы можно было понять основную часть статьи.
Hadoop - платформа для распределенной обработки данных. Именно, что платформа. Там множество элементов. В данной статье мы будем подразумевать под Hadoop, HDFS.
HDFS(Hadoop Distributed File System) - распределённая файловая система. Если уж ООООООЧЕНЬ сильно упрощать, то это файловая система с финтифлюшками. То есть просто место, где хранятся много файлов.
S3(Simple Storage Service) - объектное хранилище. Изначально это продукт AWS. И это очередное хранилище, но уже плоское. Предлагаю думать, об этом как место где лежат много файлов. Да, очень похоже на HDFS, но подход немного другой. На данный момент, S3 стало не продуктом, а стандартом. Если ты выполнишь определённые пункты, то ты можешь называться S3-подобным хранилищем. Одном из самых популярных сейчас решений - это MinIO.
MinIO - объектное хранилище данных. По факту, это другая версия S3 от AWS, но заявленное как opensource решение. Но вот на конец 2025 года у нас появилась новость, что разработчики приостановили поддержку ветки opensource. Подробнее на статье habr. Для целей данной статьи предлагаю вам думать о MinIO, как о сервисе, которая просто хранит данные.
Apache Spark - инструмент, который позволяет писать код на python, java, scala для больших данных. Так же он поддерживает sql, но этот sql нужно оформлять в какой-то код.
Trino - sql вдижок. Просто движок, который может выполнять sql запросы к данными. Если классические СУБД - это комбинация хранилища, где лежат данные, и движка, который работает над этими данными, то Trino - это только движок. Его можно подключить к разным СУБД или файловым системам.
Docker - реализация контейнерезации. Эта штука позволяет вам вместо поднятия виртуальной машины, поднимать контейнеры, которые позволяют вам работать локально.
Немного истории
Мы жили себе с нашими СУБД (Oracle, Postresql, Mysql ...) и грузили наши данные в рамках OLAP. Строили своию аналитику и показывали свои дэшборды бизнесу (на самом деле мы и сейчас так делаем).
Потом появился Hadoop и его HDFS. С приходом Hadoop начали говорить, что с Hadoop есть проблемы, но идея того, что данные можно записывать в файлы и обрабатывать их движком, который не закреплён вместе с данными, осталась. Эта идея переросла в концепцию Data Lake. Но и у этого подхода возникли свои проблемы (дублирование файлов, качество данных и т.д.). Потом появился Lake House. Но и у него есть свои проблемы. И вот чтобы решить новые проблемы, которые создали новые технологии, возник Apache Iceberg.
Iceberg не единственно решение, но кажется, что оно становиться стандартом. И именно его мы будем рассматривать сегодня.
Apаche Iceberg
Так, что такое Apache Iceberg?
На официальном сайте проекта сказано:
Iceberg is a high-performance format for huge analytic tables.
Переводится: Iceberg - высоко производительный формат для больших аналитических таблиц (простите за такой перевод, но мне показалось, что это наиболее прямой).
Как по мне это определение не даёт понимание, что всё таки такое Iceberg. Поэтому, предлагаю вам представить такую картину.
На схеме выглядит так, что Iceberg (здесь и далее я буду опускать Apache для экономии времени) - это какая-то программа или сервис. На самом деле это не так. Iceberg это прослойка между движком и данными. По факту это просто библиотека о том, как хранить данные. Не более. Не сервис. Не субд. Просто набор библиотек для каждого движка, который записывает данные.
Так, а что нам даёт Iceberg, чего нет без него? Давайте приведём пару пунктов:
поддержка ACID. Целостность данных страдает если мы используем только HDFS или S3
откат к предыдущему состоянию в таблице
schema evolution(не знаю как это нормально перевести). Изменение названия полей, таблицы ...
...
Опять же все эти красивые слова, которые вы можете видеть на любых презентациях. Вопрос в том как Iceberg это делает. Но в данной статье мы рассмотрим только вставку данных.
Если очень коротко, то это работает так ():
Мы делаем insert запрос через движок(всё равно что это Spark, Trino ...)
Движок понимает, что это запрос в Iceberg (на пример, это делается при создании spark сессии)
Движок подтягивает библиотеки для работы с Iceberg
Движок записывает данные в хранилище (пускай будет S3, но можно и в hadoop)
Движок записывает метаданные, о данных
Да, кажется что все так делают. Каждая СУБД. И да, это очень простой и абсолютно не полный путь, который проходит движок, чтобы записать данные. Но, что стоит отметить, в этом процессе так это метаданные. Это краеугольная часть Iceberg. На этих метаданных всё и держится. По сути именно в этих мета данных раскрывается Iceberg.
Iceberg и его метаданные
Мы уяснили, что эти метаданные очень важны. А что именно записывается в эти метаданные?
Во всех презентациях и видео вы увидите такую картинку с оф сайта:
Попробуем пройтись по основным частям(будто я на собесе и меня просят описать эту картинку).
Пойдём снизу вверх.
Первое, что у нас есть это data files. Это сами данные. По умолчанию это parquet файлы, но можно и другие форматы. Однако на практике, в бою, я встречал именно parquet.
Далее у нас metadata layer. Этот слой начинается с manifest file. В этом файле хранятся данные о данных в parquet(например мин. и макс. значения в столбцах). Так же вы можете видеть, что 1 manifest file может хранить данные о нескольких data files (связь один ко многим. Не зря я все эти слова учил. Пригодились гыыы)
После manifest file у нас идёт manifest list. Тут тоже 1 manifest list может хранить несколько manifest file. Тут есть важный момент. 1 manifest list - это 1 snapshot. ЕЩЁ РАЗ. 1 manifest list - это 1 snapshot. Это важное сведение. Супер важное. Теперь, у вас вопрос: А что такое snapshot?
snapshot - это состояние таблицы, которое нельзя изменить. Каждое изменения состояния таблицы, будь то вставка данных или добавление нового столбца, генерируется новый snapshot. ДА, на каждое изменение. Будьте внимательны.
Идём дальше по схеме. Далее у нас есть metadata file. В этом файле хранятся важная информация о таблице. Как таблица партиционированна, список snapshot-ов для данной таблицы. То есть metadata file - это и есть файл о таблице.
И самый последний слой - это catalog. Catalog - это внешний сервис(postgresql, rest catalog, hive metastore) в котором находится ссылка на metadata file и который является точкой входи для всех внешних движков. Если делается select запрос, то сначала движо�� идёт в catalog. Это входная точка для запроса.
Много буков, но мы тут не за этим. Предлагаю перейти к практике и посмотреть, как эта картинка будет работать на настоящем примере.
Поднимает локально необходимый инструментарий
На официальном сайте есть docker compose файл. Можно было бы им и воспользоваться, но для наших целей, нам нужно будет его немного доработать. Приведу весь файл, с доработкой и проставлю пару комментариев:
services: spark-iceberg: # подключаем spark image: tabulario/spark-iceberg container_name: spark-iceberg build: spark/ networks: iceberg_net: depends_on: - rest - minio volumes: - ./spark/warehouse:/home/iceberg/warehouse - ./spark/notebooks:/home/iceberg/notebooks/notebooks environment: - AWS_ACCESS_KEY_ID=admin - AWS_SECRET_ACCESS_KEY=password - AWS_REGION=us-east-1 ports: - 8888:8888 - 8080:8080 - 10000:10000 - 10001:10001 rest: image: apache/iceberg-rest-fixture # подключаем rest catalog container_name: iceberg-rest networks: iceberg_net: ports: - 8181:8181 environment: - AWS_ACCESS_KEY_ID=admin - AWS_SECRET_ACCESS_KEY=password - AWS_REGION=us-east-1 - CATALOG_WAREHOUSE=s3://warehouse/ - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO - CATALOG_S3_ENDPOINT=http://minio:9000 volumes: - ./rest/tmp:/tmp # Вот это добавлил minio: # объектное хранилище image: minio/minio container_name: minio environment: - MINIO_ROOT_USER=admin - MINIO_ROOT_PASSWORD=password - MINIO_DOMAIN=minio networks: iceberg_net: aliases: - warehouse.minio ports: - 9001:9001 - 9000:9000 volumes: - ./minio/tmp:/data # Вот это добавлил command: ["server", "/data", "--console-address", ":9001"] mc: # интерфейс для подключения к MinIO, чтобы можно было в браузере смотреть depends_on: - minio image: minio/mc container_name: mc networks: iceberg_net: environment: - AWS_ACCESS_KEY_ID=admin - AWS_SECRET_ACCESS_KEY=password - AWS_REGION=us-east-1 entrypoint: | /bin/sh -c " until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done; /usr/bin/mc rm -r --force minio/warehouse; /usr/bin/mc mb minio/warehouse; /usr/bin/mc policy set public minio/warehouse; tail -f /dev/null " networks: iceberg_net:
Я делаю всё под Windows. Так, что добавленный путь может отличатся под linux.
Добавилось 2 volumes для контейнеров rest и minIO. Они нужны, чтобы можно было посмотреть на сами файлы, которые мы генерируем, а так же, чтобы сохранить данные, которые будет использованы в контейнерах. То есть схема будет такой:
Быстрое описание
Apache Spark - будет движком. Он будет делать всю работу. Буду использовать pyspark.
MinIO - объектное хранилище(s3 подобное). Именно здесь будут лежать данные. В этом контейнере мы смонтировали папку, чтобы посмотреть на те данные, которые будут созданы.
rest catalog - собственно catalog из рисунка 2. Тут же будет смонтирована папочка, в которой лежит данные catalog. В этом контейнере мы смонтировали папку, чтобы посмотреть, что будет писаться в iceberg catalog. Так как rest catalog хранит данные в SQLite, то его можно будет открыть как файл.
Да, в docker compose есть ещё и mc контейнер, но для меня это интерфейс к minIO и чтобы не перегружать схему, я его не показал на схеме.
С теорией закончили. Пошли тыкаться.
Пошаговая инструкция
Создаём локально папочку. Всё равно где и как вы её назовёте. Я создал папку с названием "iceberg_tutorial"
Создаём в этой папочке файл с названием "docker-compose.yaml". В него копируем код, который был ранее в статье.
Рядом с этой папкой создаём папку ''tmp'. Эта папка будем смонтирована с папкой в doker контейнере.
Вам нужно скачать docker desktop вот отсюда и установить его. Я верю, что вы справитесь с установкой самостоятельно.
Как только вы установили deocker desktop, вам нужно его запустить. Просто найдите иконку с этим приложением на своём пк. А дальше вам нужно будет запустить docker compose.
Тут небольшое отступление. У меня возникли проблемы с запуском контейнеров через desktop приложение, поэтому я всё буду делать как привык, через консоль.
В терминале перейдите в папку "iceberg_tutorial". Давайте проверим, что docker работает. В этом же терминале введите:
docker ps
Если у вас вывелось:
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
то docker запущен и можно продолжать работать.
Теперь вводим
docker compose up -d
Тут придётся немного подождать пока все образы загрузятся. Как только всё будет запущено, то в папке "iceberg_tutorial" вы увидите, что есть создались несколько папок:
minio
rest
spark
Создание этих папок прописано в docker compose файле. Каждый из них отвечает за свой контейнер.
Сейчас у вас должно уже всё работать. Проверим, что всё запустилось через командную строку
docker ps -a
Все ваши 4 контейнера должны быть в статусе UP
Последние приготовления. Нужно подключить обозреватель для iceberg catalog. Я воспользуюсь dbeaver. Можно скачать тут. Я так же надеюсь, что вы справитесь с установкой.
Устанавливаем соединение с iceberg catalog. Для этого открываем dbeaver. Выбираем новое соединение:
В открывшемся окне нам нужно SqlLite
Откроется такое окно, в котором нужно выбрать кнопку открыть
Выбираем файл
iceberg_tutorial\rest\tmp\iceberg_catalog.db
И нам в dbeaver покажет, что у нас есть 2 таблицы
Нас будет интересовать таблица iceberg_tables. Давайте выполним запрос
select * from iceberg_tables;
Ну и ответ будет пустота
Дальше выполните запрос
select * from iceberg_namespace_properties;
И опять ничего не будет. А чёго ожидали то?
Мы выполнил�� запросы к iceberg catalog. Так как ничего мы ещё не создавали, то и ничего там нет. Но давайте наконец создадим нашу первую таблицу
Переходим по адресу:
Мы если, что его прописывали в docker-compose файле, в разделе ports.
Нам открывается jupyter
Вы можете посмотреть примеры, которые идут с контейнером. Но предлагаю это сделать позже.
Нам нужно создать собственный notebook. Для удобства перейдём в папку notebook и нажмём New, а потом выберем Python3
Вас перебросит на новое окно. Файл, я сразу переименовал в test_iceberg.ipynb. Этот же файл у меня лежит теперь в моей папке на локальном компьютере
В этом ноутбуке пропишем
from pyspark.sql import SparkSession # подключаем библиотеку для работы со sparkSQL spark = SparkSession.builder.appName("Jupyter").getOrCreate() #создаём spark сессию spark #вызываем, чтобы проверить, что оно работает
Если вам вылезет warning, то проигнорируйте его сейчас.
sparkSQL - модуль spark, который позволяет использовать sql для обработки данных
спарк сессия - это... как бы это коротко объяснить? Это точка входа. Грубо говоря в этом мести пробрасывается соединение.
Вы можете спросить: А где тут Iceberg? Тут ничего ни одного слова нет...
Будете правы, потому что все настройки лежат в конфиге. Можете выполнить в notebook команду:
spark.sparkContext._conf.getAll()
Там вы увидите куда мы ходим и т.д. и т.п. Мы не будем рассматривать конфиг, потому, что это не цель данной статьи. Поэтому продолжим.
И так, у нас есть соединение с Iceberg. Давайте создадим схему. Для этого в новой ячейки notebook напишем
%%sql # это для того, чтобы ячейка воспринималась как sql CREATE DATABASE IF NOT EXISTS test_iceberg
Схему создали. Давайте переключимся на dbeaver и выполним запрос:
Select * from iceberg_namespace_properties inp
И появились 2 строки. То есть в этой табличке отображаются схемы.
Теперь давайте создадим в spark таблицу:
%%sql CREATE TABLE test_iceberg.super_pyper_table ( id BIGINT ) USING iceberg
Да, одно поле. Нам больше и не надо.
Прежде чем пойдём дальше, давайте ещё раз посмотрим на архитектуру Iceberg.
Снова делаем запрос на Iceberg Catatlog через dbever
select * from iceberg_tables it;
И у нас появилась строка.
Получилось, мы зарегистрировали в Iceberg Catalog нашу таблицу. Сейчас нас интересует атрибут metadata_location. Его значение:
s3://warehouse/test_iceberg/super_puper_table/metadata/[много буков].metadata.json
И видим, что это полный путь до metadata file. Получается, что Iceberg Catalog хранит связку название таблицы и место где хранится metadata file для неё. Напомню, что metadata file это и есть репрезентация таблицы.
Так а давайте посмотрим на этот файл. Где его взять? Для этого нужно будет открыть MinIO. Напоминаю, что MnIO это объектное хранилище, где лежат наши данные. Давайте перейдём туда открыв:
Login и password прописаны в docker-compose
- MINIO_ROOT_USER=admin - MINIO_ROOT_PASSWORD=password
Нам откроется окно
переходим в test_iceberg и у нас открывается
проваливаемся в эту единственную папку, потом переходим в папку metadata и видим:
Его можно скачать, нажав на кнопку Download в MinIO. Да мы смонтировали папку для этих целей, но MinIO хранит этот файл в виде бинарника.
Так, что просто скачайте этот файл через кнопку Download
Открыв файл, мы увидим такое наполнение
{ "format-version": 2, "table-uuid": "64efcc91-a858-4366-9169-3d849482c43c", "location": "s3://warehouse/test_iceberg/super_pyper_table", "last-sequence-number": 0, "last-updated-ms": 1769868969208, "last-column-id": 1, "current-schema-id": 0, "schemas": [{ "type": "struct", "schema-id": 0, "fields": [{ "id": 1, "name": "id", "required": false, "type": "long" }] }], "default-spec-id": 0, "partition-specs": [{ "spec-id": 0, "fields": [] }], "last-partition-id": 999, "default-sort-order-id": 0, "sort-orders": [{ "order-id": 0, "fields": [] }], "properties": { "owner": "root", "write.parquet.compression-codec": "zstd" }, "current-snapshot-id": -1, "refs": {}, "snapshots": [], "statistics": [], "partition-statistics": [], "snapshot-log": [], "metadata-log": [] }
Тут много инфы. Не будем всё объяснять. Сосредоточимся на как мне кажется, важном.
table-uuid - уникальный идентификатор. Позволяет не привязываться к названию таблицы.
schemas - то какие поля у таблицы. Заметьте, что у поля тоже есть свой id.
current-snapshot-id - номер snapshot, который на данный момент является активным. Напомню, что это номер версии состояния таблицы. Сейчас он -1, так как данных нет. Мы просто создали таблицу.
Больше в папке super_puper_table ничего нет. Нет никаго manifes list, manifest file. Так как данных нет. Но теперь давайте добавим данные. Вернёмся в Jupyter Hub и выполним запрос
%%sql INSERT INTO test_iceberg.super_pyper_table VALUES ( 123 )
Убедимся, что данные пришли и сделаем запрос:
%%sql SELECT * FROM test_iceberg.super_pyper_table;
И данные пришли. Теперь самое важное для чего всё и затевалось. Посмотрим наша схема выглядит в реальности. Я буду идти очень медленно и с повторениями, чтобы было кристально понятно.
Первое что делаем идём в dbeaver, чтобы выполнить select запрос в iceberg catalog
select * from iceberg_tables it ;
Сейчас мы посмотрели в
Нам показывают одну строку, но теперь состояние поменялось. Поменялись 2 поля.
metadata_location
previous_metadata_location
Помните, я писал, что для каждого действия с таблицей, будь-то insert или update создаются собственные мета данные. Так вот. Мы вставили данные, поэтому нам создался новый metadata file. В iceberg catalog прошли изменения и для нашей таблицы записалось, где хранится актуальный metadata file и его предыдущая версия.
Теперь перейдём в MinIO. Если вы перейдёте по пути
warehouse/test_iceberg/super_pyper_table
то увидите, что создалась папка data.
Открыв её, вы увидите файл с вашими данными. Можете скачать его и посмотреть, но там просто parquet файл. Нас больше интересует metadata. Давайте перейдём туда.
У нас появилось 3 новых файла.
00001-9a682068-6cb7-4b19-9bed-df14bd6a5a13.metadata.json - новый metadata file.
То есть в схеме мы открыли:
Если его открыть, то увидите что он похож на предыдущий. Тут важные отличия от предыдущей весии. Заполнился раздел snapshot. В этом разделе указывается вся инфа про snapshot. И дополнительно указывается путь до manifest list:
snap-1374504624696218140-1-d11abf55-85f5-4312-8825-8eea7f5100d2.avro
Сейчас мы откроем слой:
Так как это avro формат, то его просто через редактор не открыть. Я нашёл в интеренет вот такой python скрипт:
import avro.schema from avro.datafile import DataFileReader, DataFileWriter from avro.io import DatumReader, DatumWriter # Assume a file named "users.avro" exists for this example. # If you need to create one first, you can use the code snippets in the search results. # Open the Avro file in binary read mode ('rb') try: # подставьте свой путь до файла reader = DataFileReader(open("snap-1374504624696218140-1-d11abf55-85f5-4312-8825-8eea7f5100d2.avro", "rb"), DatumReader()) print("Contents of the Avro file:") # Iterate over each record in the file for user in reader: # Print the record, which is a Python dictionary print(user) # Close the file reader reader.close() except FileNotFoundError: print("Error: 'users.avro' not found. Please create an Avro file first.") except Exception as e: print(f"An error occurred: {e}")
Которым открыл файл и у меня вывелось:
{ 'manifest_path': 's3://warehouse/test_iceberg/super_pyper_table/metadata/d11abf55-85f5-4312-8825-8eea7f5100d2-m0.avro', 'manifest_length': 6963, 'partition_spec_id': 0, 'content': 0, 'sequence_number': 1, 'min_sequence_number': 1, 'added_snapshot_id': 1374504624696218140, 'added_files_count': 1, 'existing_files_count': 0, 'deleted_files_count': 0, 'added_rows_count': 1, 'existing_rows_count': 0, 'deleted_rows_count': 0, 'partitions': [], 'key_metadata': None }
Тут куча информации, в которой можно покопаться. Самое сейчас важное для нас это:
'manifest_path': 's3://warehouse/test_iceberg/super_pyper_table/metadata/d11abf55-85f5-4312-8825-8eea7f5100d2-m0.avro',
Это путь указывает на наш следующий слой
Manifest file это тоже avro. Так, что если его открыть кодом выше, то мы увидем
{ 'status': 1, 'snapshot_id': 1374504624696218140, 'sequence_number': None, 'file_sequence_number': None, 'data_file': { 'content': 0, 'file_path': 's3://warehouse/test_iceberg/super_pyper_table/data/00000-0-a0dd32cd-829a-45b2-9351-4e1f4ad7bef7-0-00001.parquet', 'file_format': 'PARQUET', 'partition': {}, 'record_count': 1, 'file_size_in_bytes': 436, 'column_sizes': [{'key': 1, 'value': 43}], 'value_counts': [{'key': 1, 'value': 1}], 'null_value_counts': [{'key': 1, 'value': 0}], 'nan_value_counts': [], 'lower_bounds': [{'key': 1, 'value': b'{\x00\x00\x00\x00\x00\x00\x00'}], 'upper_bounds': [ { 'key': 1, 'value': b'{\x00\x00\x00\x00\x00\x00\x00' } ], 'key_metadata': None, 'split_offsets': [4], 'equality_ids': None, 'sort_order_id': 0, 'referenced_data_file': None } }
Нас тут больше всего интересует поле "file_path", которое и указывает на данные в parquet формате.
Это всё что я хотел показать. На самом деле тут ещё много полей для эксперимента, но оставлю это на ваше усмотрение. Можете продолжать экспериментировать.
