Event Streams での ksqlDB の使用

KSQL を Event Streams とともに使用して、ストリーム処理を行うことができます。

ksqlDB は、イベント・ストリーミング用の専用データベースです。 Apache Kafka用の専用ストリーム処理データベースを使用して、エンドツーエンドのイベント・ストリーミング・アプリケーションを迅速に作成するために使用します。

Event Streams で ksqlDB を最も迅速かつ簡単に実行する方法は、ksqlDB Quickstart で説明されているように Docker コンテナーを使用する方法です。

ksqlDB は、無制限の retention.ms 設定でトピックを作成する必要があるため、エンタープライズ・プランでは ksqlDB のみを使用できます。

セットアップするには、以下のステップを実行します。完了すると、「 ksqlDB クイック・スタート・ガイド」にリストされているすべての照会を実行できます。

IBM Cloud コンソールの Event Streams ダッシュボードを使用して、単一のパーティションとデフォルトの保存期間を持つ _confluent-ksql-default__command_topic というトピックを作成します。

Docker コンテナーで ksqlDB を実行する場合は、以下のステップを実行します。

a. 以下の内容の docker-compose.yml ファイルを作成します。

---
version: '2'

services:
  ksqldb-server:
    image: confluentinc/ksqldb-server:latest
    hostname: ksqldb-server
    container_name: ksqldb-server
    ports:
      - "8088:8088"
    environment:
      KSQL_LISTENERS: http://0.0.0.0:8088
      KSQL_BOOTSTRAP_SERVERS: <BOOTSTRAP_SERVERS>
      KSQL_KSQL_SERVICE_ID: default_
      KSQL_KSQL_SINK_REPLICAS: 3
      KSQL_KSQL_STREAMS_REPLICATION_FACTOR: 3
      KSQL_SECURITY_PROTOCOL: SASL_SSL
      KSQL_KSQL_INTERNAL_TOPIC_REPLICAS: 3
      KSQL_SASL_JAAS_CONFIG: org.apache.kafka.common.security.plain.PlainLoginModule required username=\"token\" password=\"<PASSWORD>\";
      KSQL_SASL_MECHANISM: PLAIN

  ksqldb-cli:
    image: confluentinc/ksqldb-cli:latest
    container_name: ksqldb-cli
    depends_on:
      - ksqldb-server
    entrypoint: /bin/sh
    tty: true

ここで、BOOTSTRAP_SERVERS および PASSWORD は、Event Streams の IBM Cloud **「サービス資格情報」**タブからの値です。

b. ファイルを作成したディレクトリーから、以下のコマンドを実行して ksqlDB サーバーを始動します。

docker-compose up

c. 以下のコマンドを実行して ksqlDB サーバーに接続し、対話式のコマンド・ライン・インターフェース (CLI) セッションを開始します。

docker exec -it ksqldb-cli ksql http://ksqldb-server:8088

Docker コンテナーで ksqlDB を実行していない場合は、以下のステップを実行します。

a. ksqlDB リポジトリーを複製します。
```
git clone https://github.com/confluentinc/ksql
```
b. ksqlDB ディレクトリーに移動して、コードをコンパイルします。
```
cd ksql
mvn clean install -Dmaven.test.skip=true
```
c. Event Streams ksqlDB 構成ファイルを作成します。

以下に例を示します。
```
bootstrap.servers=BOOTSTRAP_SERVERS
listeners=http://localhost:8088
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
ssl.protocol=TLSv1.2
ssl.enabled.protocols=TLSv1.2
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="token" password="PASSWORD";
ksql.sink.replications.default=3
```
ここで、BOOTSTRAP_SERVERS および PASSWORD は、Event Streams の IBM Cloud **「サービス資格情報」**タブからの値です。

d. 以下のコマンドを使用して、 ksqlDB サーバーを始動します。
```
./bin/ksql-server-start ./config/ibm-eventstreams.properties
```
e. 以下のコマンドを使用して、 ksqlDB CLI を開始します。
```
./bin/ksql
```

これらのステップを完了すると、「 ksqlDB クイック・スタート・ガイド」にリストされているすべての照会を実行できます。