MySQLのTable更新情報をKafka上でksql処理する

# ねらい

大仰なタイトルだがその実態は下記チュートリアルを体験してみようというもの。「Materialized Cache」というと何のことやらだが、要はTopic内の情報をksqlのtable/stream表現で扱うことらしい。このチュートリアルでは[Debezium](https://debezium.io/)を用いてMySQLの更新ログをKafka Topicに流し込み、ksqlで処理する。DebeziumはいわゆるChange Data CaptureのためのConnectorであり、この一連の処理の流れは私の秘めたる最終目的とも合致している。

元ネタ：[Materialized cache](https://docs.ksqldb.io/en/latest/tutorials/materialized/)

# 関連Qiita記事
[1.ksqlDB Quickstartを実施してみる](https://qiita.com/ssqwork/items/449b557f5237fe61e8b8)
[2.ksqlDB を使用したストリーミングクエリの作成(前半)](https://qiita.com/ssqwork/items/7776d1fca1942639396d)
[3.ksqlDB を使用したストリーミングクエリの作成(後半)](https://qiita.com/ssqwork/items/598e7c58982464ff7f46)
[4.MySQLのTable更新情報をKafka上でksql処理する](https://qiita.com/ssqwork/items/bc5ff578b1bc70ff26e1)

# チュートリアル実施

## 事前準備

**その1 : Debeziumの入手**

[Confluent Hub](https://docs.confluent.io/home/connect/confluent-hub/client.html)というダウンロードツールを用いてDebeziumを入手する。(Confluent HubはCommunity Editionには同梱されていないようだったので別途導入した。)
connect-ditributed.propertiesを書き換えようとして失敗しているが、今回はdocker composeで環境を作る際の元ネタのlibararyがあればよいだけのはずなので気にせず進める。

```:Confluent-Hub実行結果
gen@LAPTOP-O8FG8ES2:~/docker_files$ confluent-hub install --component-dir confluent-hub-components --no-prompt debezium/debezium-connector-mysql:1.1.0
Running in a "--no-prompt" mode
Implicit acceptance of the license below:
Apache 2.0
https://github.com/debezium/debezium/blob/master/LICENSE.txt
Implicit confirmation of the question: You are about to install 'debezium-connector-mysql' from Debezium Community, as published on Confluent Hub.
Downloading component Debezium MySQL CDC Connector 1.1.0, provided by Debezium Community from Confluent Hub and installing into confluent-hub-components
Adding installation directory to plugin path in the following files:
  /etc/kafka/connect-distributed.properties
Unable to update Worker's configuration file /etc/kafka/connect-distributed.properties
Unable to save to file /etc/kafka/connect-distributed.properties

gen@LAPTOP-O8FG8ES2:~/docker_files/confluent-hub-components/debezium-debezium-connector-mysql$ ls -l
total 16
drwxr-xr-x 2 gen gen 4096 Nov 11 15:35 assets
drwxr-xr-x 2 gen gen 4096 Nov 11 15:35 doc
drwxr-xr-x 2 gen gen 4096 Nov 11 15:35 lib
-rw-r--r-- 1 gen gen 2868 Nov 11 15:35 manifest.json
```

このライブラリはdocker-compose.ymlによってKSQL_CONNECT_PLUGIN_PATHで指定されたディレクトリとしてマウントされる。

```:docker-compose.yml抜粋
      ...
    volumes:
      - "./confluent-hub-components/:/usr/share/kafka/plugins/"
    environment:
      ...
      KSQL_CONNECT_PLUGIN_PATH: "/usr/share/kafka/plugins"
```

なお[How to use connector management](https://docs.ksqldb.io/en/latest/how-to-guides/use-connector-management/) に詳しいやり方が書いてあった。

**その2: mysqlのconfig準備**

docker composeで環境を作る際の元ネタになるmysqlのConfigを準備する。mysqlディレクトリを掘り、中に以下のファイルを配置しておく。どうも[「バイナリログに行の変更を全て書き出せ」](https://dev.mysql.com/doc/refman/5.6/ja/binary-log.html)という指令のようだ。わーいx'99'ログだぁ(独白)。

```:custom-config.cnf
[mysqld]
server-id                = 223344 
log_bin                  = mysql-bin 
binlog_format            = ROW 
binlog_row_image         = FULL 
expire_logs_days         = 10
gtid_mode                = ON
enforce_gtid_consistency = ON
```

```docker-compose.yml抜粋
    volumes:
      - "./mysql/custom-config.cnf:/etc/mysql/conf.d/custom-config.cnf"
```

## Start the stack

困ったことに元ネタのdocker-compose.ymlのままだとSchema Registryのコンテナ起動に失敗したため、以下の変更を行っている。それにつけても[StackOverflow](https://stackoverflow.com/questions/61927490/docker-schema-registry-and-kafka-rest-cannot-up)様様である。

```:docker-compose.ymlの変更
(元記述)
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181'
      
(変更後)
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "PLAINTEXT://broker:9092"
```

Docker Composeで環境を立ち上げた後のステータス。

```:稼働コンテナ
gen@LAPTOP-O8FG8ES2:~$ docker ps
CONTAINER ID   IMAGE                                   COMMAND                  CREATED         STATUS         PORTS                                        NAMES
b6b396095bc6   confluentinc/ksqldb-cli:0.21.0          "/bin/sh"                8 minutes ago   Up 8 minutes                                                ksqldb-cli
f7078c17dbe0   confluentinc/ksqldb-server:0.21.0       "/usr/bin/docker/run"    8 minutes ago   Up 8 minutes   0.0.0.0:8088->8088/tcp                       ksqldb-server
27eafbdc49a4   confluentinc/cp-schema-registry:7.0.0   "/etc/confluent/dock…"   8 minutes ago   Up 8 minutes   0.0.0.0:8081->8081/tcp                       schema-registry
582265bdf311   confluentinc/cp-kafka:7.0.0             "/etc/confluent/dock…"   4 hours ago     Up 8 minutes   9092/tcp, 0.0.0.0:29092->29092/tcp           broker
2ad636a2d60b   confluentinc/cp-zookeeper:7.0.0         "/etc/confluent/dock…"   4 hours ago     Up 8 minutes   2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp   zookeeper
909ab8e55de5   mysql:8.0.19                            "docker-entrypoint.s…"   4 hours ago     Up 8 minutes   0.0.0.0:3306->3306/tcp, 33060/tcp            mysql
```

```:docker_image
gen@LAPTOP-O8FG8ES2:~$ docker image ls
REPOSITORY                        TAG       IMAGE ID       CREATED         SIZE
confluentinc/cp-schema-registry   7.0.0     24c8939a7458   3 weeks ago     1.65GB
confluentinc/cp-kafka             7.0.0     42807c42b958   3 weeks ago     791MB
confluentinc/cp-zookeeper         7.0.0     3858ad02e9d1   3 weeks ago     791MB
confluentinc/ksqldb-cli           0.21.0    187badc13e5b   2 months ago    852MB
confluentinc/ksqldb-server        0.21.0    187badc13e5b   2 months ago    852MB
mysql                             8.0.19    0c27e8e5fcfa   18 months ago   546MB
```

Dockerで環境作れるのは楽だけど、数GBはダウンロードするのが嫌・・・自分で全く土地勘のないmysqlを構成するよりはいいか。

## Configure MySQL for Debezium

mysqlのコンテナに入り、MySQL CLIからレプリケーションに必要なGRANTを実施する。

```:GRANT処理
mysql> GRANT ALL PRIVILEGES ON *.* TO 'example-user' WITH GRANT OPTION;
Query OK, 0 rows affected (0.02 sec)

mysql> ALTER USER 'example-user'@'%' IDENTIFIED WITH mysql_native_password BY 'example-pw';
Query OK, 0 rows affected (0.01 sec)

mysql> FLUSH PRIVILEGES;
Query OK, 0 rows affected (0.01 sec)
```

## Create the calls table in MySQL

続けてcall-centerデータベース内にcallsテーブルを作る。コールセンター業務系の記録テーブルであるcallsには「コールした人の名前」、「コールの理由」、「コール時間(秒)」が含まれる。ところでderekよ一体何を買ったんだ？

```:DBおよびテーブル作成
(DB設定)
mysql> USE call-center;
Database changed

(テーブル作成)
mysql> CREATE TABLE calls (name TEXT, reason TEXT, duration_seconds INT);
Query OK, 0 rows affected (0.08 sec)

(行挿入)
...
mysql> INSERT INTO calls (name, reason, duration_seconds) VALUES ("derek", "refund", 325);
Query OK, 1 row affected (0.01 sec)

(テーブル確認)
mysql> select * from calls;
+---------+----------+------------------+
| name    | reason   | duration_seconds |
+---------+----------+------------------+
| michael | purchase |              540 |
| michael | help     |              224 |
| colin   | help     |              802 |
| derek   | purchase |            10204 |
| derek   | help     |              600 |
| colin   | refund   |              105 |
| michael | help     |             2030 |
| colin   | purchase |              800 |
| derek   | help     |             2514 |
| derek   | refund   |              325 |
+---------+----------+------------------+
10 rows in set (0.00 sec)
```

## Start the Debezium connector

ksqldb-cliのコンテナからksqldbCLIを起動し、まずは常にtopicの先頭からメッセージを読む設定を行う。

```:環境変数設定
ksql> SET 'auto.offset.reset' = 'earliest';
Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
```

以下のksqlでSource Connectorの使用と構成をkafkaに通知する。え、これだけでいいの？

```:Source_Connectorの開始
ksql> CREATE SOURCE CONNECTOR calls_reader WITH (
>    'connector.class' = 'io.debezium.connector.mysql.MySqlConnector',
>    'database.hostname' = 'mysql',
>    'database.port' = '3306',
>    'database.user' = 'example-user',
>    'database.password' = 'example-pw',
>    'database.allowPublicKeyRetrieval' = 'true',
>    'database.server.id' = '184054',
>    'database.server.name' = 'call-center-db',
>    'database.whitelist' = 'call-center',
>    'database.history.kafka.bootstrap.servers' = 'broker:9092',
>    'database.history.kafka.topic' = 'call-center',
>    'table.whitelist' = 'call-center.calls',
>    'include.schema.changes' = 'false'
>);

 Message
--------------------------------
 Created connector CALLS_READER
--------------------------------
```

確かにcall-center-db.call-center.callというTopicが作られている。

```:SWOW_TOPICS応答
 Kafka Topic                      | Partitions | Partition Replicas
--------------------------------------------------------------------
 _ksql-connect-configs            | 1          | 1
 _ksql-connect-offsets            | 25         | 1
 _ksql-connect-statuses           | 5          | 1
 call-center                      | 1          | 1
 call-center-db.call-center.calls | 1          | 1
 default_ksql_processing_log      | 1          | 1
--------------------------------------------------------------------
```

Topicの中身は更新ログから読み出したINSERTの内容ですね。面白い。Key formatが万歳アスキーアートになってるが大丈夫か・・・？MySQLでCREATE TABLEした際にKey Columnを宣言しなかったため、Message KeyがNullになっているせいだと予想。

```json:TOPICに書き出された内容
ksql> PRINT 'call-center-db.call-center.calls' FROM BEGINNING;
Key format: ¯\_(ツ)_/¯ - no data processed
Value format: AVRO
rowtime: 2021/11/11 11:05:32.288 Z, key: <null>, value: {"before": null, "after": {"name": "michael", "reason": "purchase", "duration_seconds": 540}, "source": {"version": "1.1.0.Final", "connector": "mysql", "name": "call-center-db", "ts_ms": 0, "snapshot": "true", "db": "call-center", "table": "calls", "server_id": 0, "gtid": null, "file": "mysql-bin.000006", "pos": 195, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1636628731343, "transaction": null}, partition: 0
rowtime: 2021/11/11 11:05:32.291 Z, key: <null>, value: {"before": null, "after": {"name": "michael", "reason": "help", "duration_seconds": 224}, "source": {"version": "1.1.0.Final", "connector": "mysql", "name": "call-center-db", "ts_ms": 0, "snapshot": "true", "db": "call-center", "table": "calls", "server_id": 0, "gtid": null, "file": "mysql-bin.000006", "pos": 195, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1636628731350, "transaction": null}, partition: 0
...
```

Debeziumが登録したAvro Schemaを調べてみると、行の「before image」、「after image」、「更新に関するMetadata」で構成されているようだ。Sink Connectorによる他DBMSへのApplyをガッツリ意識した内容だなぁ。

```json:Debeziumが作ったAvroSchema
{
  "type": "record",
  "name": "Envelope",
  "namespace": "call_center_db.call_center.calls",
  "fields": [
    {
      "name": "before",
      "type": [
        "null",
        {
          "type": "record",
          "name": "Value",
          "fields": [
            {
              "name": "name",
              "type": [
                "null",
                "string"
              ],
              "default": null
            },
            {
              "name": "reason",
              "type": [
                "null",
                "string"
              ],
              "default": null
            },
            {
              "name": "duration_seconds",
              "type": [
                "null",
                "int"
              ],
              "default": null
            }
          ],
          "connect.name": "call_center_db.call_center.calls.Value"
        }
      ],
      "default": null
    },
    {
      "name": "after",
      "type": [
        "null",
        "Value"
      ],
      "default": null
    },
    {
      "name": "source",
      "type": {
        "type": "record",
        "name": "Source",
        "namespace": "io.debezium.connector.mysql",
        "fields": [
          {
            "name": "version",
            "type": "string"
          },
          {
            "name": "connector",
            "type": "string"
          },
          {
            "name": "name",
            "type": "string"
          },
          {
            "name": "ts_ms",
            "type": "long"
          },
          {
            "name": "snapshot",
            "type": [
              {
                "type": "string",
                "connect.version": 1,
                "connect.parameters": {
                  "allowed": "true,last,false"
                },
                "connect.default": "false",
                "connect.name": "io.debezium.data.Enum"
              },
              "null"
            ],
            "default": "false"
          },
          {
            "name": "db",
            "type": "string"
          },
          {
            "name": "table",
            "type": [
              "null",
              "string"
            ],
            "default": null
          },
          {
            "name": "server_id",
            "type": "long"
          },
          {
            "name": "gtid",
            "type": [
              "null",
              "string"
            ],
            "default": null
          },
          {
            "name": "file",
            "type": "string"
          },
          {
            "name": "pos",
            "type": "long"
          },
          {
            "name": "row",
            "type": "int"
          },
          {
            "name": "thread",
            "type": [
              "null",
              "long"
            ],
            "default": null
          },
          {
            "name": "query",
            "type": [
              "null",
              "string"
            ],
            "default": null
          }
        ],
        "connect.name": "io.debezium.connector.mysql.Source"
      }
    },
    {
      "name": "op",
      "type": "string"
    },
    {
      "name": "ts_ms",
      "type": [
        "null",
        "long"
      ],
      "default": null
    },
    {
      "name": "transaction",
      "type": [
        "null",
        {
          "type": "record",
          "name": "ConnectDefault",
          "namespace": "io.confluent.connect.avro",
          "fields": [
            {
              "name": "id",
              "type": "string"
            },
            {
              "name": "total_order",
              "type": "long"
            },
            {
              "name": "data_collection_order",
              "type": "long"
            }
          ]
        }
      ],
      "default": null
    }
  ],
  "connect.name": "call_center_db.call_center.calls.Envelope"
}
```

## Create the ksqlDB calls stream

Streamを作る。Avro Schemaをかぶせておく。

```:CREATE_STREAM
ksql> CREATE STREAM calls WITH (
>    kafka_topic = 'call-center-db.call-center.calls',
>    value_format = 'avro'
>);

 Message
----------------
 Stream created
----------------
```

## Create the materialized views

Materialized views、つまりksqlDBのTableを作る。`->`を用いることでAvro Schemaにおける`After` Recordの中を指定している。このTableでは以下の関数を使用し、NAMEでGroup byしている。
-`count_distinct` : 「コール理由の種類数」を数える
-`latest_by_offset`: 「最後のコール理由」を取り出す

```:support_viewテーブル
ksql> CREATE TABLE support_view AS
>    SELECT after->name AS name,
>           count_distinct(after->reason) AS distinct_reasons,
>           latest_by_offset(after->reason) AS last_reason
>    FROM calls
>    GROUP BY after->name
>    EMIT CHANGES;

 Message
-------------------------------------------
 Created query with ID CTAS_SUPPORT_VIEW_3
-------------------------------------------
```

もう一つ作る。このtableでは以下の関数を使用し、やはりNAMEでGroup byしている。
-`count` : 「コール(理由)数」を数える
-`(sum(after->duration_seconds) / 60)` : コール時間(秒)を集計し分に変換する

```:lifetime_viewテーブル
ksql> CREATE TABLE lifetime_view AS
>    SELECT after->name AS name,
>           count(after->reason) AS total_calls,
>           (sum(after->duration_seconds) / 60) as minutes_engaged
>    FROM calls
>    GROUP BY after->name
>    EMIT CHANGES;

 Message
--------------------------------------------
 Created query with ID CTAS_LIFETIME_VIEW_5
--------------------------------------------
```

さっそくksqlによる検索を実施してみる。

```:support_viewのdrek情報
ksql> SELECT name, distinct_reasons, last_reason
>FROM support_view
>WHERE name = 'derek';
+---------------------------------------------------------+---------------------------------------------------------+---------------------------------------------------------+
|NAME                                                     |DISTINCT_REASONS                                         |LAST_REASON                                              |
+---------------------------------------------------------+---------------------------------------------------------+---------------------------------------------------------+
|derek                                                    |3                                                        |refund                                                   |
Query terminated
```

```:lifetime_viewのderek情報
ksql> SELECT name, total_calls, minutes_engaged
>FROM lifetime_view
>WHERE name = 'derek';
+---------------------------------------------------------+---------------------------------------------------------+---------------------------------------------------------+
|NAME                                                     |TOTAL_CALLS                                              |MINUTES_ENGAGED                                          |
+---------------------------------------------------------+---------------------------------------------------------+---------------------------------------------------------+
|derek                                                    |4                                                        |227                                                      |
Query terminated
```

上記二つのテーブルをJOINしたテーブルを作ってみる。

```:2つのTableのJOIN
ksql> CREATE TABLE support_view_enriched AS
>SELECT support_view.name as name, distinct_reasons, last_reason, total_calls, minutes_engaged
> FROM support_view
> LEFT JOIN lifetime_view
>   on support_view.name = lifetime_view.name
> EMIT CHANGES;

 Message
-----------------------------------------------------
 Created query with ID CTAS_SUPPORT_VIEW_ENRICHED_11
-----------------------------------------------------
```

ksql検索結果は下記の通り。JOIN成功。

```:support_view_enrichedのderek情報
ksql> SELECT name, distinct_reasons, last_reason, total_calls, minutes_engaged FROM support_view_enriched WHERE name='derek';
+---------------------------------+---------------------------------+---------------------------------+---------------------------------+---------------------------------+
|NAME                             |DISTINCT_REASONS                 |LAST_REASON                      |TOTAL_CALLS                      |MINUTES_ENGAGED                  |
+---------------------------------+---------------------------------+---------------------------------+---------------------------------+---------------------------------+
|derek                            |3                                |refund                           |4                                |227                              |
Query terminated
```

## 新たなレコード追加

MySQL CLIからderekのコールに関する新レコードを挿入。彼はモンスタークレーマーなのだ！

```:MySQL_CLIから新レコード追加
mysql> INSERT INTO calls (name, reason, duration_seconds) VALUES ("derek", "complain", 1200);
Query OK, 1 row affected (0.07 sec)
```

ksql検索結果にちゃんと反映されている。debeziumがMySQLログから情報を引っ張ってTopicに書き込み、それをksqlDBがリアルタイム処理していることが分かる。

```:support_view_enrichedのderek情報
ksql> SELECT * FROM support_view_enriched WHERE name='derek';
+---------------------------------+---------------------------------+---------------------------------+---------------------------------+---------------------------------+
|NAME                             |DISTINCT_REASONS                 |LAST_REASON                      |TOTAL_CALLS                      |MINUTES_ENGAGED                  |
+---------------------------------+---------------------------------+---------------------------------+---------------------------------+---------------------------------+
|derek                            |4                                |complain                         |5                                |247                              |
Query terminated
```



# まとめ

- ソースが何であれ、Topic内に格納されているメッセージのAvro Schemaさえ押さえておけばksql処理は可能
- DB更新をソースとする場合、更新出現が完全に業務依存になるためksql処理設計の際は十分注意が必要。今回取り扱ったコールセンター業務のコールログのように単調に追加されるだけであれば扱いは楽だが、行の削除があるDBに対するksql表現はかなりややこしくなりそう。というかうまくサンプルが思いつかない。
- Debeziumはすごい。