ClickHouse®のApache Kafka®とAivenの接続

<font color="Grey">*[Connecting Apache Kafka® and Aiven for ClickHouse®](https://aiven.io/blog/connecting-kafka-and-clickhouse)の翻訳です。*</font>

データをダウンサンプリングすることなく、膨大なデータに対して高速なレスポンスタイムで分析を実行する方法をご紹介します。

# ClickHouse®のためのApache Kafka®とAivenの接続

検索デベロッパーセンター

Apache Kafka®とClickHouse®は全く異なりますが、多くの共通点もあります。どちらもオープンソースであり、拡張性が高く、不変のデータに最適で、大量のデータを処理することができます。そのため、これらのテクノロジーは競合するのではなく、実際には互いに補完し合っているのです。


Apache Kafkaはリアルタイムのデータフィードを処理するのに優れている。しかし、場合によっては古いレコードに戻って、後からデータを分析・処理する必要があります。ストリーミング・プラットフォームであるApache Kafkaは、大規模なデータチャンクにアクセスしたり、OLAP（オンライン分析処理）エンジンとして機能したりするのに最適化されていないため、これは難しい。


一方、ClickHouseは、ペタバイト級のデータを扱えるように設計されたスケーラブルで信頼性の高いストレージ・ソリューションであると同時に、高速なオンライン分析処理のための強力なツールであり、多くの企業でデータ分析に使用されています。


この2つの技術を組み合わせることで、ClickHouseの高性能なデータウェアハウスを実現し、Apache Kafkaから常に新鮮なデータを取得することで、常に最新の状態を保つことができます。


Apache Kafkaのトピックは、リアルタイムのデータが流れる川と考えることができます。一方、ClickHouseは、すべてのデータが最終的に行き着く海です。


それでは、袖をまくってこの2つのデータ・ソリューションを実際に統合してみましょう。以下、ステップ・バイ・ステップでサービスを作成し、それらを統合し、いくつかのクエリ実験を実行する。


## サービスの作成


セットアップを簡単にするために、Aivenが運営するマネージドバージョンのApache KafkaとClickHouseを使用します。もしまだAivenのアカウントを持っていなくても心配はいらない。[登録](https://console.aiven.io/signup/email)はほんの一歩で、この実験に無料トライアルを使うことができる。


[Aivenのコンソール](https://console.aiven.io/)から直接Aiven for ClickHouseとAiven for Apache Kafkaサービスを作成することができます。以下の例では、`apache-kafka-service`と`clickhouse-service`をサービス名として使用していますが、もっとクリエイティブにすることができます ;)


なお、Aiven for ClickHouseはインテグレーションを追加できるようにするため、少なくともスタートアッププランが必要です。


サービスを作成したら、完全にデプロイされ `RUNNING` 状態になるまで待ちます。これで準備完了です！


## Apache Kafka の準備


Apache KafkaからClickHouseにデータを移動するためには、まずApache Kafkaにデータがある必要がある。そこで、まずはApache Kafkaにトピックを作成します。[Aivenのコンソールから直接](https://docs.aiven.io/docs/products/kafka/howto/create-topic)できます。名前を`measurements`とする。ここでは、想像上のデバイスの連続測定値を送信する。


新しいデータの継続的なフローを模倣するために、短いbashスクリプトを使用します。このスクリプトでは、イベントのタイムスタンプ、デバイスのID、値の3つのプロパティを持つJSONオブジェクトを作成する。そして、このオブジェクトを `kcat` を使って `measurements` というトピックに送信する。kcat`の設定方法については、[こちらの記事を参照](https://docs.aiven.io/docs/products/kafka/howto/kcat)。



```
`#!/bin/bash

while ：
do
 stamp=$(日付 +%s)
 id=$((RANDOM%100))
 val=$((RANDOM%1000))
 echo "{\"timestamp\":$stamp,\"device\_id\":$id,\"value\":$val}"\
 | kcat -F kcat.config -P -t measurements
done`クリップボードにコピー
```

スクリプトを起動し、実行したままにしておくと、継続的にメッセージを作成し、トピックに送信する。


Apache Kafka側での作業は完了です。では、ClickHouseに移動しよう。


## Aiven for ClickHouseをApache Kafkaに接続する。


Aiven for ClickHouseサービスをApache Kafkaサービスと統合することは可能ですが、同じAivenプロジェクト内に2つのサービスを持つことで、簡単に統合することができます。


Aiven for ClickHouseとApache Kafkaを統合するには、2つのステップを行う必要があります：


1.接続を確立する。
2.統合データの構造と出所を指定する。


これらの手順を[Aiven CLI](https://docs.aiven.io/docs/tools/cli)の助けを借りて行う。


まず、`clickhouse_kafka`型のインテグレーションを作成し、サービス名、ソースにApache Kafka、デスティネーションにClickHouseを指定して接続を確立します：



```
avn service integration-create
 --integration-type clickhouse_kafka ￤-ソースサービス
 --source-service apache-kafka-service ╱ --service clickhouse-service
 --dest-service clickhouse-service`Copy to clipboard
```

このコマンドを実行しても（問題がなければ）何も返ってこない。しかし、Aiven for ClickHouseサービスで利用可能なデータベースのリストを確認すると（例えば[Aiven's console](https://console.aiven.io/)を使って）、`service_apache-kafka-service`という新しいデータベースがあることに気づくでしょう。作成されたデータベースの名前は、`service_`とApache Kafkaサービス名の組み合わせです。


Apache Kafkaサービスからどのようなデータを取得するかはまだ指定していないので、データベースは空のままです。JSONペイロードでデータソースを定義できますが、まずは統合のIDを見つける必要があります。このコマンドを実行することで取得できる：



```
avn service integration-list clickhouse-service | grep apache-kafka-service`クリップボードにコピーする
```

![avn service integration-list terminal output](https://cdn.sanity.io/images/sczeoy4w/production/d8afc96704b04356131e676474a66d1969a2f41b-1704x130.png?q=80&fit=min&auto=format&dpr=1.5)


私の場合、統合IDは`88546a37-5a8a-4c0c-8bd7-80960e3adab0`でした。あなたの場合は違う[UUID](https://en.wikipedia.org/wiki/Universally_unique_identifier)になるでしょう。


統合IDを知ることで、それを指定する接続に適切な設定を行うことができます：


* トピック `measurements` からデータを取得したい。
* データは JSON 形式（特に `JSONEachRow` ）である。
* データは3つのカラムを持つテーブルに変換される：データは `timestamp`、`device_id`、`value` の 3 つのカラムを持つテーブルに変換される：



```
 avnサービス統合-update 88546a37-5a8a-4c0c-8bd7-80960e3adab0 \
 --user-config-json '{ {
 "tables"：[
 {
 "name"："measurements_from_kafka"、
 "columns"：[
 {"name"："timestamp", "type"："DateTime"}、
 {name"：「device_id", "type"：「int8"}、
 {name"：「value"、"type"：「Int16"} である。
[ ],
 "トピック"：[name": "測定値"}]、
 "data_format"："JSONEachRow"、
 "group_name"："measurements_from_kafka_consumer"
 }
[ ]
 クリップボードにコピーする
```

ClickHouseは、`group_name`フィールドで指定したコンシューマーグループを使用して、トピックからどのメッセージが消費されたかを追跡します。デフォルトでは、各エントリーを1回ずつ読み込みます。データを2回取得したい場合は、別のグループ名でテーブルのコピーを作成します。


## Clickhouse から Kafka メッセージをオンザフライで読み込む


ClickHouseからApache Kafkaトピックからデータを読み込むには、すでにこのセットアップで十分です。  

ClickHouseのSQLコマンドを実行する最も便利な方法は、**clickhouse-client**を使用することです。実行方法がわからない場合は、[CLIでClickHouse®クラスタに接続する](https://docs.aiven.io/docs/products/clickhouse/howto/use-cli)を確認してください。


例えば、私は docker を使い、以下のコマンドでクライアントを実行しました。USERNAME、PASSWORD、HOST、PORTはあなたの値で置き換えてください。



```
ドッカーラン
 --rm clickhouse/clickhouse-client ˶ˆ꒳ˆ˵
 --ユーザー名
 --パスワード PASSWORD
 --host HOST
 --ポート
 --secure`Copy to clipboard
```

クライアントに入ったら、データベースのリストを確認できます。



```
SHOW DATABASES`クリップボードにコピーする
```

![データベースの端末出力を表示](https://cdn.sanity.io/images/sczeoy4w/production/4db3961fd96b5e2bb0abbe14ef19684d9e63772b-858x502.png?q=80&fit=min&auto=format&dpr=1.5)


インテグレーション`service_apache-kafka-service`（もしかしたら違う名前かもしれません！）を確立して作成したものが表示されます。


このデータベースからテーブルのリストを取得すると、統合設定で指定したテーブル名が表示されます。



```
SHOW TABLES FROM `service_apache-kafka-service``Copy to clipboard
```

![show tables terminal output](https://cdn.sanity.io/images/sczeoy4w/production/57498b816a584c04b083344abcf82082fcdcc7d6-866x269.png?q=80&fit=min&auto=format&dpr=1.5)


でその構造を再確認することもできる。



```
DESCRIBE `service_apache-kafka-service`.measurements_from_kafka`Copy to clipboard
```

![describe tables terminal output](https://cdn.sanity.io/images/sczeoy4w/production/9e519c766d273ef40df06f3fb86f5165632a3445-1928x434.png?q=80&fit=min&auto=format&dpr=1.5)


さて、このテーブルから直接読み込みたいかもしれない。しかし、メッセージを消費できるのは一度だけだということを覚えておいてほしい！つまり、一度読み込んだ項目は消えてしまうのだ。それでも、以下のコマンドを実行することを止めることはできない：



```
SELECT ˶* FROM `service_apache-kafka-service`.measurements_from_kafka LIMIT 100`Copy to clipboard
```


```
SELECT count( \*) FROM `service_apache-kafka-service`.measurements_from_kafka`Copy to clipboard
```

![接続テーブルの端末出力から直接項目を選択する](https://cdn.sanity.io/images/sczeoy4w/production/d9dfda0c13b811cbe392a804b2796350c2c8414e-1084x311.png?q=80&fit=min&auto=format&dpr=1.5)


しかし、これはApache Kafkaからデータを消費する最も便利な方法ではありません。ほとんどの場合、ClickHouseのデータ・アイテムをコピーして保存しておきたいでしょう。これこそが次のセクションで行うことです。


## KafkaメッセージをClickhouseテーブルに保持する


Apache KafkaからClickHouseに送られてくるデータを保存するには、2つのパーツが必要です：


1.すべてのデータが永久に保存される保存先テーブル。
2.コネクタテーブル (`measurements_from_kafka`) とデスティネーションテーブル間のブリッジとなるマテリアライズドビュー。


この2つのクエリで作成できます：



```
CREATE TABLE device_measurements (timestamp DateTime, device_id Int8, value Int16)
ENGINE = ReplicatedMergeTree()
ORDER BY timestamp;`クリップボードにコピーする
```


```
CREATE MATERIALIZED VIEW materialised_view TO device_measurements AS
SELECT \* FROM `service_apache-kafka-service`.measurements_from_kafka;`Copy to clipboard
```

マテリアライズドビューを作成すると、実際には裏でトリガーが追加されます。このトリガーは、テーブル `measurements_from_kafka` に追加された新しいデータに反応する。一度トリガーがかかると、データはマテリアライズドビューを通過して（必要であれば変換することもできます） `device_measurements` テーブルに入ります。


実行すると、データが流れていることを確認できる：



```
SELECT COUNT(*) from device_measurements`Copy to clipboard
```

デバイスからのすべての測定値をカウントするクエリを実行し、どのデバイスの平均値が高いかを確認できます。ここでは、`bar` 関数を使用したシンプルで視覚的なメカニズムを使用します。



```
セレクト
 device_id、
 count() as readings_number、
 bar(avg(value), 0, 1000, 100) as average_measurement_value
FROM device_measurements
GROUP BY device_id
ORDER BY device_id ASC`クリップボードにコピー
```

![バーターミナル出力でデータを視覚化する](https://cdn.sanity.io/images/sczeoy4w/production/510274a24a9d839605dadb379f3fa793551dde67-2090x938.png?q=80&fit=min&auto=format&dpr=1.5)


## 結論


これで、Aiven for ClickHouseにデータを取り込み、マテリアライズド・ビューを使ってデータを保存するスキルが身につきました。他にもいくつか資料がありますので、ご興味のある方はご覧ください：


* ClickHouseとは何か、[どのようにして高いパフォーマンスを実現するのか](https://aiven.io/blog/what-is-clickhouse)
* [Aiven for ClickHouseを始める](https://docs.aiven.io/docs/products/clickhouse/getting-started)
* [Aiven for ClickHouseをApache Kafkaと統合する](https://docs.aiven.io/docs/products/clickhouse/howto/integrate-kafka)方法に関するドキュメント（ここでは省略されている詳細もあります）。
* [Aiven for ClickHouseで利用可能なその他の統合](https://docs.aiven.io/docs/products/clickhouse/howto/list-integrations)に関する情報