Каталог SeaweedFS

ClickHouse поддерживает интеграцию с несколькими каталогами (Unity, Glue, REST, Polaris и т. д.). В этом руководстве описано, как запрашивать данные с помощью ClickHouse и каталога SeaweedFS.

SeaweedFS — это распределённое хранилище файлов и объектов с открытым исходным кодом и S3-совместимым шлюзом. Его S3-бакеты таблиц включают оба компонента развертывания Iceberg: встроенный REST-каталог Iceberg предоставляет метаданные таблиц, а бакет таблиц хранит данные таблиц в виде файлов Parquet по той же конечной точке S3:

• Единый сервис — метаданные каталога и данные Parquet обслуживаются одним процессом без отдельной базы данных метаданных
• Соответствие REST API спецификации REST-каталога Iceberg
• Обслуживание на стороне сервера — автоматическая компактация файлов Parquet и удаление устаревших снимков без внешнего сервиса обслуживания

Для локальной разработки и тестирования SeaweedFS и ClickHouse можно запустить с помощью Docker Compose. Этот подход идеально подходит для обучения, прототипирования и разработки.

1. Docker и Docker Compose: убедитесь, что Docker установлен и запущен.
2. Версии: SeaweedFS 4.42 или новее; ClickHouse 26.8 или новее (версии начиная с 25.8 поддерживают чтение и вставку, но для создания таблиц через каталог требуется версия 26.8).
3. Python с PyIceberg (необязательно): используется ниже для заполнения демонстрационных данных.

Шаг 1: Создайте папку для запуска примера, затем создайте файл s3config.json с учётными данными для шлюза S3 и каталога:

{
  "identities": [
    {
      "name": "analyst",
      "credentials": [
        {
          "accessKey": "tutorialkey",
          "secretKey": "tutorialsecret"
        }
      ],
      "actions": ["Admin", "Read", "Write", "List", "Tagging"]
    }
  ]
}

Шаг 2: Создайте файл docker-compose.yml со следующей конфигурацией:

services:
  seaweedfs:
    image: chrislusf/seaweedfs:latest
    command: mini -dir=/data -s3.config=/etc/seaweedfs/s3config.json -tableBucket=analytics -admin.port=12646
    ports:
      - "8333:8333"   # S3 endpoint
      - "8181:8181"   # Iceberg REST catalog
    volumes:
      - ./s3config.json:/etc/seaweedfs/s3config.json
      - seaweedfs_data:/data
    networks:
      - iceberg_net

  clickhouse:
    image: clickhouse/clickhouse-server:latest
    container_name: seaweedfs-clickhouse
    ports:
      - "8123:8123"
      - "9000:9000"
    depends_on:
      - seaweedfs
    networks:
      - iceberg_net

volumes:
  seaweedfs_data:

networks:
  iceberg_net:
    driver: bridge

Команда mini запускает весь стек SeaweedFS в одном контейнере. Флаг -tableBucket=analytics заранее создаёт бакет S3 Tables с именем analytics, который используется в качестве хранилища Iceberg. -admin.port=12646 обеспечивает, что административный порт gRPC, производный от этого значения в SeaweedFS, находится ниже диапазона эфемерных портов Linux, иначе при запуске его может раньше занять другое соединение.

Шаг 3: Выполните следующую команду, чтобы запустить сервисы:

docker compose up -d

Каталог изначально пуст. Создайте таблицу и добавьте в неё несколько строк с помощью PyIceberg (pip install pyiceberg pyarrow):

import pyarrow as pa

from pyiceberg.catalog.rest import RestCatalog

catalog = RestCatalog(
    "seaweedfs",
    uri="http://localhost:8181",
    warehouse="s3://analytics",
    credential="tutorialkey:tutorialsecret",
    **{
        "s3.endpoint": "http://localhost:8333",
        "s3.access-key-id": "tutorialkey",
        "s3.secret-access-key": "tutorialsecret",
        "s3.region": "us-east-1",
        "s3.path-style-access": "true",
    },
)

rows = pa.table({
    "id": pa.array([1, 2, 3, 4, 5, 6], pa.int64()),
    "region": ["NA", "EU", "EU", "APAC", "NA", "EU"],
    "amount": pa.array([12.5, 40.0, 7.25, 99.9, 3.5, 61.0], pa.float64()),
})

catalog.create_namespace("sales")
table = catalog.create_table("sales.orders", schema=rows.schema)
table.append(rows)

Подключитесь к контейнеру ClickHouse:

docker exec -it seaweedfs-clickhouse clickhouse-client

Затем создайте подключение к каталогу SeaweedFS:

SET allow_experimental_database_iceberg = 1;

CREATE DATABASE lake
ENGINE = DataLakeCatalog('http://seaweedfs:8181/v1', 'tutorialkey', 'tutorialsecret')
SETTINGS catalog_type = 'rest',
    warehouse = 's3://analytics',
    storage_endpoint = 'http://seaweedfs:8333/analytics',
    catalog_credential = 'tutorialkey:tutorialsecret',
    oauth_server_uri = 'http://seaweedfs:8181/v1/oauth/tokens'

Аргументы движка содержат учетные данные S3, которые ClickHouse использует для чтения данных таблицы, а catalog_credential и oauth_server_uri используются для аутентификации в самом каталоге по потоку OAuth2 Client Credentials. SeaweedFS принимает один и тот же ключ доступа и секретный ключ в обоих случаях.

Теперь, когда соединение установлено, можно выполнять запросы к каталогу SeaweedFS. Например:

USE lake;

SHOW TABLES;

┌─name─────────┐
│ sales.orders │
└──────────────┘

Чтобы выполнить запрос к таблице:

SELECT region, sum(amount) AS total
FROM `sales.orders`
GROUP BY region
ORDER BY total DESC;

┌─region─┬──total─┐
│ EU     │ 108.25 │
│ APAC   │   99.9 │
│ NA     │     16 │
└────────┴────────┘

Вы также можете создавать таблицы в каталоге SeaweedFS и записывать в них данные непосредственно из ClickHouse:

SET allow_experimental_database_iceberg = 1;
SET allow_experimental_insert_into_iceberg = 1;
SET write_full_path_in_iceberg_metadata = 1;

CREATE TABLE lake.`sales.returns` (id Int64, reason String)
ENGINE = IcebergS3('http://seaweedfs:8333/analytics/sales/returns/', 'tutorialkey', 'tutorialsecret');

INSERT INTO lake.`sales.returns` VALUES (1, 'damaged'), (2, 'wrong size');

SELECT * FROM lake.`sales.returns` ORDER BY id;

┌─id─┬─reason─────┐
│  1 │ damaged    │
│  2 │ wrong size │
└────┴────────────┘

Предложение движка IcebergS3 задает путь в хранилище для новой таблицы, а write_full_path_in_iceberg_metadata указывает ClickHouse зарегистрировать в каталоге полный путь к таблице.

При коммите вставки ClickHouse каталог SeaweedFS исправляет метаданные, которые экспериментальный writer пока не создает: заполняет отсутствующие ID полей в манифестах, преобразует пути к файлам относительно бакета в абсолютные пути и задает для таблицы сопоставление имен по умолчанию. После этого строгие reader, такие как PyIceberg и Spark, могут читать строки, записанные ClickHouse. Для этого требуется SeaweedFS версии 4.42 или выше.

Чтобы загрузить данные из каталога SeaweedFS в ClickHouse, сначала создайте локальную таблицу ClickHouse:

CREATE TABLE default.orders
(
    `id` Int64,
    `region` String,
    `amount` Float64
)
ENGINE = MergeTree()
ORDER BY (region, id);

Затем загрузите данные из таблицы каталога SeaweedFS с помощью INSERT INTO SELECT:

INSERT INTO default.orders
SELECT * FROM lake.`sales.orders`;