データレイクはじめに

ClickHouse でオープンテーブルフォーマットのデータをクエリし、高速化し、書き戻すための実践的な入門ガイドです。

要点

データレイクのテーブルをクエリし、MergeTree で高速化して、その結果を Iceberg に書き戻すまでを実践的に紹介します。すべての手順で公開データセットを使用し、Cloud と OSS の両方で動作します。

このガイドのスクリーンショットは ClickHouse Cloud の SQL コンソールで取得したものです。すべてのクエリは Cloud とセルフマネージド環境の両方で動作します。

ClickHouse では、オープンテーブルフォーマットを読み取る方法として、テーブル関数、テーブルエンジン、DataLakeCatalog データベースエンジンの 3 つがあります。テーブルがデータカタログ (Glue、Unity Catalog、REST など) にある場合は、DataLakeCatalog を使って接続してください。これにより、1 つの関数ですべての Iceberg/Delta テーブルにアクセスできます。以下のテーブル関数およびテーブルエンジンのセクションは、アドホッククエリに最適であり、特定のストレージパスが分かっていてカタログを使わない場合にも適しています。

1. Iceberg データを直接クエリする

   最も手早く始めるには、特にアドホッククエリを実行する場合やカタログを使わない場合は、icebergS3() テーブル関数を使います。S3 上の Icebergテーブルを指定すれば、セットアップ不要ですぐにクエリできます。

   スキーマを確認します:

   DESCRIBE icebergS3('https://datasets-documentation.s3.amazonaws.com/lake_formats/iceberg/')
   

   クエリを実行する:

   SELECT
       url,
       count() AS cnt
   FROM icebergS3('https://datasets-documentation.s3.amazonaws.com/lake_formats/iceberg/')
   GROUP BY url
   ORDER BY cnt DESC
   LIMIT 5
   

   ClickHouse は S3 から Iceberg のメタデータを直接読み取り、スキーマを自動的に推論します。同じアプローチは deltaLake()、hudi()、paimon() でも使用できます。

   詳細はこちら: オープンテーブルフォーマットの直接クエリ では、4 つのフォーマットすべてに加え、分散読み取り向けのクラスター バリアントやストレージバックエンドのオプション (S3、Azure、HDFS、local) について説明しています。

2. 永続的なテーブルエンジンを作成する

   カタログを使わず、同じ path に繰り返しクエリする場合は、Iceberg テーブルエンジンを使ってテーブルを作成すると、毎回 path を指定する必要がありません。データは S3 に保持されたままで、複製されません:

   CREATE TABLE hits_iceberg
       ENGINE = IcebergS3('https://datasets-documentation.s3.amazonaws.com/lake_formats/iceberg/')
   

   では、通常のClickHouseテーブルと同じようにクエリできます:

   SELECT
       url,
       count() AS cnt
   FROM hits_iceberg
   GROUP BY url
   ORDER BY cnt DESC
   LIMIT 5
   

   このテーブルエンジンでは、データキャッシュ、メタデータキャッシュ、スキーマ進化、タイムトラベルをサポートしています。テーブルエンジンの機能の詳細については直接クエリガイドを、機能の完全な比較についてはサポートマトリクスを参照してください。

3. カタログに接続する

   組織でデータカタログを使用している場合は、このインテグレーション パスを推奨します。カタログはテーブルメタデータと検出を一元化できるため、ストレージ path ごとに table definition を管理する代わりに、DataLakeCatalog データベースエンジンで一度接続するだけで済みます。接続を作成した後にアップストリームで追加されたテーブルも含め、カタログ内のすべてのテーブルが ClickHouse テーブルとして表示されます。

   カタログを使用している場合に推奨

   Glue、Unity Catalog、REST、およびその他のサポートされているカタログで本番 workloads を運用する場合は、DataLakeCatalog を使用してください。table function と table engine は特定の path がわかっている場合には使えますが、カタログの拡大に合わせて同期された状態は維持されず、テーブルごとに別個の credentials または path が必要になります。

   以下は、AWS Glue に接続する例です。

   CREATE DATABASE my_lake
   ENGINE = DataLakeCatalog
   SETTINGS
       catalog_type = 'glue',
       region = '<your-region>',
       aws_access_key_id = '<your-access-key>',
       aws_secret_access_key = '<your-secret-key>'
   

   カタログの種類ごとに、それぞれ専用の接続設定が必要です。サポートされているカタログの一覧と各カタログの設定オプションについては、カタログ ガイドを参照してください。

   テーブルを参照してクエリを実行する:

   SHOW TABLES FROM my_lake;
   

   SELECT count(*) FROM my_lake.`<database>.<table>`
   

   ClickHouse はネイティブで複数のネームスペースをサポートしていないため、<database>.<table> はバッククォートで囲む必要があります。

   詳細: データカタログへの接続 では、Delta と Iceberg の例を交えながら、Unity Catalog のセットアップ全体を順を追って説明しています。

4. クエリを実行する

   上記で使用した方法が table function、table engine、または DataLakeCatalog のいずれであっても、同じ ClickHouse SQL をそのまま使用できます。カタログを使う本番環境では、DataLakeCatalog データベース経由でクエリしてください。ほかの例も、手早いテストや path ベースのアクセスには引き続き役立ちます:

   -- Table function
   SELECT url, count() AS cnt
   FROM icebergS3('https://datasets-documentation.s3.amazonaws.com/lake_formats/iceberg/')
   GROUP BY url ORDER BY cnt DESC LIMIT 5
   
   -- Table engine
   SELECT url, count() AS cnt
   FROM hits_iceberg
   GROUP BY url ORDER BY cnt DESC LIMIT 5
   
   -- Catalog
   SELECT url, count() AS cnt
   FROM my_lake.`<database>.<table>`
   GROUP BY url ORDER BY cnt DESC LIMIT 5
   

   クエリの構文はまったく同じで、変わるのは FROM 句だけです。データソースが変わっても、ClickHouse SQL のすべての関数、JOIN、集計は同じように使用できます。

5. ClickHouse にデータの一部を読み込む

   Iceberg を直接クエリするのは便利ですが、パフォーマンスはネットワークスループットとファイルレイアウトに左右されます。分析用途では、データをネイティブな MergeTree テーブルに読み込んでください。

   まず、Iceberg テーブルに対してフィルタしたクエリを実行し、ベースラインを確認します。

   SELECT
       url,
       count() AS cnt
   FROM hits_iceberg
   WHERE counterid = 38
   GROUP BY url
   ORDER BY cnt DESC
   LIMIT 5
   

   このクエリでは、Iceberg は counterid フィルターを認識しないため、S3 内のデータセット全体がスキャンされます。数秒かかることがあります。

   次に、MergeTree テーブルを作成してデータを読み込みます。

   CREATE TABLE hits_clickhouse
   (
       url String,
       eventtime DateTime,
       counterid UInt32
   )
   ENGINE = MergeTree()
   ORDER BY (counterid, eventtime);
   

   INSERT INTO hits_clickhouse
   SELECT url, eventtime, counterid
   FROM hits_iceberg
   

   同じクエリをMergeTreeテーブルに対してもう一度実行します。

   SELECT
       url,
       count() AS cnt
   FROM hits_clickhouse
   WHERE counterid = 38
   GROUP BY url
   ORDER BY cnt DESC
   LIMIT 5
   

   counterid は ORDER BY キーの先頭カラムであるため、ClickHouse のスパースプライマリインデックスは該当するグラニュールまで直接絞り込み、1 億行すべてをスキャンする代わりに counterid = 38 の行だけを読み取ります。その結果、処理速度が大幅に向上します。

   分析の高速化 ガイドでは、LowCardinality 型、全文索引、最適化された順序キーをさらに活用し、2 億 8300 万行のデータセットで 約 40 倍の高速化 を実証しています。

   詳細はこちら: MergeTree による分析の高速化 では、スキーマ最適化、全文索引、そしてパフォーマンスの比較を導入前後で包括的に解説しています。

6. Iceberg への書き戻し

   ClickHouse は Iceberg テーブルにデータを書き戻すこともできるため、リバース ETL ワークフローを実現できます。これにより、集計結果やデータの一部を公開し、他のツール (Spark、Trino、DuckDB など) で利用できるようになります。

   出力先となる Iceberg テーブルを作成します。

   CREATE TABLE output_iceberg
   (
       url String,
       cnt UInt64
   )
   ENGINE = IcebergS3('https://your-bucket.s3.amazonaws.com/output/', 'access_key', 'secret_key')
   

   集計結果を書き込む:

   SET allow_experimental_insert_into_iceberg = 1;
   
   INSERT INTO output_iceberg
   SELECT
       url,
       count() AS cnt
   FROM hits_clickhouse
   GROUP BY url
   ORDER BY cnt DESC
   

   生成されたIcebergテーブルは、Iceberg互換の任意のエンジンで読み取ることができます。

   詳細はこちら: オープンテーブルフォーマットへのデータの書き込み では、UK Price Paidデータセットを使用して生データと集計結果を書き込む方法を説明しており、ClickHouseの型をIcebergにマッピングする際のスキーマに関する考慮事項も取り上げています。

​

次のステップ

ここまででワークフロー全体を確認できたので、各領域をさらに詳しく見ていきましょう。

• カタログへの接続 — カタログを利用するワークロードに推奨。Delta と Iceberg を含む Unity Catalog の完全ガイド
• 直接クエリする — 4 つのフォーマット、クラスター構成のバリエーション、テーブルエンジン、キャッシュ
• 分析の高速化 — スキーマの最適化、索引付け、約 40 倍高速化のデモ
• データレイクへの書き込み — 生データの書き込み、集計データの書き込み、型マッピング
• サポートマトリクス — フォーマットとストレージバックエンドごとの機能比較
• ベストプラクティス — アクセス方法の選択、パフォーマンス設定、ワークロードパターン