Kubernetes Service を使用して IBM Cloud Object Storage への Event Streams の接続

KafkaConnectランタイムをKubernetes Serviceクラスタで実行する。 次に、IBM Cloud® Object Storageを起動する。Sink Connectorを起動し、Event Streamsの KafkaトピックからIBM Cloud® Object Storageサービスのインスタンスにデータをアーカイブする。

コネクターは、Kafka からメッセージのバッチをコンシュームし、メッセージ・データをオブジェクトとして Cloud Object Storage サービスのバケットにアップロードします。

ステップ 1. 前提条件のインストール

以下のソフトウェアおよびサービスがインストールされていることを確認します。

  • Event Streams インスタンス - 標準プランまたはエンタープライズ・プラン。 資格情報を作成する必要があります。

  • 少なくとも 1 つのバケットを持つ Cloud Object Storage サービスのインスタンスです。

  • IBM Cloud® Kubernetes Service クラスター。 テストを目的として無料でプロビジョンできます。

    クラスターへの CLI アクセスも必要です。 詳しくは、CLI および API のセットアップ (Setting up the CLI and API) を参照してください。

  • Kubectl の最新バージョンです。

  • Git

ステップ 2 kafka-connect リポジトリーの複製

必要なファイルを含んでいる以下の 2 つのリポジトリーを複製します。

ステップ 3。 Kafka Connect 構成の作成

  1. この構成をセットアップする必要があるのは 1 回のみです。Event Streams は、将来の使用のために保管します。

    event-streams-samplesプロジェクトから、kafka-connect/IKS directory、 connect-distributed.properties を編集し <BOOTSTRAP_SERVERS> 一か所だけ置き換えます <APIKEY> 3 つの場所で Event Streams 資格情報に置き換えます。

    <BOOTSTRAP_SERVERS> をコンマ区切りリストとして指定します。 有効でない場合は、エラーになります。

    <APIKEY> はマシン上に平文で表示されますが、IBM Cloud® Kubernetes Service にプッシュされると秘密情報になります。

    Kafka Connect では、信頼性とスケーラビリティーの理由から、複数のワーカーを実行できます。 ご使用の Kubernetes Service クラスターに複数のノードがあり、複数の Connect ワーカーが必要な場合は、kafka-connect.yamlファイルを編集し、エントリー replicas: 1を編集します。

  2. 次に、以下のコマンドを実行します。

    以下のコマンドを実行して、シークレットを作成します。

    kubectl create secret generic connect-distributed-config --from-file=connect-distributed.properties

    以下のコマンドを実行して、configmap を作成します。

    kubectl create configmap connect-log4j-config --from-file=connect-log4j.properties

ステップ 4: Kafka Connect のデプロイ

以下のコマンドを実行して、kafka-connect.yamlファイル内の構成を適用します。

kubectl apply -f ./kafka-connect.yaml

ステップ 5. Kafka Connect が実行中であることの検証

Kafka Connect が実行されていることを検証するには、以下の例のように、ポート 8083 上の kafkaconnect - service にポートを転送します。

kubectl port-forward service/kafkaconnect-service 8083

ポート転送に使用した端末を開いたままにし、次のステップに別の端末を使用します。

Connect REST API がhttp://localhost:8083で使用可能になります。 APIの詳細については、以下を参照してください。 KafkaConnect REST Interfaceをご覧ください。

これで、Kubernetes Service にデプロイされ、実行されている Kafka Connect ランタイムが作成されました。 次に、Object Storageコネクタを構成し、起動する。

ステップ 6. cos-sink JSON ファイルの構成

kafka-connect-ibmcos-sink/config/にあるcos-sink.jsonファイルを編集して、少なくとも必要なプロパティーが情報で完成するようにします。 設定プロパティcos.object.deadline.seconds、cos.interval.seconds、cos.object.recordsはオプションとしてリストされていますが、これらのプロパティの少なくとも1つをデフォルト以外の値に設定する必要があります。

cos-sink.json ファイルのプロパティー

cos-sink.jsonファイル内のプレースホルダーを独自の値に置き換えます。

cos-sink.json ファイルのプロパティー
ファイル・プロパティー 説明
cos.api.key 必須。 Cloud Object Storage サービス・インスタンスへの接続に使用される API キーです。
cos.bucket.location 必須。 Cloud Object Storage サービス・バケットの場所です。 例えば、地域バケット eu-gbの場合、または グローバル・バケット euの場合です。
cos.bucket.name 必須。 データを書き込むCloud Object Storage サービス・バケットの名前です。
cos.bucket.resiliency 必須。 Cloud Object Storage バケットの回復力です。 cross-region、regional、または single-site のいずれかでなければなりません。
cos.service.crn 必須。 Cloud Object Storage サービス・インスタンスの CRN です。 正しい CRN を入力したことを確認してください。これは、二重コロンで終わるリソース・インスタンス ID です (例: crn:v1:staging:public:cloud-object-storage:global:a/8c226dc8c8bfb9bc3431515a16957954:b25fe12c-9cf5-4ee8-8285-2c7e6ae707f6::)。
cos.endpoint.visibility オプション。 パブリック・インターネットを介してクラウド Object Storage サービスに接続するには、 public を指定します。 例えば、IBM Cloud Kubernetes Serviceクラスターなど、IBM Cloudネットワーク内で動作するコネクターから接続する場合はprivateを指定します。 デフォルトはプライベートです。
cos.object.deadline.seconds オプション。 Kafka からの最初のレコードを読み取ってから、これまでに読み取ったすべてのレコードを Cloud Object Storage オブジェクトに書き込むまでの秒数 (接続タスク・インスタンスの測定時間)です。 このプロパティは、Kafkaレコードがトピックに生成される間に長い休止時間がある状況で役立ちます。 これにより、このコネクターによって受信されたすべてのレコードが、指定された時間内に必ず Object Storage に書き込まれるようになります。
cos.object.interval.seconds オプション。 Kafka からの最初のレコードを読み取ってから、これまでに読み取ったすべてのレコードを Cloud Object Storage オブジェクトに書き込むまでの秒数 (Kafka レコードのタイム・スタンプで測定)です。
cos.object.records オプション。 1 つのオブジェクトに結合する Kafka レコードの最大数です。

IBM Cloud コンソールを使用する Object Storage 資格情報の取得

  1. ダッシュボードで Object Storage サービスを見つけます。
  2. サービス・タイルをクリックします。
  3. **「サービス資格情報」**をクリックします。
  4. **「新規資格情報 (New Credential)」**をクリックします。
  5. 名前や役割など、新規資格情報の詳細を入力し、**「追加」**をクリックします。 新規資格情報が資格情報リストに表示されます。
  6. **「資格情報の表示」**を使用してこの資格情報をクリックし、JSON 形式の詳細を表示します。

手順 7. コネクターをその構成で開始します

以下のコマンドを実行して、前のステップで指定した構成で Object Storage コネクターを開始します。

curl -X POST -H "Content-Type: application/json" http://localhost:8083/connectors --data "@./cos-sink.json"

ステップ 8. コネクターのモニター

コネクターは以下の場所で確認できます。

http://localhost:8083/connectors/cos-sink/status

コネクターの状態が未実行の場合は、コネクターを再始動してください。

ステップ 9。 コネクターの削除

以下のコマンドを使用して、コネクターを削除できます。

curl -X DELETE
http://localhost:8083/connectors/cos-sink