ClickHouseのHTTPインターフェースをAzure Data Factoryで使用する

azureBlobStorage テーブル関数は、Azure Blob StorageからClickHouseにデータを取り込むための迅速で便利な方法です。しかし、以下の理由から常に適しているわけではありません。

あなたのデータがAzure Blob Storageに保存されていない場合 — 例えば、Azure SQL Database、Microsoft SQL Server、またはCosmos DBなどに保存されている可能性があります。
セキュリティポリシーがBlob Storageへの外部アクセスを完全に防止している可能性がある — 例えば、ストレージアカウントがロックダウンされていて、公開エンドポイントがない場合などです。

そのようなシナリオでは、Azure Data Factoryを使用して ClickHouse HTTPインターフェースでAzureサービスからClickHouseにデータを送信できます。

この方法はフローを逆転させます: ClickHouseがAzureからデータを引き出すのではなく、Azure Data FactoryがデータをClickHouseにプッシュします。このアプローチは通常、あなたのClickHouseインスタンスがインターネットからアクセス可能であることを必要とします。

参考

Azure Data Factoryのセルフホステッドインテグレーションランタイムを使用することで、ClickHouseインスタンスをインターネットに露出させることを避けることが可能です。この設定により、プライベートネットワーク経由でデータを送信できます。ただし、この記事の範囲を超えています。より詳しい情報は公式ガイドでご覧いただけます: セルフホステッドインテグレーションランタイムの作成と設定

ClickHouseをRESTサービスに変える

Azure Data FactoryはJSON形式でHTTPを介して外部システムにデータを送信することをサポートしています。この機能を使用して、ClickHouse HTTPインターフェースを介して直接ClickHouseにデータを挿入できます。詳細はClickHouse HTTPインターフェースドキュメントを参照してください。

この例では、宛先テーブルを指定し、入力データ形式をJSONとして定義し、より柔軟なタイムスタンプ解析を可能にするオプションを含めるだけで済みます。

INSERT INTO my_table
SETTINGS 
    date_time_input_format='best_effort',
    input_format_json_read_objects_as_strings=1
FORMAT JSONEachRow

このクエリをHTTPリクエストの一部として送信するには、クリックハウスエンドポイントのクエリパラメータにURLエンコードされた文字列として渡すだけです:

https://your-clickhouse-url.com?query=INSERT%20INTO%20my_table%20SETTINGS%20date_time_input_format%3D%27best_effort%27%2C%20input_format_json_read_objects_as_strings%3D1%20FORMAT%20JSONEachRow%0A

参考

Azure Data Factoryは、組み込みのencodeUriComponent関数を使用してこのエンコーディングを自動的に処理できるため、手動で行う必要はありません。

これで、JSON形式のデータをこのURLに送信できます。データはターゲットテーブルの構造に一致する必要があります。以下は、3つのカラム col_1、col_2、col_3 を持つテーブルを想定したcurlを使用したシンプルな例です。

curl \
    -XPOST "https://your-clickhouse-url.com?query=<our_URL_encded_query>" \
    --data '{"col_1":9119,"col_2":50.994,"col_3":"2019-06-01 00:00:00"}'

また、オブジェクトのJSON配列またはJSON Lines（改行区切りのJSONオブジェクト）を送信することもできます。Azure Data FactoryはJSON配列形式を使用し、ClickHouseのJSONEachRow入力と完全に互換性があります。

このステップでは、ClickHouse側で特別な作業を行う必要はありません。HTTPインターフェースはRESTのようなエンドポイントとして機能するために必要なものをすでに提供しています — 追加の設定は不要です。

ClickHouseがRESTエンドポイントのように動作するようになったので、次にAzure Data Factoryがそれを使用するように設定します。

次のステップでは、Azure Data Factoryインスタンスを作成し、ClickHouseインスタンスへのリンクサービスを設定し、 RESTシンク用のデータセットを定義し、AzureからClickHouseにデータを送信するためのCopy Dataアクティビティを作成します。

Azure Data Factoryインスタンスの作成

このガイドは、Microsoft Azureアカウントにアクセスでき、サブスクリプションとリソースグループがすでに設定されていることを前提としています。すでにAzure Data Factoryが設定されている場合は、このステップをスキップし、既存のサービスを使用して次のステップに進むことができます。

Microsoft Azureポータルにログインし、リソースの作成をクリックします。
左のカテゴリペインで分析を選択し、人気のサービスリストからData Factoryをクリックします。
サブスクリプションとリソースグループを選択し、新しいData Factoryインスタンスの名前を入力し、リージョンを選択してバージョンをV2のままにします。
確認 + 作成をクリックし、次に作成をクリックしてデプロイを開始します。

デプロイが成功裏に完了すると、新しいAzure Data Factoryインスタンスを使用し始めることができます。

新しいRESTベースのリンクサービスの作成

Microsoft Azureポータルにログインし、Data Factoryインスタンスを開きます。
Data Factoryの概要ページで、スタジオを起動をクリックします。
左側のメニューで管理を選択し、リンクサービスに移動して、+ 新規をクリックして新しいリンクサービスを作成します。
新しいリンクサービス検索バーにRESTと入力し、RESTを選択して続行をクリックして RESTコネクタインスタンスを作成します。
リンクサービス構成ペインで、新しいサービスの名前を入力し、基本URLフィールドをクリックし、次に動的コンテンツを追加をクリックします（このリンクはフィールドが選択されている時のみ表示されます）。
動的コンテンツペインで、パラメータ化されたURLを作成できます。これにより、異なるテーブルのデータセットを作成するときにクエリを後で定義できるので、リンクサービスが再利用できるようになります。
フィルタ入力の横にある**"+"**をクリックして新しいパラメータを追加し、名前をpQueryとし、タイプをStringに設定し、デフォルト値をSELECT 1に設定します。保存をクリックします。
式フィールドに以下を入力し、OKをクリックします。your-clickhouse-url.comを実際のClickHouseインスタンスのアドレスに置き換えてください。

@{concat('https://your-clickhouse-url.com:8443/?query=', encodeUriComponent(linkedService().pQuery))}

New Linked Service Expression Field Filled

メインフォームに戻り、基本認証を選択し、ClickHouse HTTPインターフェースに接続するために使用したユーザー名とパスワードを入力し、接続をテストをクリックします。すべてが正しく構成されている場合、成功メッセージが表示されます。
作成をクリックして設定を完了します。

これで、新しく登録されたRESTベースのリンクサービスがリストに表示されるはずです。

ClickHouse HTTPインターフェース用の新しいデータセットを作成する

ClickHouse HTTPインターフェース用にリンクサービスが設定されたので、Azure Data FactoryがClickHouseにデータを送信する際に使用するデータセットを作成できます。

この例では、環境センサーのデータの一部を挿入します。

選択したClickHouseクエリコンソールを開きます — これはClickHouse CloudのWeb UI、CLIクライアント、またはクエリを実行するために使用する他のインターフェースのいずれかです — そしてターゲットテーブルを作成します:

CREATE TABLE sensors
(
    sensor_id UInt16,
    lat Float32,
    lon Float32,
    timestamp DateTime,
    temperature Float32
)
ENGINE = MergeTree
ORDER BY (timestamp, sensor_id);

Azure Data Factory Studioで、左側のペインから「作成」を選択します。データセット項目の上にカーソルを合わせ、3点アイコンをクリックし、新しいデータセットを選択します。
検索バーにRESTと入力し、RESTを選択し、続行をクリックします。データセットの名前を入力し、前のステップで作成したリンクサービスを選択します。OKをクリックしてデータセットを作成します。
これで、新しく作成したデータセットが左側のファクトリリソースペインのデータセットセクションに表示されます。データセットを選択してプロパティを開きます。リンクサービスで定義されたpQueryパラメータが表示されます。値テキストフィールドをクリックし、次に動的なコンテンツを追加をクリックします。
開いたペインに、次のクエリを貼り付けます:

INSERT INTO sensors
SETTINGS 
    date_time_input_format=''best_effort'', 
    input_format_json_read_objects_as_strings=1 
FORMAT JSONEachRow

危険

クエリ内のすべてのシングルクォート ' は二重のシングルクォート '' に置き換える必要があります。これはAzure Data Factoryの式パーサーによって要求されます。エスケープしないと、すぐにはエラーが表示されない場合がありますが、データセットを使用または保存しようとするときに失敗します。例えば、'best_effort'は''best_effort''のように書く必要があります。

式を保存するにはOKをクリックします。接続をテストをクリックします。すべてが正しく構成されていれば、接続成功メッセージが表示されます。ページの上部にあるすべて公開をクリックして変更を保存します。

サンプルデータセットの設定

この例では、完全な環境センサーのデータセットではなく、センサーのデータセットサンプルの小さなサブセットを使用します。

参考

このガイドを焦点を絞って保つために、Azure Data Factoryでのソースデータセットの作成手順については詳しく説明しません。サンプルデータをお好みのストレージサービスにアップロードできます — 例えば、Azure Blob Storage、Microsoft SQL Server、またはAzure Data Factoryがサポートする別のファイル形式などです。

データセットをAzure Blob Storage（または別の好ましいストレージサービス）にアップロードします。その後、Azure Data Factory Studioでファクトリリソースペインに移動します。アップロードされたデータを指す新しいデータセットを作成します。すべてを公開をクリックして変更を保存します。

ClickHouseにデータを転送するためのCopyアクティビティの作成

入力データセットと出力データセットの両方を構成したので、Copy Dataアクティビティを設定して、サンプルデータセットからClickHouseのsensorsテーブルにデータを転送できます。

Azure Data Factory Studioを開き、作成タブに移動します。ファクトリリソースペインで、Pipelineにカーソルを合わせ、3点アイコンをクリックして新しいパイプラインを選択します。
アクティビティペインで、移動と変換セクションを展開し、データのコピーアクティビティをキャンバスにドラッグします。
ソースタブを選択し、先ほど作成したソースデータセットを選択します。
シンクタブに移動し、センサー用のClickHouseデータセットを選択します。リクエストメソッドをPOSTに設定します。HTTP圧縮タイプがなしに設定されていることを確認します。

注意
HTTP圧縮は、Azure Data FactoryのCopy Dataアクティビティでは正しく機能しません。有効にすると、Azureはゼロバイトのみのペイロードを送信します — サービスのバグの可能性があります。圧縮は無効のままにしておいてください。

参考
バッチサイズは10,000のデフォルトを維持するか、さらなる増加をお勧めします。詳細については、挿入戦略の選択 / 同期的な場合のバッチ挿入をご覧ください。
キャンバスの上部にあるデバッグをクリックしてパイプラインを実行します。少し待つと、アクティビティがキューに送信され実行されます。すべてが正しく構成されていれば、タスクは成功のステータスで終了するはずです。
完了したら、すべてを公開をクリックしてパイプラインとデータセットの変更を保存します。