Debezium の RHEL へのインストール

前提条件

• ソースデータベースがデプロイされ、Debezium コネクターがデータベースにアクセスできる。

• コネクターがソースデータベースにアクセスするために必要な以下の情報を把握している。

  • データベースホストの名前または IP アドレス
  • データベースに接続するためのポート番号
  • コネクターがデータベースへのサインインに使用できるアカウントの名前
  • データベースユーザーアカウントのパスワード
  • データベースの名前
  • コネクターの情報の取得元であるテーブルの名前
  • コネクターの変更イベントの出力先である Kafka ブローカーの名前
  • コネクターのデータベース履歴情報の送信先である Kafka トピックの名前

手順

• Debezium コネクターに適用する設定を JSON 形式で指定します。
  

  次の例は、Debezium MySQL コネクターの簡単な設定を示しています。

  {
    "name": "inventory-connector",  1
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector", 2
      "tasks.max": "1",  3
      "database.hostname": "mysql",  4
      "database.port": "3306", 5
      "database.user": "debezium", 6
      "database.password": "dbz", 7
      "database.server.id": "184054",  8
      "topic.prefix": "dbserver1",   9
      "table.include.list": "public.inventory",  10
      "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",  11
      "schema.history.internal.kafka.topic": "dbhistory.inventory"  12
    }
  }

  1

  Kafka Connect クラスターに登録するコネクターの名前。

  2

  コネクタークラスの名前。

  3

  同時に動作できるタスクの数。一度に操作できるタスクは 1 つだけです。

  4

  ホストデータベースインスタンスのホスト名または IP アドレス

  5

  データベースインスタンスのポート番号。

  6

  Debezium がデータベースに接続するユーザーアカウントの名前。

  7

  データベースユーザーアカウントのパスワード

  8

  コネクターの一意の数値 ID。
  このプロパティーは MySQL コネクターにのみ使用されます。

  9

  コネクターが変更をキャプチャーするサーバーのデータベースサーバーまたはクラスターの論理識別子として機能する文字列。指定された文字列は名前空間を指定します。Avro コンバーターが使用されている場合、Debezium は、コネクターが書き込む各 Kafka トピック、Kafka Connect スキーマの名前、および対応する Avro スキーマの名前空間にこの名前の接頭辞を付けます。

  10

  コネクターが変更イベントをキャプチャーするテーブルのリスト。

  11

  コネクターがデータベーススキーマ履歴を送信する Kafka ブローカーの名前。指定されたブローカーは、コネクターが出力する変更イベントも受け取ります。

  12

  スキーマ履歴を格納する Kafka トピックの名前。
  コネクターの再起動後、コネクターは停止した時点からデータベースログの読み取りを再開し、オフライン中に発生したトランザクションのイベントを出力します。コネクターは、未読トランザクションの変更イベントを Kafka に書き込む前に、スキーマ履歴をチェックしてから、元のトランザクションが発生したときに有効だったスキーマを適用します。

Additional information

• コネクターのタイプごとに設定できる設定プロパティーの詳細は、Debezium ユーザーガイド のコネクターのデプロイメントに関するセクションを参照してください。

前提条件

• Debezium のデプロイ先であるホスト環境が、サポートされている設定 で Red Hat Enterprise Linux、AMQ Streams、および Java を実行している。
  

  • AMQ Streams のインストール方法の詳細は、AMQ Streams のインストール を参照してください。
  • 単一の ZooKeeper ノードおよび単一の Kafka ノードが含まれる実稼働ではない、基本的な AMQ Streams クラスターをインストールする方法の詳細は、単一ノードの AMQ Streams クラスターの実行 を参照してください。

• ホストの管理者権限 (sudo アクセス) がある。
• Apache ZooKeeper と Apache Kafka ブローカーが実行している。
• Kafka Connect が、スタンドアロンモードではなく、分散モード で実行されている。
• AMQ Streams のインストール時に作成された kafka ユーザーの認証情報を把握している。
• ソースデータベースがデプロイされ、Debezium をデプロイするホストがデータベースにアクセスできる。
• コネクターの設定方法 を理解している。

関連情報

• コネクタープラグインの追加

前提条件

• 「Red Hat Enterprise Linux での AMQ Streams を使用した Debezium のデプロイ」 の手順に基づいて、Red Hat Enterprise Linux にコネクターをデプロイしています。以下の手順を実行します。

  1. ホストのターミナルウィンドウで、次のコマンドを入力して、Kafka Connect API からコネクターのリストを要求します。

     $ curl -H "Accept:application/json" localhost:8083/connectors/

     クエリーは、デプロイされたコネクターの名前を返します。以下に例を示します。

     ["inventory-connector"]

  2. ホストのターミナルウィンドウで次のコマンドを入力して、コネクターが実行しているタスクを表示します。

     $ curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/inventory-connector

     コマンドは、以下の例のような出力を返します。

     HTTP/1.1 200 OK
     Date: Thu, 06 Feb 2020 22:12:03 GMT
     Content-Type: application/json
     Content-Length: 531
     Server: Jetty(9.4.20.v20190813)
     
     {
       "name": "inventory-connector",
       ...
       "tasks": [
         {
           "connector": "inventory-connector",
           "task": 0
         }
       ]
     }

  3. Kafka クラスター内のトピックのリストを表示します。
     ターミナルウィンドウから /opt/kafka/bin/ に移動し、以下のシェルスクリプトを実行します。

     ./kafka-topics.sh --bootstrap-server=localhost:9092 --list

     Kafka ブローカーは、コネクターが作成するトピックのリストを返します。使用可能なトピックは、コネクターの snapshot.mode、snapshot.include.collection.list、および table.include.list の設定プロパティーの設定によって異なります。デフォルトでは、コネクターはデータベース内の非システムテーブルごとにトピックを作成します。

  4. トピックの内容を表示します。
     ターミナルウィンドウから /opt/kafka/bin/ に移動し、kafka-console-consumer.sh シェルスクリプトを実行して、前のコマンドで返されたトピックの 1 つの内容を表示します。
     

     以下に例を示します。

     ./kafka-console-consumer.sh \
     >     --bootstrap-server localhost:9092 \
     >     --from-beginning \
     >     --property print.key=true \
     >     --topic=dbserver1.inventory.products_on_hand

     トピックの各イベントについて、このコマンドは、以下の出力のような情報を返します。

     例2.1 Debezium 変更イベントの内容

     {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Key"},"payload":{"product_id":101}}	{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"2.5.4.Final-redhat-00001","connector":"mysql","name":"inventory_connector_mysql","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1638985247805,"transaction":null}}

     上記の例では、payload 値は、コネクタースナップショットがテーブル inventory.products_on_hand から読み込み ("op" ="r") イベントを生成したことを示しています。product_id レコードの "before" 状態は null であり、レコードに以前の値が存在しないことを示しています。"after" 状態は、product_id 101 を持つ項目の quantity が 3 であることを示しています。