オープンテーブルフォーマットへのデータ書き込み

ClickHouse からオブジェクトストレージ内の Iceberg テーブルにデータを書き戻し、長期保存と下流での活用に利用します。

前のガイドでは、オープンテーブルフォーマットをそのままクエリし、高速分析のためにデータを MergeTree にロードしました。多くのアーキテクチャでは、データはその逆方向、つまり ClickHouse からオープンテーブルフォーマットへ戻す形でも流す必要があります。これには、主に次の 2 つのシナリオがあります。

• 長期ストレージへのオフロード - データはリアルタイム分析レイヤーとして ClickHouse に取り込まれ、ダッシュボードや運用レポートを支えます。データがリアルタイムの対象期間を過ぎたら、相互運用可能なフォーマットで、耐久性が高くコスト効率にも優れた長期保存を実現するために、オブジェクトストレージ内の Iceberg に書き出せます。
• Reverse ETL - ClickHouse 内で実行した変換、集計、エンリッチメントによって、下流のツールや他チームが利用する派生データセットが生成されます。これらの結果を Iceberg テーブルに書き込むことで、より広いデータエコシステム全体で利用できるようになります。

どちらの場合でも、INSERT INTO SELECT を使うことで、ClickHouse テーブルからオブジェクトストレージに保存された Iceberg テーブルへデータを移動できます。

オープンテーブルフォーマットへの書き込みは、現在 Iceberg テーブルのみ をサポートしています。Delta Lake テーブルの部分的なサポートは現在開発中です。テーブルをカタログで管理していてはなりません。

​

ソースデータセットを準備する

このガイドでは、UK Price Paid データセット (イングランドとウェールズにおけるすべての住宅不動産取引を記録した公開データ) を使用します。

​

MergeTreeテーブルを作成してデータを投入する

CREATE DATABASE uk;

CREATE TABLE uk.uk_price_paid
(
    price UInt32,
    date Date,
    postcode1 LowCardinality(String),
    postcode2 LowCardinality(String),
    type Enum8('terraced' = 1, 'semi-detached' = 2, 'detached' = 3, 'flat' = 4, 'other' = 0),
    is_new UInt8,
    duration Enum8('freehold' = 1, 'leasehold' = 2, 'unknown' = 0),
    addr1 String,
    addr2 String,
    street LowCardinality(String),
    locality LowCardinality(String),
    town LowCardinality(String),
    district LowCardinality(String),
    county LowCardinality(String)
)
ENGINE = MergeTree
ORDER BY (postcode1, postcode2, addr1, addr2);

公開CSVソースからテーブルに直接データを取り込みます：

INSERT INTO uk.uk_price_paid
SELECT
    toUInt32(price_string) AS price,
    parseDateTimeBestEffortUS(time) AS date,
    splitByChar(' ', postcode)[1] AS postcode1,
    splitByChar(' ', postcode)[2] AS postcode2,
    transform(a, ['T', 'S', 'D', 'F', 'O'], ['terraced', 'semi-detached', 'detached', 'flat', 'other']) AS type,
    b = 'Y' AS is_new,
    transform(c, ['F', 'L', 'U'], ['freehold', 'leasehold', 'unknown']) AS duration,
    addr1,
    addr2,
    street,
    locality,
    town,
    district,
    county
FROM url(
    'http://prod1.publicdata.landregistry.gov.uk.s3-website-eu-west-1.amazonaws.com/pp-complete.csv',
    'CSV',
    'uuid_string String,
    price_string String,
    time String,
    postcode String,
    a String,
    b String,
    c String,
    addr1 String,
    addr2 String,
    street String,
    locality String,
    town String,
    district String,
    county String,
    d String,
    e String'
) SETTINGS max_http_get_redirects=10;

30906560 rows in set. Elapsed: 59.852 sec. Processed 30.91 million rows, 5.41 GB (516.39 thousand rows/s., 90.40 MB/s.)
Peak memory usage: 485.15 MiB.

​

Icebergテーブルにデータを書き込む

​

Iceberg テーブルを作成する

Iceberg にデータを書き込むには、IcebergS3 テーブルエンジンを使用してテーブルを作成します。

なお、スキーマは MergeTree の元テーブルに比べて簡略化する必要があります。ClickHouse は Iceberg や基盤となる Parquet ファイルよりも豊富な型システムをサポートしているため、Enum、LowCardinality、UInt8 などの型は Iceberg ではサポートされておらず、互換性のある型にマッピングする必要があります。

CREATE TABLE uk.uk_iceberg
(
    price UInt32,
    date Date,
    postcode1 String,
    postcode2 String,
    type UInt32,
    is_new UInt32,
    duration UInt32,
    addr1 String,
    addr2 String,
    street String,
    locality String,
    town String,
    district String,
    county String
)
ENGINE = IcebergS3('https://datasets-documentation.s3.amazonaws.com/lake_formats/iceberg_uk_price_paid/', '<aws_access_key>', '<aws_secret_key>', '<session_token>')

​

データの一部を挿入する

INSERT INTO SELECT を使用して、MergeTree テーブルから Iceberg テーブルへデータを書き込みます。この例では、ロンドンの取引だけを書き込みます。

SET allow_experimental_insert_into_iceberg = 1;

INSERT INTO uk.uk_iceberg SELECT *
FROM uk.uk_price_paid
WHERE town = 'LONDON'

2346741 rows in set. Elapsed: 1.419 sec. Processed 30.91 million rows, 153.43 MB (21.78 million rows/s., 108.15 MB/s.)
Peak memory usage: 371.60 MiB.

​

Icebergテーブルをクエリする

データは現在、オブジェクトストレージに Iceberg 形式で保存されており、ClickHouse や Iceberg を読み取るその他のツールからクエリできます。

SELECT
    locality,
    count()
FROM uk.uk_iceberg
WHERE locality != ''
GROUP BY locality
ORDER BY count() DESC
LIMIT 10

┌─locality────┬─count()─┐
│ LONDON      │  896796 │
│ WALTHAMSTOW │    8610 │
│ LEYTON      │    3525 │
│ CHINGFORD   │    3133 │
│ HORNSEY     │    2794 │
│ STREATHAM   │    2760 │
│ WOOD GREEN  │    2443 │
│ ACTON       │    2155 │
│ LEYTONSTONE │    2102 │
│ EAST HAM    │    2085 │
└─────────────┴─────────┘

10 rows in set. Elapsed: 0.329 sec. Processed 457.86 thousand rows, 2.62 MB (1.39 million rows/s., 7.95 MB/s.)
Peak memory usage: 12.19 MiB.

​

集計結果を書き込む

Iceberg テーブルは、生データの行を保存するだけに限りません。集計や変換の出力、つまり ClickHouse 内で実行される ETL プロセスの結果も保持できます。これは、事前計算したサマリーを レイクハウス に公開し、下流のシステムで利用できるようにする際に役立ちます。

​

集計用のIcebergテーブルを作成する

CREATE TABLE uk.uk_avg_town
(
    price Float64,
    town String
)
ENGINE = IcebergS3('https://datasets-documentation.s3.amazonaws.com/lake_formats/iceberg_uk_avg_town/', '<aws_access_key>', '<aws_secret_key>', '<session_token>')

​

集計データを挿入

町ごとの平均不動産価格を算出し、その結果をIcebergに直接書き込みます：

INSERT INTO uk.uk_avg_town SELECT
    avg(price) AS price,
    town
FROM uk.uk_price_paid
GROUP BY town

1173 rows in set. Elapsed: 0.480 sec. Processed 30.91 million rows, 185.44 MB (64.34 million rows/s., 386.05 MB/s.)
Peak memory usage: 4.18 MiB.

​

集計テーブルにクエリを実行する

これで、他のツールや別のClickHouseインスタンスから、この事前計算済みのデータセットを参照できるようになります。

SELECT
    town,
    price
FROM uk.uk_avg_town
ORDER BY price DESC
LIMIT 10

┌─town───────────────┬──────────────price─┐
│ GATWICK            │ 28232811.583333332 │
│ THORNHILL          │             985000 │
│ VIRGINIA WATER     │  984633.2938574939 │
│ CHALFONT ST GILES  │  863347.7280187573 │
│ COBHAM             │    775251.47313278 │
│ PURFLEET-ON-THAMES │           772651.8 │
│ BEACONSFIELD       │  746052.9327405858 │
│ ESHER              │  686708.4969745865 │
│ KESTON             │  654541.1774842045 │
│ GERRARDS CROSS     │  639109.4084023251 │
└────────────────────┴────────────────────┘

10 rows in set. Elapsed: 0.210 sec.