NPSaaS および Kafka

概要

Apache Kafka は、一般的なアプリケーション間でデータを移動するために使用できるパブリッシュ/サブスクライブ・メッセージング・システムです。

Kafka JDBC コネクターを介して IBM® Netezza® Performance Server for IBM Cloud Pak® for Data as a Service インスタンスを Kafka と統合した後、以下のいずれかとして NPSaaS を使用できます。

データを Kafkaに取り込むデータ・ソース。
Kafkaからデータを読み取るデータ・シンク。

データソースとしてNPSaaSを使用する

e-コマース企業は、商品リストを NPSaaS データベースに保管します。アプリ内検索エクスペリエンスを簡素化し、リアルタイム分析にアクセスするために、コンシューマー・アプリ (例えば、 Elasticsearch および Apache Flink) はリストにアクセスできます。

この場合、データは Kafka JDBC ソース・コネクターを介して NPSaaS から読み取られ、 Kafka がデータをストリーミングします。コンシューマー・アプリケーションはストリームから読み取り、データをさらに処理します。

次の図は、データ・ソースとしてのデータ・フロー NPSaaS を示しています。

データ・ソースとしてのNPSaaS — イメージ 1。この図は、 Kafka が JDBC ソース・コネクターを介して Netezza からデータを読み取り、コンシューマー・アプリケーションがそのデータにアクセスできるようにする方法を示しています。

NPSaaS をデータ・シンクとして使用する

患者の成果を向上させ、リスク要因を効率的に特定し、介入時間を短縮するために、病院は生理学的データから意味のある洞察を抽出します。さまざまなチャネルからのさまざまなデータ・セットが、到着時に分析されます。

この場合、着信データは Kafka を介してストリーミングされてから、計算されます。これらの生産者は、さまざまなチャネルからの生理的データのソースです。

データは処理された後、 Kafka JDBC シンク・コネクターを介して患者履歴の記録のために NPSaaS に保管されます。

次の図は、データ・シンクとしての NPSaaS のデータ・フローを示しています。

NPSaaS(データ・シンクとして) — イメージ 2。この図は、さまざまなプロデューサーからの着信データがどのようにストリーミングされ、 JDBC ドライバーを介して Kafka によって計算され、 Netezzaに保管されるかを示しています。

NPSaaS と Kafka の統合

NPSaaS インスタンスを Kafkaと統合する場合は、 Kafka JDBC コネクターを使用する必要があります。

Kafka JDBC コネクターは、ソースおよびシンクの JDBC コネクターをサポートします。ソース・コネクターを使用すると、 NPSaaS から Kafka トピックにデータを転送できます。シンク・コネクターを使用すると、 Kafka JDBC コネクターを使用して、 Kafka トピックから NPSaaSにデータを転送できます。

JDBC Kafka コネクターのセットアップ

plugin.pathを編集して、 Kafkaのライブラリーにドライバーをインストールする必要があります。

Javaをセットアップします。

Kafka を機能させるには、 Java 8 以降が必要です。 JDBC コネクターの場合は、 Java 11 が必要です。
```
sudo yum install -y java-11-openjdk-headless
```
Netezza JDBC ドライバーを Kafka libs にコピーします。
```
cp path/to/nzjdbc3.jar kafka/libs
```
コネクターをセットアップします。

この例では、Aiven コネクターが使用されています。

a) コネクターをダウンロードします。
```
curl -SLO https://github.com/aiven/jdbc-connector-for-apache-kafka/releases/download/v6.6.2/jdbc-connector-for-apache-kafka-6.6.2.tar
```
b) パッケージを解凍します。
```
tar xvf jdbc-connector-for-apache-kafka-6.6.2.tar
```
c) Kafka config/connect-[distributed/standalone].properties 内の plugin.path を編集して、解凍したフォルダーを指すようにします。これにより、 Kafka はプラグインとコネクター jar を検出してロードできます。

分散接続を開始します。

connect-distributed.sh config/connect-distributed.properties

出力:

[2022-06-29 02:29:43,356] INFO Started o.e.j.s.ServletContextHandler@5562c2c9{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler:915)
[2022-06-29 02:29:43,356] INFO REST resources initialized; server is started and ready to handle requests (org.apache.kafka.connect.runtime.rest.RestServer:303)
[2022-06-29 02:29:43,356] INFO Kafka Connect started (org.apache.kafka.connect.runtime.Connect:57)

コネクターを登録します。

コネクターの登録を試行する前に、データベースと表が NPSaaS に存在している必要があります。

ソース・コネクタについては、以下のコマンドを実行する。

curl -s -X POST -H "Content-Type: Application/json" --data '{ "name": "test-source-jdbc", "config": { "connector.class": "io.aiven.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url":"jdbc:netezza://localhost:5480/db1;user=user1;password=secret","connection.user":"user1","connection.password":"secret","dialect.name":"GenericDatabaseDialect","topic.prefix":"my","mode": "bulk", "poll.interval.ms":"60000", "table.whitelist":"TEST_TABLE", "batch.max.rows":"10000000" } }' http://localhost:8083/connectors | jq
{
  "name": "test-source-jdbc",
  "config": {
    "connector.class": "io.aiven.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:netezza://localhost:5480/db1;user=user1;password=secret,
    "connection.user": ""user1,
    "connection.password": "secret",
    "dialect.name": "GenericDatabaseDialect",
    "topic.prefix": "my",
    "mode": "bulk",
    "poll.interval.ms": "60000",
    "table.whitelist": "TEST_TABLE",
    "batch.max.rows": "10000000",
    "name": "test-source-jdbc"
  },
  "tasks": [],
  "type": "source"
}

シンク・コネクターの場合は、以下のコマンドを実行します。

curl -s -X POST -H "Content-Type: Application/json" --data '{ "name": "test-sink", "config": { "connector.class": "io.aiven.connect.jdbc.JdbcSinkConnector", "tasks.max": "2", "connection.url":"jdbc:netezza://localhost:5480/db1;user=user1;password=secret","connection.user":"user1","connection.password":"secret","dialect.name":"GenericDatabaseDialect","topics": "TEST_TABLE", "insert.mode": "insert" } }' http://localhost:8083/connectors | jq
{
  "name": "test-sink",
  "config": {
    "connector.class": "io.aiven.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "2",
    "connection.url": "jdbc:netezza://localhost:5480/db1;user=user1;password=secret",
    "connection.user": "user1",
    "connection.password": "secret",
    "dialect.name": "GenericDatabaseDialect",
    "topics": "TEST_TABLE",
    "insert.mode": "insert",
    "name": "test-sink"
  },
    "tasks": [],
    "type": "sink"
  }

登録が成功したかどうかを確認する。

a) コネクターをリストします。

curl -s http://localhost:8083/connectors/ | jq
[
  "test-source-jdbc",
  "test-sink"
]

b) コネクターの状況を確認します。

curl -s http://localhost:8083/connectors/test-source-jdbc/status | jq
{
     "name": "test-source-jdbc",
     "connector": {
         "state": "RUNNING",
         "worker_id": "10.11.112.15:8083"
  },
    "tasks": [
      {
          "id": 0,
          "state": "RUNNING",
          "worker_id": "10.11.112.15:8083"
       }
     ],
     "type": "source"
}

NPSaaS ログを調べて、コネクターが正しく機能するかどうかを確認できます。

ソース・コネクターの場合:

2022-06-29 04:13:09.057046 PDT [27743]  DEBUG:  QUERY: SELECT 1
2022-06-29 04:13:09.061443 PDT [27743]  DEBUG:  QUERY: SELECT * FROM "DB1"."USER1"."TEST_TABLE"
ANALYZE
2022-06-29 04:13:09.064209 PDT [27743]  DEBUG:  QUERY: SELECT * FROM "DB1"."USER1"."TEST_TABLE"

シンク・コネクターの場合:

2022-06-29 09:08:32.379442 PDT [18976]  DEBUG:  QUERY: CREATE EXTERNAL TABLE bulkETL_18976_0 ( c0 nvarchar(8),c1 nvarchar(10),c2 nvarchar(4),c3 nvarchar(6),c4 nvarchar(7),c5 nvarchar(9),c6 nvarchar(7),c7 nvarchar(9),c8 nvarchar(12),c9 nvarchar(12),c10 nvarchar(25),c11 nvarchar(13),c12 nvarchar(1) )  USING (  DATAOBJECT('/tmp/junk')  REMOTESOURCE 'jdbc'  DELIMITER ' '  ESCAPECHAR '\'  CTRLCHARS 'YES'  CRINSTRING 'YES'  ENCODING 'INTERNAL'  MAXERRORS 1 QUOTEDVALUE 'YES' );
2022-06-29 09:08:32.525024 PDT [18976]  DEBUG:  QUERY: INSERT INTO "TEST_TABLE"("C1","C2","C3","C4","C5","C6","C7","C8","DATE_PROD","TIME_PROD","TIMESTMP","TIMETZ_PROD","C18") VALUES(bulkETL_18976_0.c0,bulkETL_18976_0.c1,bulkETL_18976_0.c2,bulkETL_18976_0.c3,bulkETL_18976_0.c4,bulkETL_18976_0.c5,bulkETL_18976_0.c6,bulkETL_18976_0.c7,bulkETL_18976_0.c8,bulkETL_18976_0.c9,bulkETL_18976_0.c10,bulkETL_18976_0.c11,bulkETL_18976_0.c12)