ClickHouse と Databricks の統合

ClickHouse を Databricks と統合

ClickHouse Spark コネクタは Databricks とシームレスに連携します。このガイドでは、Databricks 向けのプラットフォーム固有のセットアップ、インストール、利用パターンについて説明します。

​

Databricks での API の選択

デフォルトでは、Databricks は Unity Catalog を使用しており、Spark のカタログ登録はブロックされます。この場合は、TableProvider API (フォーマットベースのアクセス) を必ず使用する必要があります。

一方、No isolation shared アクセスモードのクラスターを作成して Unity Catalog を無効にすると、代わりに Catalog API を使用できます。Catalog API では、設定を一元管理でき、Spark SQL とネイティブに統合できます。

Unity Catalog の状態	推奨 API	注記
有効 (デフォルト)	TableProvider API (フォーマットベース)	Unity Catalog により Spark のカタログ登録はブロックされます
無効 (No isolation shared)	Catalog API	”No isolation shared” アクセスモードのクラスターが必要です

​

Databricks へのインストール

​

オプション 1: Databricks UI から JAR をアップロードする

1. ランタイム JAR をビルドするか、ダウンロードします。

   clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}-{{ stable_version }}.jar
   

2. JAR を Databricks のワークスペースにアップロードします。

   • Workspace に移動し、目的のフォルダーを開きます
   • Upload をクリックし、JAR ファイルを選択します
   • JAR はワークスペースに保存されます

3. クラスターにライブラリをインストールします。

   • Compute に移動し、対象のクラスターを選択します
   • Libraries タブをクリックします
   • Install New をクリックします
   • DBFS または Workspace を選択し、アップロードした JAR ファイルを指定します
   • Install をクリックします

4. ライブラリを読み込むため、クラスターを再起動します

​

オプション 2: Databricks CLI でインストール

# JARをDBFSにアップロード
databricks fs cp clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}-{{ stable_version }}.jar \
  dbfs:/FileStore/jars/

# クラスターにインストール
databricks libraries install \
  --cluster-id <your-cluster-id> \
  --jar dbfs:/FileStore/jars/clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}-{{ stable_version }}.jar

​

オプション 3: Maven Coordinates (推奨)

1. Databricksワークスペースに移動します:

   • Compute → クラスターを選択します
   • Libraries タブをクリックします
   • Install New をクリックします
   • Maven タブを選択します

2. Maven coordinatesを追加します:

com.clickhouse.spark:clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }}

3. Install をクリックし、ライブラリを読み込むためにクラスターを再起動します

​

TableProvider API を使用する

Unity Catalog が有効 (デフォルト) な場合、Unity Catalog によって Spark のカタログ登録がブロックされるため、必ず TableProvider API (フォーマットベースのアクセス) を使用する必要があります。“No isolation shared” アクセスモードのクラスターを使用して Unity Catalog を無効にしている場合は、代わりに Catalog API を使用できます。

​

データの読み込み

Python

# TableProvider API を使用して ClickHouse から読み込む
df = spark.read \
    .format("clickhouse") \
    .option("host", "your-clickhouse-cloud-host.clickhouse.cloud") \
    .option("protocol", "https") \
    .option("http_port", "8443") \
    .option("database", "default") \
    .option("table", "events") \
    .option("user", "default") \
    .option("password", dbutils.secrets.get(scope="clickhouse", key="password")) \
    .option("ssl", "true") \
    .load()

# スキーマは自動的に推論されます
df.display()

Scala

val df = spark.read
  .format("clickhouse")
  .option("host", "your-clickhouse-cloud-host.clickhouse.cloud")
  .option("protocol", "https")
  .option("http_port", "8443")
  .option("database", "default")
  .option("table", "events")
  .option("user", "default")
  .option("password", dbutils.secrets.get(scope="clickhouse", key="password"))
  .option("ssl", "true")
  .load()

df.show()

​

データの書き込み

Python

# ClickHouse に書き込みます。テーブルが存在しない場合は自動的に作成されます
df.write \
    .format("clickhouse") \
    .option("host", "your-clickhouse-cloud-host.clickhouse.cloud") \
    .option("protocol", "https") \
    .option("http_port", "8443") \
    .option("database", "default") \
    .option("table", "events_copy") \
    .option("user", "default") \
    .option("password", dbutils.secrets.get(scope="clickhouse", key="password")) \
    .option("ssl", "true") \
    .option("order_by", "id") \  # 必須: 新しいテーブルを作成する場合は ORDER BY を指定します
    .option("settings.allow_nullable_key", "1") \  # 必須: ORDER BY に NULL 許容カラムが含まれる場合、ClickHouse Cloud では必要です
    .mode("append") \
    .save()

Scala

df.write
  .format("clickhouse")
  .option("host", "your-clickhouse-cloud-host.clickhouse.cloud")
  .option("protocol", "https")
  .option("http_port", "8443")
  .option("database", "default")
  .option("table", "events_copy")
  .option("user", "default")
  .option("password", dbutils.secrets.get(scope="clickhouse", key="password"))
  .option("ssl", "true")
  .option("order_by", "id")  // 必須: 新しいテーブルを作成する場合は ORDER BY を指定します
  .option("settings.allow_nullable_key", "1")  // 必須: ORDER BY に NULL 許容カラムが含まれる場合、ClickHouse Cloud では必要です
  .mode("append")
  .save()

この例では、Databricks でシークレットスコープが事前設定されていることを前提としています。設定手順については、Databricks の Secret management のドキュメント を参照してください。

​

Databricks 固有の考慮事項

​

アクセスモードの要件

ClickHouse Spark コネクタでは、Dedicated (旧称 Single User) アクセスモードが必要です。Unity Catalog が有効な構成では、Databricks が外部 DataSource V2 コネクタをブロックするため、Standard (旧称 Shared) アクセスモードはサポートされません。

Access Mode	Unity Catalog	Supported
Dedicated (Single User)	有効	✅ はい
Dedicated (Single User)	無効	✅ はい
Standard (Shared)	有効	❌ いいえ
Standard (Shared)	無効	✅ はい

​

シークレット管理

ClickHouse の認証情報を安全に保存するには、Databricks のシークレットスコープを使用します。

# シークレットにアクセスする
password = dbutils.secrets.get(scope="clickhouse", key="password")

設定手順については、Databricks の Secret management のドキュメント を参照してください。

​

ClickHouse Cloud への接続

Databricks から ClickHouse Cloud に接続する場合は、次の設定を行います。

1. HTTPS プロトコルを使用します (protocol: https, http_port: 8443)
2. SSLを有効にします (ssl: true)

​

例

​

完全なワークフローの例

Python

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# ClickHouse コネクタを使用して Spark を初期化
spark = SparkSession.builder \
    .config("spark.jars.packages", "com.clickhouse.spark:clickhouse-spark-runtime-3.4_2.12:0.9.0") \
    .getOrCreate()

# ClickHouse から読み込む
df = spark.read \
    .format("clickhouse") \
    .option("host", "your-host.clickhouse.cloud") \
    .option("protocol", "https") \
    .option("http_port", "8443") \
    .option("database", "default") \
    .option("table", "source_table") \
    .option("user", "default") \
    .option("password", dbutils.secrets.get(scope="clickhouse", key="password")) \
    .option("ssl", "true") \
    .load()

# データを変換
transformed_df = df.filter(col("status") == "active")

# ClickHouse に書き込む
transformed_df.write \
    .format("clickhouse") \
    .option("host", "your-host.clickhouse.cloud") \
    .option("protocol", "https") \
    .option("http_port", "8443") \
    .option("database", "default") \
    .option("table", "target_table") \
    .option("user", "default") \
    .option("password", dbutils.secrets.get(scope="clickhouse", key="password")) \
    .option("ssl", "true") \
    .option("order_by", "id") \
    .mode("append") \
    .save()

Scala

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col

// ClickHouse コネクタを使用して Spark を初期化
val spark = SparkSession.builder
  .config("spark.jars.packages", "com.clickhouse.spark:clickhouse-spark-runtime-3.4_2.12:0.9.0")
  .getOrCreate()

// ClickHouse から読み込む
val df = spark.read
  .format("clickhouse")
  .option("host", "your-host.clickhouse.cloud")
  .option("protocol", "https")
  .option("http_port", "8443")
  .option("database", "default")
  .option("table", "source_table")
  .option("user", "default")
  .option("password", dbutils.secrets.get(scope="clickhouse", key="password"))
  .option("ssl", "true")
  .load()

// データを変換
val transformedDF = df.filter(col("status") === "active")

// ClickHouse に書き込む
transformedDF.write
  .format("clickhouse")
  .option("host", "your-host.clickhouse.cloud")
  .option("protocol", "https")
  .option("http_port", "8443")
  .option("database", "default")
  .option("table", "target_table")
  .option("user", "default")
  .option("password", dbutils.secrets.get(scope="clickhouse", key="password"))
  .option("ssl", "true")
  .option("order_by", "id")
  .mode("append")
  .save()

​

関連ドキュメント

• Spark Native Connector Guide - コネクタに関する完全なドキュメント
• TableProvider API Documentation - フォーマットベースのアクセスの詳細
• Catalog API Documentation - カタログベースのアクセスの詳細