Создание хранилища данных на AWS: Apache Iceberg, AWS Glue, Athena и DuckDB

Использование Apache Iceberg на AWS с Athena, Glue/Spark и DuckDB

Делиться

Создание хранилища данных не обязательно должно быть таким уж сложным. В этой статье я покажу вам, как разработать базовый, «стартовый» вариант, использующий таблицу Iceberg в хранилище AWS S3. После регистрации таблицы в AWS Glue вы сможете выполнять запросы к ней и выполнять её мутацию из Amazon Athena, в том числе с помощью:

• Запросы о путешествиях во времени ,

• Объединение, обновление и удаление данных

• Оптимизация и очистка таблиц.

Я также покажу вам, как локально проверять те же таблицы из DuckDB, и мы также увидим, как использовать Glue/Spark для вставки дополнительных табличных данных.

Наш пример, возможно, и простой, но он продемонстрирует настройку, различные инструменты и процессы, которые можно внедрить для создания более обширного хранилища данных. У всех современных поставщиков облачных услуг есть аналоги сервисов AWS, о которых я говорю в этой статье, поэтому вам должно быть довольно просто воспроизвести то, что я здесь обсуждаю, в Azure, Google Cloud и других сервисах.

Чтобы убедиться, что мы все понимаем одинаково, вот краткое объяснение некоторых ключевых технологий, которые мы будем использовать.

AWS Клей/Искра

AWS Glue — это полностью управляемый бессерверный ETL-сервис от Amazon, который оптимизирует подготовку и интеграцию данных для аналитики и машинного обучения. Он автоматически обнаруживает и каталогизирует метаданные из различных источников, таких как S3, в централизованном хранилище данных. Кроме того, он может создавать настраиваемые скрипты Spark ETL на Python для выполнения этих задач на масштабируемой бессерверной платформе Apache Spark. Это делает его отличным решением для создания озёр данных на Amazon S3, загрузки данных в хранилища, такие как Amazon Redshift, а также очистки и преобразования данных. И всё это без необходимости управления инфраструктурой.

AWS Афина

AWS Athena — это интерактивный сервис запросов, который упрощает анализ данных непосредственно в Amazon S3 с использованием стандартного SQL. Благодаря бессерверной платформе нет необходимости в управлении или выделении серверов; просто укажите Athena доступ к данным S3, определите схему (обычно с помощью AWS Glue) и начните выполнять SQL-запросы. Сервис часто используется для анализа по запросу, создания отчетов и исследования больших наборов данных в таких форматах, как CSV, JSON, ORC или Parquet.

Таблицы айсберга

Таблицы Iceberg — это открытый формат таблиц для наборов данных, предоставляющий возможности баз данных для данных, хранящихся в озёрах данных, таких как объектное хранилище Amazon S3. Традиционно в S3 можно создавать, читать и удалять объекты (файлы), но их обновление невозможно. Формат Iceberg устраняет это ограничение, а также предлагает другие преимущества, включая транзакции ACID, эволюцию схемы, скрытое секционирование и возможности перемещения во времени.

DuckDB

DuckDB — это аналитическая база данных в оперативной памяти, написанная на C++ и предназначенная для аналитических SQL-нагрузок. С момента своего выпуска пару лет назад она приобрела большую популярность и стала одним из ведущих инструментов обработки данных, используемых инженерами и специалистами по данным, благодаря своей базе SQL, производительности и универсальности.

Обзор сценария

Допустим, вам поручено создать небольшую аналитическую таблицу для заказов, которая будет работать на уровне склада, но вы пока не хотите внедрять тяжёлую платформу. Вам потребуется:

• Безопасная запись (без сломанных читателей, без частичных фиксаций)
• Изменения на уровне строк (ОБНОВЛЕНИЕ/УДАЛЕНИЕ/СЛИЯНИЕ, а не только добавление)
• Чтения на определенный момент времени (для аудита и отладки)
• Локальная аналитика на основе точных данных о производстве для быстрых проверок

Что мы построим

1. Создание таблицы Iceberg в Glue и S3 с помощью Athena
2. Загрузка и изменение строк ( INSERT/UPDATE/DELETE/MERGE )
3. Путешествие во времени к предыдущим снимкам (по временной метке и идентификатору снимка)
4. Ускоряйте работу с помощью OPTIMIZE и VACUUM
5. Прочитать ту же таблицу локально из DuckDB (доступ S3 через DuckDB Secrets)
6. Посмотрите, как добавлять новые записи в нашу таблицу с помощью кода Glue Spark.

Итак, вкратце, мы будем использовать:-

• S3 для хранения данных
• Каталог Glue для метаданных/обнаружения таблиц
• Athena для бессерверных операций чтения и записи SQL
• DuckDB для дешевой локальной аналитики по той же таблице Iceberg
• Искра для обработки грунта

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

Настройка среды разработки

Я предпочитаю изолировать локальные инструменты в отдельной среде. Используйте для этого любой удобный вам инструмент; я покажу на примере conda, поскольку обычно так и делаю. Для демонстрации я запущу весь код в среде Jupyter Notebook.

# создаем и активируем локальную среду conda create -n iceberg-demo python=3.11 -y conda activate iceberg-demo # устанавливаем duckdb CLI + пакет Python и awscli для быстрого тестирования pip install duckdb awscli jupyter

Предпосылки

Поскольку мы будем использовать сервисы AWS, вам понадобится учётная запись AWS. Кроме того,

• Контейнер S3 для озера данных (например, s3://my-demo-lake/warehouse/)
• База данных Glue (мы ее создадим)
• Athena Engine версии 3 в вашей рабочей группе
• Роль IAM или пользователь для Athena с разрешениями S3 + Glue

1/ Настройка Athena

После входа в AWS откройте Athena в консоли и укажите рабочую группу, версию движка и расположение выходных данных S3 (для результатов запросов). Для этого найдите значок меню в виде гамбургера в левом верхнем углу главного экрана Athena. Щелкните по нему, чтобы открыть новый блок меню слева. Там вы увидите ссылку «Администрирование» -> «Рабочие группы». Вам автоматически будет назначена основная рабочая группа. Вы можете оставить её или создать новую. Какой бы вариант вы ни выбрали, отредактируйте его и убедитесь, что выбраны следующие параметры.

• Аналитический движок — Athena SQL. Вручную установите версию движка 3.0.
• Выберите конфигурацию результатов запроса, управляемую клиентом, и введите требуемую информацию о контейнере и учетной записи.

2/ Создайте таблицу Iceberg в Athena

Мы будем хранить события заказов и позволим Iceberg прозрачно управлять разделением. Я буду использовать «скрытый» раздел по дате, указанной в метке времени, для распределения операций записи/чтения. Вернитесь на домашнюю страницу Athena и откройте редактор SQL-запросов Trino. Ваш экран должен выглядеть так.

Введите и выполните следующий SQL-код. Измените названия контейнеров/таблиц по своему усмотрению.

— Это автоматически создает базу данных Glue — если у вас ее еще нет, CREATE DATABASE IF NOT EXISTS analytics; CREATE TABLE analytics.sales_iceberg ( order_id bigint, customer_id bigint, ts timestamp, status string, amount_usd double ) PARTITIONED BY (day(ts)) LOCATION 's3://your_bucket/warehouse/sales_iceberg/' TBLPROPERTIES ( 'table_type' = 'ICEBERG', 'format' = 'parquet', 'write_compression' = 'snappy' )

3) Загрузка и изменение данных (ВСТАВКА / ОБНОВЛЕНИЕ / УДАЛЕНИЕ / СЛИЯНИЕ)

Athena поддерживает настоящий Iceberg DML, позволяя вставлять строки, обновлять и удалять записи, а также обновлять и вставлять данные с помощью оператора MERGE. В основе Iceberg лежит ACID на основе снимков с удалением файлов; читатели сохраняют согласованность, а писатели работают параллельно.

Посейте несколько рядов.

ВСТАВИТЬ В analytics.sales_iceberg ЗНАЧЕНИЯ (101, 1, временная метка «2025-08-01 10:00:00», «создано», 120,00), (102, 2, временная метка «2025-08-01 10:05:00», «создано», 75,50), (103, 2, временная метка «2025-08-02 09:12:00», «создано», 49,99), (104, 3, временная метка «2025-08-02 11:47:00», «создано», 250,00);

Быстрая проверка на благонадежность.

ВЫБРАТЬ * ИЗ analytics.sales_iceberg УПОРИТЬ ПО order_id; order_id | customer_id | ts | status | amount_usd ———-+————+————————+———-+———— 101 | 1 | 2025-08-01 10:00:00 | создано | 120,00 102 | 2 | 2025-08-01 10:05:00 | создано | 75,50 103 | 2 | 2025-08-02 09:12:00 | создано | 49,99 104 | 3 | 2025-08-02 11:47:00 | создано | 250,00

Обновить и удалить.

ОБНОВЛЕНИЕ analytics.sales_iceberg SET status = 'paid' WHERE order_id IN (101, 102) — удаляет заказ 103 УДАЛИТЬ ИЗ analytics.sales_iceberg WHERE status = 'created' AND amount_usd < 60

Идемпотентные обновления с MERGE

Давайте будем считать заказ 104 возвращенным и создадим новый заказ 105.

ОБЪЕДИНИТЬ С analytics.sales_iceberg AS t USING ( VALUES (104, 3, временная метка '2025-08-02 11:47:00', 'refunded', 250.00), (105, 4, временная метка '2025-08-03 08:30:00', 'created', 35.00) ) AS s(order_id, customer_id, ts, status, amount_usd) ON s.order_id = t.order_id ЕСЛИ СОВПАДАЕТ, ТО ОБНОВИТЬ SET customer_id = s.customer_id, ts = s.ts, status = s.status, amount_usd = s.amount_usd ЕСЛИ НЕ СОВПАДАЕТ, ТО ВСТАВИТЬ (order_id, customer_id, ts, status, amount_usd) ЗНАЧЕНИЯ (s.order_id, s.customer_id, s.ts, s.status, s.amount_usd);

Теперь вы можете выполнить повторный запрос, чтобы увидеть: 101/102 → оплачено , 103 удалено, 104 → возмещено и 105 → создано. (Если вы запускаете это в «реальной» учётной записи, вы заметите, что количество объектов S3 растёт — подробнее об обслуживании чуть позже.)

ВЫБЕРИТЕ * ИЗ analytics.sales_iceberg УПОРЯДОЧИТЬ ПО order_id # order_id customer_id ts status amount_usd 1 101 1 2025-08-01 10:00:00.000000 оплачено 120.0 2 105 4 2025-08-03 08:30:00.000000 создано 35.0 3 102 2 2025-08-01 10:05:00.000000 оплачено 75.5 4 104 3 2025-08-02 11:47:00.000000 возвращено 250.0

4) Путешествие во времени (и путешествие между версиями)

Именно здесь раскрывается настоящая ценность Iceberg. Вы можете запросить таблицу в её состоянии на определённый момент времени или по определённому идентификатору снимка. В Athena используйте следующий синтаксис:

— Путешествие во времени к полудню 2 августа (UTC) SELECT order_id, status, amount_usd FROM analytics.sales_iceberg FOR TIMESTAMP AS OF TIMESTAMP '2025-08-02 12:00:00 UTC' ORDER BY order_id; — Или путешествие по версии (замените идентификатор на фактический идентификатор снимка из вашей таблицы) SELECT * FROM analytics.sales_iceberg FOR VERSION AS OF 949530903748831860;

Чтобы получить различные идентификаторы версий (снимков), связанные с определенной таблицей, используйте этот запрос.

ВЫБРАТЬ * ИЗ «analytics».»sales_iceberg$snapshots» УПОРЯДОЧИТЬ ПО commited_at DESC;

5) Поддержание здоровья ваших данных: оптимизация и очистка

Запись на уровне строк (UPDATE/DELETE/MERGE) создаёт множество файлов удаления и может фрагментировать данные. Два оператора обеспечивают скорость и экономию места:

• ОПТИМИЗИРУЙТЕ… ПЕРЕЗАПИШИТЕ ДАННЫЕ С ПОМОЩЬЮ BIN_PACK — сжимает небольшие/фрагментированные файлы и сворачивает удаленные данные в данные
• VACUUM — удаляет старые снимки и очищает ненужные файлы

— сжать «горячие» данные (за вчера) и объединить удаленные данные ОПТИМИЗИРОВАТЬ analytics.sales_iceberg ПЕРЕЗАПИСАТЬ ДАННЫЕ С ПОМОЩЬЮ BIN_PACK ГДЕ ts >= date_trunc('day', current_timestamp — interval '1' day); — удалить старые снимки и удалить ненужные файлы VACUUM analytics.sales_iceberg;

6) Локальная аналитика с DuckDB (только чтение)

Здорово иметь возможность проверять работоспособность производственных таблиц с ноутбука, не запуская кластер. Расширения DuckDB httpfs + iceberg упрощают эту задачу.

6.1 Установка и загрузка расширений

Откройте блокнот Jupyter и введите следующее.

# httpfs обеспечивает поддержку S3; iceberg добавляет читателей Iceberg. import duckdb as db db.sql(«install httpfs; load httpfs;») db.sql(«install iceberg; load iceberg;»)

6.2. Предоставьте учетные данные S3 в DuckDB «правильным» способом (Секреты)

В DuckDB есть небольшой, но мощный менеджер секретов. Самая надёжная конфигурация в AWS — это поставщик цепочки учётных данных , который повторно использует всё, что может найти AWS SDK (переменные среды, роль IAM и т. д.). Поэтому вам необходимо убедиться, что, например, ваши учётные данные AWS CLI настроены.

db.sql(«»»СОЗДАТЬ СЕКРЕТ ( ТИП s3, ПРОВАЙДЕР credential_chain )»»»)

После этого любые операции чтения s3://… в этом сеансе DuckDB будут использовать секретные данные.

6.3 Укажите DuckDB на метаданные таблицы Iceberg

Самый явный способ — сослаться на конкретный файл метаданных (например, на последний в папке metadata/ вашей таблицы:)

Чтобы получить их список, используйте этот запрос

result = db.sql(«»» SELECT * FROM glob('s3://your_bucket/warehouse/**') ORDER BY file «»») print(result) … … s3://your_bucket_name/warehouse/sales_iceberg/metadata/00000-942a25ce-24e5-45f8-ae86-b70d8239e3bb.metadata.json │ s3://your_bucket_name/warehouse/sales_iceberg/metadata/00001-fa2d9997-590e-4231-93ab-642c0da83f19.metadata.json │ s3://имя_вашего_ведра/склад/sales_iceberg/metadata/00002-0da3a4af-64af-4e46-bea2-0ac450bf1786.metadata.json │ s3://имя_вашего_ведра/склад/sales_iceberg/metadata/00003-eae21a3d-1bf3-4ed1-b64e-1562faa445d0.metadata.json │ s3://имя_вашего_ведра/склад/sales_iceberg/metadata/00004-4a2cff23-2bf6-4c69-8edc-6d74c02f4c0e.metadata.json … … …

Найдите файл metadata.json с наибольшим начальным номером имени файла , в моём случае 00004. Затем вы можете использовать его в таком запросе для получения последней позиции вашей базовой таблицы.

# Использовать файл метаданных с наибольшим номером (в моем случае это, похоже, последний файл 00004) result = db.sql(«»» SELECT * FROM iceberg_scan('s3://your_bucket/warehouse/sales_iceberg/metadata/00004-4a2cff23-2bf6-4c69-8edc-6d74c02f4c0e.metadata.json') LIMIT 10 «»») print(result) ┌──────────┬──────────────────┬─────────────────────────────┬─────────────────┐ │ order_id │ customer_id │ ts │ status │ amount_usd │ │ int64 │ int64 │ timestamp │ varchar │ double │ ├──────────┼────────────────┼───────────────────────┼───────────────┼──────────────┤ │ 105 │ 4 │ 2025-08-03 08:30:00 │ создано │ 35,0 │ │ 104 │ 3 │ 2025-08-02 11:47:00 │ возвращено │ 250,0 │ │ 101 │ 1 │ 2025-08-01 10:00:00 │ оплачено │ 120,0 │ │ 102 │ 2 │ 2025-08-01 10:05:00 │ оплачено │ 75,5 │ └──────────┴────────────────┴──────────────────────────────┴────────────────┘

Нужен конкретный снимок? Воспользуйтесь этим, чтобы получить список.

result = db.sql(«»» SELECT * FROM iceberg_snapshots('s3://your_bucket/warehouse/sales_iceberg/metadata/00004-4a2cff23-2bf6-4c69-8edc-6d74c02f4c0e.metadata.json') «»») print(«Доступные моментальные снимки:») print(result) Доступные моментальные снимки: ┌────────────────────┬─────────────────────────┬──────── ──────────────────┬─────────────────────────────────── ───────────────────────────────────────────────── ───────────────────────────────────────────────────────┐ │ порядковый номер │ идентификатор_снимка │ временная_метка_мс │ список_манифестов │ │ uint64 │ uint64 │ временная_метка │ varchar │ ├───────────────────┼───────────────────────┼────────── ──────────────────┼───────────────────────────────── ───────────────────────────────────────────────── ────────────────────────────────────────────────────────────┤ │ 1 │ 5665457382547658217 │ 2025-09-09 10:58:44.225 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-5665457382547658217-1-bb7d0497-0f97-4483-98e2-8bd26ddcf879.avro │ │ 3 │ 8808557756756599285 │ 2025-09-09 11:19:24.422 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-8808557756756599285-1-f83d407d-ec31-49d6-900e-25bc8d19049c.avro │ │ 2 │ 31637314992569797 │ 2025-09-09 11:08:08.805 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-31637314992569797-1-000a2e8f-b016-4d91-9942-72fe9ddadccc.avro │ │ 4 │ 4009826928128589775 │ 2025-09-09 11:43:18.117 │ s3://your_bucket/warehouse/sales_iceberg/metadata/snap-4009826928128589775-1-cd184303-38ab-4736-90da-52e0cf102abf.avro │ └────────────────────┴──────────────────────────┴──────── ───────────────────┴─────────────────────────────────── ───────────────────────────────────────────────── ────────────────────────────────────────────────────────┘

7) Дополнительно: написание в Spark/Glue

Если вы предпочитаете Spark для записи больших объёмов данных, Glue может читать и записывать таблицы Iceberg, зарегистрированные в каталоге Glue. Вы, вероятно, всё равно захотите использовать Athena для ad-hoc SQL, перемещения во времени и обслуживания, но большие CTAS/ETL-задания можно реализовать через задания Glue. (Только учтите, что совместимость версий и разрешения AWS LakeFormation могут быть проблемой, поскольку Glue и Athena могут немного тормозить в версиях Iceberg.)

Вот пример кода Glue Spark, который вставляет несколько новых строк данных, начиная с order_id = 110, в нашу существующую таблицу. Перед запуском необходимо добавить следующий параметр задания Glue (в разделе «Сведения о задании Glue» -> «Дополнительные параметры» -> «Параметры задания»).

Ключ: —conf Значение: spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions import sys import random from datetime import datetime from pyspark.context import SparkContext from awsglue.utils import getResolvedOptions from awsglue.context import GlueContext from awsglue.job import Job from pyspark.sql import Row # ——————————————————— # Инициализация задания Glue # ——————————————————— args = getResolvedOptions(sys.argv, ['JOB_NAME']) sc = SparkContext() glueContext = GlueContext(sc) spark = glutContext.spark_session job = Job(glueContext) job.init(args['JOB_NAME'], args) # ——————————————————— # Force Iceberg + Конфигурации каталога Glue (только динамические) # ——————————————————— spark.conf.set(«spark.sql.catalog.glue_catalog», «org.apache.iceberg.spark.SparkCatalog») spark.conf.set(«spark.sql.catalog.glue_catalog.warehouse», «s3://your_bucket/warehouse/») spark.conf.set(«spark.sql.catalog.glue_catalog.catalog-impl», «org.apache.iceberg.aws.glue.GlueCatalog») spark.conf.set(«spark.sql.catalog.glue_catalog.io-impl», «org.apache.iceberg.aws.s3.S3FileIO») spark.conf.set(«spark.sql.defaultCatalog», «glue_catalog») # ——————————————————— # Отладка: вывести список каталогов для подтверждения регистрации glue_catalog # ——————————————————— print(«Текущие доступные каталоги:») spark.sql(«ПОКАЗАТЬ КАТАЛОГИ»).show(truncate=False) # ——————————————————— # Чтение существующей таблицы Iceberg (необязательно) # ——————————————————— existing_table_df = glueContext.create_data_frame.from_catalog( database=»analytics», table_name=»sales_iceberg» ) print(«Существующая схема таблицы:») exist_table_df.printSchema() # ——————————————————— # Создание 5 новых записей # ——————————————————— new_records_data = [] for i in range(5): order_id = 110 + i record = { «order_id»: order_id, «customer_id»: 1000 + (i % 10), «price»: round(random.uniform(10.0, 500.0), 2), «created_at»: datetime.now(), «status»: «completed» } new_records_data.append(record) new_records_df = spark.createDataFrame([Row(**r) for r in new_records_data]) print(f»Creating {new_records_df.count()} new records:») new_records_df.show() # Регистрация временного представления для вставки SQL new_records_df.createOrReplaceTempView(«new_records_temp») # ——————————————————— # Вставка в таблицу Iceberg (при необходимости псевдонимы столбцов) # ——————————————————— spark.sql(«»» INSERT INTO analytics.sales_iceberg (order_id, customer_id, ts, status, amount_usd) SELECT order_id, customer_id, created_at AS ts, status, price AS amount_usd FROM new_records_temp «»») print(» Успешно добавлено 5 новых записей в analytics.sales_iceberg») # ——————————————————— # Зафиксировать задание Glue # ——————————————————— job.commit()

Перепроверьте у Афины.

выберите * из analytics.sales_iceberg заказ по order_id # order_id customer_id ts статус amount_usd 1 101 1 2025-08-01 10:00:00.000000 оплачено 120.0 2 102 2 2025-08-01 10:05:00.000000 оплачено 75.5 3 104 3 2025-08-02 11:47:00.000000 возвращено 250.0 4 105 4 2025-08-03 08:30:00.000000 создано 35.0 5 110 1000 2025-09-10 16:06:45.505935 завершено 248.64 6 111 1001 2025-09-10 16:06:45.505947 завершено 453.76 7 112 1002 2025-09-10 16:06:45.505955 завершено 467.79 8 113 1003 2025-09-10 16:06:45.505963 завершено 359.9 9 114 1004 2025-09-10 16:06:45.506059 завершено 398.52

Будущие шаги

Отсюда вы можете:

• Создайте больше таблиц с данными.

• «Добавить эволюцию схемы»

• Экспериментируйте с эволюцией разделов (например, меняйте разделы таблицы с дня на час по мере роста объемов),

• Добавьте запланированное обслуживание. Например, EventBridge , Step и Lambdas можно использовать для запуска OPTIMIZE/VACUUM по расписанию .

Краткое содержание

В этой статье я попытался представить чёткий план создания хранилища данных Iceberg на AWS. Он должен послужить руководством для инженеров данных, которые хотят объединить простое объектное хранилище со сложными корпоративными хранилищами данных.

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

Используя Apache Iceberg в облачной системе хранения данных, такой как Amazon S3 , я продемонстрировал, как преобразовывать статические файлы в динамические управляемые таблицы, поддерживающие транзакции ACID, мутации на уровне строк (MERGE, UPDATE, DELETE) и перемещения во времени, и все это без выделения отдельного сервера.

Я также показал, что с помощью новых аналитических инструментов, таких как DuckDB , можно локально читать небольшие и средние озёра данных. А когда объёмы данных вырастают и становятся слишком большими для локальной обработки, я показал, насколько легко перейти на платформу обработки данных корпоративного класса, такую как Spark .

Источник: towardsdatascience.com