Практика: сборка пайплайна данных

~ 60 минут
В прошлом уроке вы познакомились с Data Lakehouse. Вы уже знаете достаточно, чтобы собрать пайплайн на практике. Этим сейчас и займёмся.
Реальный пример, когда это может понадобится - сохранение медицинских снимков пациентов для дальнейшей обработки нейросетями и непосредственно структурированных данных о пациентах.
Схема достаточно простая — мы отправим данные, используя Airflow в Nessie, и поищем их там, убедившись, что они появились как в качестве метаданных в Iceberg, так и в виде файлов в MinIO.
  1. Начнём с базы — установим каталог и хранилище. Запускать будем с помощью docker-compose.

YAML

services:
  #Nessie catalog
  catalog:
    image: projectnessie/nessie: latest
    container_name: catalog
    networks:
      iceberg-net:
    ports:
      - 19120:19120
      
  ##Storage
  storage:
    image: minio/minio:latest
    container_name: storage
    environment:
      - MINIO_ROOT_USER=admin
      - MINIO_ROOT_PASSWORD=password
      - MINIO_DOMAIN=storage
    networks:
      iceberg-net:
    ports:
      - 9001:9001
      - 9000:9000
    command: [ "server", "/data", "--console-address", ":9001" ]
  # Minio Client Container
  mc:
    depends_on:
      - storage
    image: minio/mc:latest
    container_name: mc
    networks:
      iceberg-net:
        aliases:
          - minio.storage
    environment:
      - AWS_ACCESS_KEY_ID=admin
      - AWS_SECRET_ACCESS_KEY=password
      - AWS_REGION=us-east-1
      - AWS_DEFAULT_REGION=us-east-1
    entrypoint: >
      /bin/sh -c " until (/usr/bin/mc config host add minio http://storage: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:
  1. Запускаем.

BASH

docker-compose up -d
  1. Проверяем, что S3-совместимое хранилище MinIO поднялось — http://localhost:9000.
Благодаря использованию MinIO клиента для инициализации у нас уже создана корзина warehouse, которая будет использоваться для хранения данных и метаданных.
  1. Следующим шагом проверяем Nessie — http://localhost:19120.
Тут мы увидим минималистичный интерфейс отображения списка данных со ссылками на метаданные в S3 в виде репозитория.
Данных пока нет, и загрузить их прямо здесь не получится, — нам нужен клиент, который будет работать с Nessie через REST API.
  1. Добавим Dremio как портал самообслуживания.

YAML

services:
  ##Self management portal
  dremio:
    platform: linux/x86_64
    image: dremio/dremio-oss:latest
    networks:
      iceberg-net:
    ports:
      - 9047:9047
      - 31010:31010
      - 32010:32010
    container_name: dremio
    environment:
      - DREMIO_JAVA_SERVER_EXTRA_OPTS=-Dpaths.dist=file:///opt/dremio/data/dist

  #Nessie catalog
  catalog:
    image: projectnessie/nessie
    container_name: catalog
    networks:
      iceberg-net:
    ports:
      - 19120:19120
      
  ##Storage
  storage:
    image: minio/minio
    container_name: storage
    environment:
      - MINIO_ROOT_USER=admin
      - MINIO_ROOT_PASSWORD=password
      - MINIO_DOMAIN=storage
    networks:
      iceberg-net:
    ports:
      - 9001:9001
      - 9000:9000
    command: [ "server", "/data", "--console-address", ":9001" ]
  # Minio Client Container
  mc:
    depends_on:
      - storage
    image: minio/mc
    container_name: mc
    networks:
      iceberg-net:
        aliases:
          - minio.storage
    environment:
      - AWS_ACCESS_KEY_ID=admin
      - AWS_SECRET_ACCESS_KEY=password
      - AWS_REGION=us-east-1
      - AWS_DEFAULT_REGION=us-east-1
    entrypoint: >
      /bin/sh -c " until (/usr/bin/mc config host add minio http://storage: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:
  1. Запускаем.

BASH

docker-compose up -d
  1. Открываем http://localhost:9047. Сначала заполняем учётные данные, например, так:
Тут тоже минималистичный интерфейс.
  1. Чтобы продолжить построение Data Lakehouse, нужно подключиться из портала к каталогу. Для этого нажмите “Add Source” и выберите в списке Nessie.
  1. Дальше нужно настроить источник http://catalog:19120/api/v2.
  1. На вкладке Storage мы вводим данные нашего S3.
  2. Для root path: warehouse
  3. Для access key: admin
  4. Для secret key: password
  1. Добавляем в connection properties:
  2. fs.s3a.path.style.access: true
  3. fs.s3a.endpoint: storage:9000
  4. dremio.s3.compat: true
Убедитесь, что напротив “Encrypt Connection” галочка не стоит. Должно получиться, как на скриншоте:
  1. Мы видим подключение и текущую ветку данных.
  1. Теперь нужно добавить данные в наш Data Lakehouse. Воспользуемся уже знакомым Airflow и добавим в docker-compose файл общий блок настроек x-airflow-common, веб-сервер, планировщик, базу и блок инициализации airflow-init.

YAML

x-airflow-common: &airflow-common
  build:
    context: ./airflow
    dockerfile: Dockerfile
  environment: &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: LocalExecutor #локальный запуск
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__FERNET_KEY: ''
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
    AIRFLOW__API__AUTH_BACKENDS: 'airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session'
    AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
    AIRFLOW__WEBSERVER__SECRET_KEY: 'your_airflow_webserver_sec_key'
    _PIP_ADDITIONAL_REQUIREMENTS: ''
    AIRFLOW_INPUT_DIR: '/opt/airflow/dag-inputs'
  volumes:
    - ./airflow/dags:/opt/airflow/dags
    - ./airflow/dags/sql:/opt/airflow/dags/sql
    - ./airflow/requirements.txt:/opt/airflow/requirements.txt
    - ./data:/opt/airflow/sample_files
  user: "${AIRFLOW_UID:-50000}:0"
  depends_on: &airflow-common-depends-on
    postgres:
      condition: service_healthy
  networks:
    iceberg-net:

services:
  
  #Airflow
  airflow-webserver:
    <<: *airflow-common

    ports:
      - "8080:8080"
    depends_on:
      postgres:
        condition: service_healthy
      airflow-init:
        condition: service_completed_successfully
    command: webserver
    networks:
      iceberg-net:
    healthcheck:
      test: [ "CMD-SHELL", "curl -f http://localhost:8080/health || exit 1" ]
      interval: 30s
      timeout: 10s
      retries: 3

  airflow-scheduler:
    <<: *airflow-common
    networks:
      iceberg-net:
    command: scheduler
    depends_on:
      postgres:
        condition: service_healthy
      airflow-webserver:
        condition: service_healthy
    healthcheck:
      test: [ "CMD-SHELL", "curl -f http://localhost:8080/health || exit 1" ]
      interval: 30s
      timeout: 10s
      retries: 3

  airflow-init:
    <<: *airflow-common
    depends_on:
      postgres:
        condition: service_healthy
    command: >
      bash -c " airflow db init && airflow users create \ --username admin \ --firstname admin \ --lastname admin \ --role Admin \ --email admin@sample.ru \ --password admin 

      "
    networks:
      iceberg-net:


  postgres:
    image: postgres:16.0
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    logging:
      options:
        max-size: 10m
        max-file: "3"
    healthcheck:
      test:
        - CMD
        - pg_isready
        - -U
        - airflow
      interval: 10s
      retries: 5
      start_period: 5s
    restart: always
    networks:
      iceberg-net:

  ##Self management portal
  dremio:
    platform: linux/x86_64
    image: dremio/dremio-oss:latest
    networks:
      iceberg-net:
    ports:
      - 9047:9047
      - 31010:31010
      - 32010:32010
    container_name: dremio
    environment:
      - DREMIO_JAVA_SERVER_EXTRA_OPTS=-Dpaths.dist=file:///opt/dremio/data/dist

  #Nessie catalog
  catalog:
    image: projectnessie/nessie
    container_name: catalog
    networks:
      iceberg-net:
    ports:
      - 19120:19120
      
  ##Storage
  storage:
    image: minio/minio
    container_name: storage
    environment:
      - MINIO_ROOT_USER=admin
      - MINIO_ROOT_PASSWORD=password
      - MINIO_DOMAIN=storage
    networks:
      iceberg-net:
    ports:
      - 9001:9001
      - 9000:9000
    command: [ "server", "/data", "--console-address", ":9001" ]
  # Minio Client Container
  mc:
    depends_on:
      - storage
    image: minio/mc
    container_name: mc
    networks:
      iceberg-net:
        aliases:
          - minio.storage
    environment:
      - AWS_ACCESS_KEY_ID=admin
      - AWS_SECRET_ACCESS_KEY=password
      - AWS_REGION=us-east-1
      - AWS_DEFAULT_REGION=us-east-1
    entrypoint: >
      /bin/sh -c " until (/usr/bin/mc config host add minio http://storage: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:
  1. В директории проекта создадим папку airflow и data. Внутри папки airflow создадим папку dags и папку sql внутри неё, файл Dockerfile с содержимым

BASH

FROM apache/airflow:2.9.1-python3.9
USER root

# Install Python dependencies
COPY requirements.txt /requirements.txt
RUN pip3 install --upgrade pip
RUN pip3 install --no-cache-dir -r /requirements.txt
и файл requirements.txt с содержимым.
psycopg2-binary
dremio-simple-query
  1. dremio-simple-query — библиотека для доступа к Dremio, загружать данные будем через портал, поскольку это позволяет исполнять обычный SQL. Есть вариант работы напрямую с Nessie, но он затрудняет разработку DAGa из-за использования специфических конструкций.
  2. Последним шагом создадим файл dag_lakehouse.py (можете назвать как удобно) в папке dags.
После всех этих действий должна получиться структура:
  1. Файл с кодом DAGa мы заполним далее. В папку data положим файл sample.csv.

CSV

id,order_number,total,discount,buyer_id

1,12345,1000,100,648821

2,123456,500,20,6488214

3,123457,700,10,6488211

4,123458,200,0,6488219

5,123459,3000,100,648801
  1. Теперь займёмся кодом DAGa.

PYTHON

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from dremio_simple_query.connect import DremioConnection
from dremio_simple_query.connect import get_token
from os import getenv
import csv

#Делаем подключение
def get_dremio_client():
    
    ## Адрес API Dremio
    login_endpoint = "http://dremio:9047/apiv2/login"

    ## Логин/пароль для Basic Auth
    payload = {
        "userName": 'admin',
        "password": 'admin123'
    }

    ## Получаем токен
    token = get_token(uri = login_endpoint, payload=payload)

    ## Grps endpoint Dremio для быстрого взаимодействия
    arrow_endpoint="grpc://dremio:32010"

    ## Подключаемся
    dremio = DremioConnection(token, arrow_endpoint)    
    return dremio     

# Создаём таблицу
def create_table():
    
    dremio = get_dremio_client()
    df = dremio.toPandas("""
        CREATE TABLE IF NOT EXISTS catalog.orders (
            id BIGINT,
            order_number BIGINT,
            total NUMERIC(18,2),
            discount NUMERIC(18,2),
            buyer_id BIGINT
        ) PARTITION BY (order_number);""")

# Функция для чтения данных и генерации SQL-запросов
def generate_insert_queries():
    CSV_FILE_PATH = 'sample_files/sample.csv'
    with open( CSV_FILE_PATH, 'r') as csvfile:
        csvreader = csv.reader(csvfile)
    
        # Генерим запросы
        insert_queries = []
        is_header = True
        for row in csvreader:
            if is_header:
                is_header = False
                continue
            insert_query = f"INSERT INTO catalog.orders (id,order_number,total,discount,buyer_id) VALUES ({row[0]}, {row[1]}, {row[2]},{row[3]},{row[4]});"
            insert_queries.append(insert_query)
        
        # Сохраняем запросы
        with open('./dags/sql/insert_queries.sql', 'w') as f:
            for query in insert_queries:
                f.write(f"{query}\n")

#Отправляем данные в Dremio->Nessie->S3                               
def insert_data():  
    dremio = get_dremio_client()
    queries = []
    with open('./dags/sql/insert_queries.sql',mode='r') as file:
        queries = [line.rstrip() for line in file]
    for query in queries:
         dremio.toPandas(query)

#Определяем DAG
with DAG(
        dag_id="dremio_query_dag",
        start_date=datetime(2024, 12, 1),
        schedule_interval="@once", 
        catchup=False
    ) as dag:
    
        #Оператор для создания таблицы
        create_table = PythonOperator(
            task_id="run_dremio_create_table",
            python_callable=create_table
        )
        
        #Оператор для генерации запросов
        generate_queries = PythonOperator(
            task_id='generate_insert_queries',
            python_callable=generate_insert_queries
        )
        
        #Оператор для вставки данных
        run_insert_queries = PythonOperator(
            task_id="run_insert_queries",
            python_callable=insert_data
        )
        
        #Последовательность выполнения
        create_table>>generate_queries>>run_insert_queries
Функции создания таблиц, генерации запросов и отправки данных переиспользованы из предыдущего урока с минимальными изменениями в части подключения к Dremio и имени таблицы catalog.orders, где catalog — название нашего подключения к Nessie в Dremio.

Полный docker-compose файл

YAML

x-airflow-common: &airflow-common
  build:
    context: ./airflow
    dockerfile: Dockerfile
  environment: &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: LocalExecutor #локальный запуск
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__FERNET_KEY: ''
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
    AIRFLOW__API__AUTH_BACKENDS: 'airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session'
    AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
    AIRFLOW__WEBSERVER__SECRET_KEY: 'your_airflow_webserver_sec_key'
    _PIP_ADDITIONAL_REQUIREMENTS: ''
    AIRFLOW_INPUT_DIR: '/opt/airflow/dag-inputs'
  volumes:
    - ./airflow/dags:/opt/airflow/dags
    - ./airflow/dags/sql:/opt/airflow/dags/sql
    - ./airflow/requirements.txt:/opt/airflow/requirements.txt
    - ./data:/opt/airflow/sample_files
  user: "${AIRFLOW_UID:-50000}:0"
  depends_on: &airflow-common-depends-on
    postgres:
      condition: service_healthy
  networks:
    iceberg-net:

services:
  
  #Airflow
  airflow-webserver:
    <<: *airflow-common

    ports:
      - "8080:8080"
    depends_on:
      postgres:
        condition: service_healthy
      airflow-init:
        condition: service_completed_successfully
    command: webserver
    networks:
      iceberg-net:
    healthcheck:
      test: [ "CMD-SHELL", "curl -f http://localhost:8080/health || exit 1" ]
      interval: 30s
      timeout: 10s
      retries: 3

  airflow-scheduler:
    <<: *airflow-common
    networks:
      iceberg-net:
    command: scheduler
    depends_on:
      postgres:
        condition: service_healthy
      airflow-webserver:
        condition: service_healthy
    healthcheck:
      test: [ "CMD-SHELL", "curl -f http://localhost:8080/health || exit 1" ]
      interval: 30s
      timeout: 10s
      retries: 3

  airflow-init:
    <<: *airflow-common
    depends_on:
      postgres:
        condition: service_healthy
    command: >
      bash -c " airflow db init && airflow users create \ --username admin \ --firstname admin \ --lastname admin \ --role Admin \ --email admin@sample.ru \ --password admin 

      "
    networks:
      iceberg-net:


  postgres:
    image: postgres:16.0
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    logging:
      options:
        max-size: 10m
        max-file: "3"
    healthcheck:
      test:
        - CMD
        - pg_isready
        - -U
        - airflow
      interval: 10s
      retries: 5
      start_period: 5s
    restart: always
    networks:
      iceberg-net:

  ##Self management portal
  dremio:
    platform: linux/x86_64
    image: dremio/dremio-oss:latest
    networks:
      iceberg-net:
    ports:
      - 9047:9047
      - 31010:31010
      - 32010:32010
    container_name: dremio
    environment:
      - DREMIO_JAVA_SERVER_EXTRA_OPTS=-Dpaths.dist=file:///opt/dremio/data/dist

  #Nessie catalog
  catalog:
    image: projectnessie/nessie
    container_name: catalog
    networks:
      iceberg-net:
    ports:
      - 19120:19120
      
  ##Storage
  storage:
    image: minio/minio
    container_name: storage
    environment:
      - MINIO_ROOT_USER=admin
      - MINIO_ROOT_PASSWORD=password
      - MINIO_DOMAIN=storage
    networks:
      iceberg-net:
    ports:
      - 9001:9001
      - 9000:9000
    command: [ "server", "/data", "--console-address", ":9001" ]
  # Minio Client Container
  mc:
    depends_on:
      - storage
    image: minio/mc
    container_name: mc
    networks:
      iceberg-net:
        aliases:
          - minio.storage
    environment:
      - AWS_ACCESS_KEY_ID=admin
      - AWS_SECRET_ACCESS_KEY=password
      - AWS_REGION=us-east-1
      - AWS_DEFAULT_REGION=us-east-1
    entrypoint: >
      /bin/sh -c " until (/usr/bin/mc config host add minio http://storage: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:
  1. Запускаем.

BASH

docker-compose up -d
  1. Заходим в Airflow и проверяем наш DAG — http://localhost:8080.
Он может появиться не сразу — тогда следует проверить состояние контейнеров, немного подождать и обновить.

BASH

docker-compose ps -a
docker-compose logs -f <id>
  1. Запускаем DAG (стрелка запуска справа). Если произошла ошибка, можно провалиться в DAG, выбрать нужный запуск, открыть Graph и, кликнув на упавший кубик, посмотреть логи (Logs справа от Graph).
Airflow предоставляет отличную возможность поправить код DAGa и не перезапускать контейнер. Если файл пробрасывается в контейнер, то изменения подтянутся автоматически и нужно будет просто запустить ещё раз.
  1. После успешного выполнения идём в Dremio, Nessie и S3, чтобы проверить данные.
На этом уровне уже лежат наши данные: один объект = одна директория. Если провалиться в папку, то мы увидим файл с расширением .parquet.
Его можно скачать и посмотреть любым просмотрщиком для Parquet.
А если зайти в папку metadata (на два уровня наверх), то мы увидим файлы с метаданными, avro файлы-манифесты и avro файлы-списки манифестов.
У нас было пять запросов на вставку, поэтому файлов метаданных шесть. Пять — для данных и один основной, и именно на него указывает каталог Nessie в ветке main.
  1. Взглянем на список манифестов, на который указывает файл метаданных.

JSON

{
  "format-version" : 2,
  "table-uuid" : "b284c55f-6361-4fe3-b92c-a09e96d93e0f",
  "location" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13",
  "last-sequence-number" : 6,
  "last-updated-ms" : 1734469350842,
  "last-column-id" : 5,
  "current-schema-id" : 0,
  /*
   Информация о схеме таблицы
  */
  "schemas" : [ {
    "type" : "struct",
    "schema-id" : 0,
    "fields" : [ {
      "id" : 1,
      "name" : "id",
      "required" : false,
      "type" : "long"
    }, {
      "id" : 2,
      "name" : "order_number",
      "required" : false,
      "type" : "long"
    }, {
      "id" : 3,
      "name" : "total",
      "required" : false,
      "type" : "decimal(18, 2)"
    }, {
      "id" : 4,
      "name" : "discount",
      "required" : false,
      "type" : "decimal(18, 2)"
    }, {
      "id" : 5,
      "name" : "buyer_id",
      "required" : false,
      "type" : "long"
    } ]
  } ],
  "default-spec-id" : 0,
  "partition-specs" : [ {
    "spec-id" : 0,
    "fields" : [ {
      "name" : "order_number",
      "transform" : "identity",
      "source-id" : 2,
      "field-id" : 1000
    } ]
  } ],
  "last-partition-id" : 1000,
  "default-sort-order-id" : 0,
  "sort-orders" : [ {
    "order-id" : 0,
    "fields" : [ ]
  } ],
  "properties" : {
    "compatibility.snapshot-id-inheritance.enabled" : "true",
    "write.metadata.delete-after-commit.enabled" : "false",
    "commit.manifest.target-size-bytes" : "153600",
    "gc.enabled" : "false",
    "write.parquet.compression-codec" : "zstd"
  },
  /*
  Указатель на актуальный снимок
  */
  "current-snapshot-id" : 3239599064039413042,
  "refs" : {
    "main" : {
      "snapshot-id" : 3239599064039413042,
      "type" : "branch"
    }
  },
  /*
    Снимки данных
  */
  "snapshots" : [ {
    "sequence-number" : 1,
    "snapshot-id" : 6476163763181576686,
    "timestamp-ms" : 1734469339148,
    "summary" : {
      "operation" : "append",
      "changed-partition-count" : "0",
      "total-records" : "0",
      "total-files-size" : "0",
      "total-data-files" : "0",
      "total-delete-files" : "0",
      "total-position-deletes" : "0",
      "total-equality-deletes" : "0",
      "iceberg-version" : "Apache Iceberg 1.6.0-229d8f6-20240725064828-ee9f85d (commit ee9f85dee581c173bf88b61ef2912f7401c277b7)"
    },
    "manifest-list" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/snap-6476163763181576686-1-8d154a07-6dd7-4f9b-b8e8-11c50ea36308.avro",
    "schema-id" : 0
  }, {
    "sequence-number" : 2,
    "snapshot-id" : 2874920170473069900,
    "parent-snapshot-id" : 6476163763181576686,
    "timestamp-ms" : 1734469347474,
    "summary" : {
      "operation" : "append",
      "dremio-job-id" : "189e1920-3b61-93db-6d31-c6ad867f7e00",
      "added-data-files" : "1",
      "added-records" : "1",
      "total-records" : "1",
      "total-files-size" : "0",
      "total-data-files" : "1",
      "total-delete-files" : "0",
      "total-position-deletes" : "0",
      "total-equality-deletes" : "0",
      "iceberg-version" : "Apache Iceberg 1.6.0-229d8f6-20240725064828-ee9f85d (commit ee9f85dee581c173bf88b61ef2912f7401c277b7)"
    },
    "manifest-list" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/snap-2874920170473069900-1-35610e20-c8e6-4f8a-a3c8-ea0a59ee2877.avro",
    "schema-id" : 0
  }, {
    "sequence-number" : 3,
    "snapshot-id" : 1166829870316273135,
    "parent-snapshot-id" : 2874920170473069900,
    "timestamp-ms" : 1734469348301,
    "summary" : {
      "operation" : "append",
      "dremio-job-id" : "189e191b-e0cb-8f52-a826-b53198d3cb00",
      "added-data-files" : "1",
      "added-records" : "1",
      "total-records" : "2",
      "total-files-size" : "0",
      "total-data-files" : "2",
      "total-delete-files" : "0",
      "total-position-deletes" : "0",
      "total-equality-deletes" : "0",
      "iceberg-version" : "Apache Iceberg 1.6.0-229d8f6-20240725064828-ee9f85d (commit ee9f85dee581c173bf88b61ef2912f7401c277b7)"
    },
    "manifest-list" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/snap-1166829870316273135-1-bfececa8-96b2-4a97-9854-1320d185ec18.avro",
    "schema-id" : 0
  }, {
    "sequence-number" : 4,
    "snapshot-id" : 2656286237141665234,
    "parent-snapshot-id" : 1166829870316273135,
    "timestamp-ms" : 1734469349243,
    "summary" : {
      "operation" : "append",
      "dremio-job-id" : "189e191a-992e-5be3-fa5e-c239297d4000",
      "added-data-files" : "1",
      "added-records" : "1",
      "total-records" : "3",
      "total-files-size" : "0",
      "total-data-files" : "3",
      "total-delete-files" : "0",
      "total-position-deletes" : "0",
      "total-equality-deletes" : "0",
      "iceberg-version" : "Apache Iceberg 1.6.0-229d8f6-20240725064828-ee9f85d (commit ee9f85dee581c173bf88b61ef2912f7401c277b7)"
    },
    "manifest-list" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/snap-2656286237141665234-1-e31b7226-8a0a-4ce6-a347-0cac17a28199.avro",
    "schema-id" : 0
  }, {
    "sequence-number" : 5,
    "snapshot-id" : 6622575447602035384,
    "parent-snapshot-id" : 2656286237141665234,
    "timestamp-ms" : 1734469350018,
    "summary" : {
      "operation" : "append",
      "dremio-job-id" : "189e191a-3172-0fef-cbd2-8502af25be00",
      "added-data-files" : "1",
      "added-records" : "1",
      "total-records" : "4",
      "total-files-size" : "0",
      "total-data-files" : "4",
      "total-delete-files" : "0",
      "total-position-deletes" : "0",
      "total-equality-deletes" : "0",
      "iceberg-version" : "Apache Iceberg 1.6.0-229d8f6-20240725064828-ee9f85d (commit ee9f85dee581c173bf88b61ef2912f7401c277b7)"
    },
    "manifest-list" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/snap-6622575447602035384-1-eba4ece4-80b6-41bb-8785-cf3c077ffe77.avro",
    "schema-id" : 0
  }, 
  /*
  Актуальный снимок
  */
  {
    "sequence-number" : 6,
    "snapshot-id" : 3239599064039413042,
    "parent-snapshot-id" : 6622575447602035384,
    "timestamp-ms" : 1734469350842,
    /*
        Информация о происхождении
    */
    "summary" : {
      "operation" : "append",
      "dremio-job-id" : "189e1918-ae7e-aaf7-b6de-4a0a0a001f00",
      "added-data-files" : "1",
      "added-records" : "1",
      "total-records" : "5",
      "total-files-size" : "0",
      "total-data-files" : "5",
      "total-delete-files" : "0",
      "total-position-deletes" : "0",
      "total-equality-deletes" : "0",
      "iceberg-version" : "Apache Iceberg 1.6.0-229d8f6-20240725064828-ee9f85d (commit ee9f85dee581c173bf88b61ef2912f7401c277b7)"
    },
    //указатель на список манифестов
    "manifest-list" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/snap-3239599064039413042-1-af8ef938-336e-4efd-9f8d-960fa4e7e98b.avro",
    "schema-id" : 0
  } ],
  "statistics" : [ ],
  "partition-statistics" : [ ],
  "snapshot-log" : [ {
    "timestamp-ms" : 1734469339148,
    "snapshot-id" : 6476163763181576686
  }, {
    "timestamp-ms" : 1734469347474,
    "snapshot-id" : 2874920170473069900
  }, {
    "timestamp-ms" : 1734469348301,
    "snapshot-id" : 1166829870316273135
  }, {
    "timestamp-ms" : 1734469349243,
    "snapshot-id" : 2656286237141665234
  }, {
    "timestamp-ms" : 1734469350018,
    "snapshot-id" : 6622575447602035384
  }, {
    "timestamp-ms" : 1734469350842,
    "snapshot-id" : 3239599064039413042
  } ],
  /*
    Список метаданных данных
  */
  "metadata-log" : [ {
    "timestamp-ms" : 1734469339148,
    "metadata-file" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/00000-f36fbb8d-9e59-4f8a-8860-4b392c57bea9.metadata.json"
  }, {
    "timestamp-ms" : 1734469347474,
    "metadata-file" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/00001-565f36e8-6472-4ca0-848e-f210c743c1e0.metadata.json"
  }, {
    "timestamp-ms" : 1734469348301,
    "metadata-file" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/00002-2118f72d-86ed-47f5-89fc-3a8a49596d60.metadata.json"
  }, {
    "timestamp-ms" : 1734469349243,
    "metadata-file" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/00003-0bfdf9a6-9078-471e-814b-fb7109a5117d.metadata.json"
  }, {
    "timestamp-ms" : 1734469350018,
    "metadata-file" : "s3://warehouse/orders_9aed8624-4fd6-471a-908e-adc4d8e2ac13/metadata/00004-3d5d55ba-0781-4238-81d4-c1a35845b767.metadata.json"
  } ]
}
В нём есть как раз названия конкретных файлов-манифестов для этой версии.
  1. Если открыть один из них, внутри, помимо служебной информации, будет содержаться ссылка на файл в хранилище.
Пайплайн полностью готов! Осталось посмотреть на интересную возможность каталога Nessie — работу с данными как с кодом в отдельной ветке репозитория.
  1. Для этого открываем портал http://localhost:9047 и выбираем SQL Runner.
  1. В открывшемся окне выбираем Context (верхний правый угол картинки рядом со знаком функции), чтобы в скриптах не вводить постоянно префикс catalog.
  1. Теперь создаём ветку от данных.

SQL

CREATE BRANCH ingest in catalog;

USE BRANCH ingest;
UPDATE orders SET total=-1 WHERE id=1;
SELECT * from orders;
Если переключиться на основную ветку main, мы увидим, что данные не изменились: total равен 1000.

SQL

USE BRANCH main;
SELECT * FROM orders;
  1. Чтобы влить ветку в основную, выполним команду и запросим данные из main.

SQL

MERGE BRANCH ingest INTO main IN catalog;
USE BRANCH main;
SELECT * FROM orders;
Теперь данные из ветки стали основными.
Возможность работать с данными в ветках даёт полную свободу в рамках «данные как код». С ними можно экспериментировать и не бояться сломать, ведь в случае порчи данных можно вернуться к снимку.
Data Lakehouse и реализация спецификации Apache Iceberg помогают построить отказоустойчивую архитектуру данных, но всё-таки это своего рода монолит.
Помимо сложности настройки, он имеет присущие монолиту в архитектуре данных особенности в виде разделения ответственности между продуктовой командой и командой поддержки решения, что порождает снижение time to market из-за исправления багов и проблем с инфраструктурой.
Чтобы закрепить новый опыт, выполните несколько заданий. И подведём итоги урока.

Задание 1

Компания разрабатывает Data Lakehouse и хочет использовать MinIO в качестве S3-совместимого хранилища. Однако при запуске контейнера MinIO не создаётся корзина warehouse, несмотря на то, что конфигурация docker-compose.yml была скопирована из документации. В чём может быть причина?

Задание 2

При настройке Dremio для работы с Nessie в качестве каталога и MinIO в качестве хранилища в Data Lakehouse были допущены ошибки. Какие из перечисленных проблем могут привести к невозможности работы с хранилищем?

Задание 3

В ходе работы с Data Lakehouse через Nessie возникла необходимость создать ветку для тестирования данных перед внесением изменений в основную ветку.
Какая последовательность SQL-команд позволит создать новую ветку, обновить данные и проверить, что изменения не затронули основную ветку?

Задание 4

Компания "BigData Insights" разрабатывает пайплайн для Data Lakehouse, используя такие инструменты, как MinIO для хранения, Nessie для управления версиями данных и Apache Iceberg для контроля таблиц и метаданных. Ваша задача — помочь команде инженеров построить и проверить пайплайн данных для эффективной аналитики.
Инженеры настраивают пайплайн для передачи и хранения данных. Какой порядок действий правильный?
Подключение Nessie как каталога метаданных.
Шаг 2
Запуск MinIO для S3-совместимого хранилища.
Шаг 1
Настройка Airflow для передачи данных в хранилище.
Шаг 3

Итоги

  • Вы поработали с MinIO, Nessie, Dremio и Airflow и научились их интегрировать. Все эти компоненты помогают удобно хранить данные, управлять их версиями, быстро анализировать и автоматизировать процессы. Это нужно для построения надёжных систем работы с большими данными в компаниях.
  • Вы построили пайплайн, включая настройку хранилища данных и управление версиями. А также научились использовать Apache Iceberg для управления метаданными и построения отказоустойчивой архитектуры.
  • Узнали, как работать с ветками в Nessie для безопасного тестирования данных.
В следующем уроке речь пойдёт о распределённых архитектурах данных.