【第５回】Confluent Platformのデモを試してみる(ksqlDB編)

###概要
今回は、[第一回](https://qiita.com/Alex-hiro/items/0209b2faec9c00cf3ca1)で構築したConfluentのクラスタを利用しConfluent Control CenterにログインしてksqlDBの機能を確認してみたいと思います。

###Confluent Control Centerにログイン
１.以下のURLでWebブラウザからコンフルエントコントロールセンターのGUIにアクセスします。

```sh
http://localhost:9021
````

ここでは、ksqlDBは認証されKafkaクラスターに接続してksqlDBのコマンドファイルで定義されているクエリを既に実行していることが確認できます。

2.左メニューから「ksqlDB」をクリックします。
3.ksqlDB Applicationの箇所から「wikipedia」を選択します。
![スクリーンショット 2021-11-16 15.55.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/d79d2957-b7a3-1605-a7e5-2862038868e9.png)
4.2の手順でwikipediaをクリックして、「Flow」を表示することでこのデモで作成されたストリームとテーブルおよびそれらの関連性を確認することができます。GUIだとわかりやすいですね。
![スクリーンショット 2021-11-16 16.48.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/9911177e-790e-cac4-4be5-0316d3d4472d.png)
5.次にksqlDBCLIを実行してksqlDBCLIプロンプトを表示します。

```sh
docker-compose exec ksqldb-cli bash -c 'ksql -u ksqlDBUser -p ksqlDBUser http://ksqldb-server:8088'
````
6.既存のksqlDBのストリーム（Streams)を表示します。
![スクリーンショット 2021-11-16 16.38.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/179abdaf-3f52-b5c9-abdf-64d3b18af378.png)
7.ここで「WIKIPEDIA」をクリックして既存のksqlDBストリームのスキーマの詳細を表示します。
![スクリーンショット 2021-11-16 16.45.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/8e122d8d-176f-758a-b3a1-cb06880edd93.png)
8.既存のksqlDBテーブルを表示します。このテーブルは[タンブリンウィンドウ](https://docs.ksqldb.io/en/latest/concepts/time-and-windows-in-ksqldb-queries/?&_ga=2.34566564.44184953.1637044072-965107260.1611725508#tumbling-window)内の発生をカウントします。
![スクリーンショット 2021-11-16 16.57.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/03166ca7-b2c7-4674-d3b9-98f0f1ba68d5.png)
9.継続して実行されている既存のksqlDBのクエリ(Persistent queries)を見てみましょう。
![スクリーンショット 2021-11-16 17.05.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/f203f381-0c98-0368-7014-a18cb5b1a06b.png)
10.次にさまざまなksqlDBストリームおよびテーブルからのメッセージを表示します。選択したストリームをクリックして、「Query stream」をクリックしてクエリエディタを開きます。エディタには「select * from WIKIPEDIA EMIT CHANGES;」
といったクエリがあらかじめ入力されており、このクエリの指示に従って新しく届いたデータの結果がここに表示されます。
![スクリーンショット 2021-11-16 17.14.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/e39a8766-a4ac-06ea-34d8-c6a53d565c7b.png)
11.ksqlDB Editorをクリックして「SHOW PROPERTIES;」と入力し実行します。この実行結果からksqlDBサーバのプロパティが確認できます。この内容を[docker-compose.yml](https://github.com/confluentinc/cp-demo/blob/7.0.0-post/docker-compose.yml)で照合も可能です。
![スクリーンショット 2021-11-26 17.09.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/74236234-da1a-a4f2-9bbb-a110a5fd2f07.png)
12.[ksqlDBプロセスログ](https://docs.ksqldb.io/en/latest/reference/processing-log/?_ga=2.245158917.1090705967.1637912303-965107260.1611725508)は、処理中のレコードのエラーをキャプチャできるので開発者はksqlDBクエリのデバッグに役立ちます。この例のプロセスログは[log4jproperties file](https://github.com/confluentinc/cp-demo/blob/7.0.0-post/scripts/helper/log4j-secure.properties)で構成されているmTLSの認証を使用してkafkaトピックにエントリを書き込みます。動作確認のために以下の不完全なクエリを２０秒間ksqlDBエディタで実行します。

```sh
SELECT 1/0 FROM wikipedia EMIT CHANGES;
````
このクエリからレコードは返ってこないはずです。ksqlDBはレコードごとにプロセスログにエラーを書き込みます。ここでは、対応するksqlDBストリーム"KSQL_PROCESSING_LOG"をksqlDBエディタで表示をしてみます。(この時auto.offset.reset=Earliestを設定します。)

```sh
SELECT * FROM KSQL_PROCESSING_LOG EMIT CHANGES;
````
以下のような表示がされプロセスログにエラーが出力されていることが確認できました。
![スクリーンショット 2021-11-26 18.21.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/128268/fd3da941-65a6-070e-b0a4-0d456542e8e0.png)

上記を試してみてksqlDBはSQLライクな構文でkafkaのデータにアクセスできるのとGUIから利用できるので運用担当者にとっても扱いやすと思いました。

次回は、コンシューマーについてみていきたいと思っています。
今回も最後までお読み頂きありがとうございました！