Scala: Apache Kafka® のサンプルコード

このチュートリアルでは、Apache Kafka® クラスターにメッセージを生成し、このクラスターからメッセージを消費する Scala クライアントアプリケーションを実行します。

チュートリアルの実行後に、提供された ソースコード をリファレンスとして使用して、独自の Kafka クライアントアプリケーションを開発します。

前提条件

クライアント

  • Scala sbt

Kafka クラスター

このチュートリアルは、Confluent Cloud を使用して取り組むのが最も簡単です。そうすればローカル Kafka クラスターを実行する必要はありません。Confluent Cloud にサインアップするときは、プロモーションコード C50INTEG を適用すると、$50 相当を無料で使用できます(詳細)。Console から LEARN をクリックしてクラスターをプロビジョニングし、Clients をクリックして、クライアントアプリケーションに対して設定するクラスター固有の構成と認証情報を取得します。サポートされている CLIREST API、またはコミュニティでサポートされている Confluent Cloud 向け ccloud-stack ユーティリティ を使用してもかまいません。

Confluent Cloud の使用を希望しない場合は、ローカルホスト や他のリモートサーバーで実行されている Kafka クラスターでこのチュートリアルに取り組むこともできます。

セットアップ

  1. confluentinc/examples GitHub リポジトリのクローンを作成し、6.2.4-post ブランチをチェックアウトします。

    git clone https://github.com/confluentinc/examples
    cd examples
    git checkout 6.2.4-post
    
  2. Scala のサンプルのディレクトリに変更します。

    cd clients/cloud/scala/
    
  3. Kafka クラスターに接続するための構成パラメーターを含むローカルファイル($HOME/.confluent/java.config など)を作成します。以下のテンプレートのいずれかをベースとして、クラスターへの接続情報でこのファイルをカスタマイズします。{{ BROKER_ENDPOINT }}{{CLUSTER_API_KEY }}、および {{ CLUSTER_API_SECRET }} に値を代入します(この値を手作業で見つける方法については、「Configure Confluent Cloud Clients」を参照してください。 または、Confluent Cloud 向け ccloud-stack ユーティリティ を使用して値を自動作成します)。

    • Confluent Cloud のテンプレート構成ファイル

      # Required connection configs for Kafka producer, consumer, and admin
      bootstrap.servers={{ BROKER_ENDPOINT }}
      security.protocol=SASL_SSL
      sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}';
      sasl.mechanism=PLAIN
      # Required for correctness in Apache Kafka clients prior to 2.6
      client.dns.lookup=use_all_dns_ips
      
      # Best practice for Kafka producer to prevent data loss 
      acks=all
      
    • ローカルホストのテンプレート構成ファイル

      # Kafka
      bootstrap.servers=localhost:9092
      

基本プロデューサーおよびコンシューマー

このサンプルでは、プロデューサーアプリケーションが Kafka クラスターのトピックに Kafka データを書き込みます。Kafka クラスターにトピックがまだ存在しない場合、プロデューサーアプリケーションは Kafka Admin Client API を使用してトピックを作成します。Kafka に書き込まれる各レコードには、ユーザー名(alice など)を表すキーと、json フォーマットのカウント値({"count": 0} など)があります。コンシューマーアプリケーションは、同じ Kafka トピックを読み取り、各レコードの処理中にカウントのローリング合計を保持します。

レコードの生成

  1. Scala コードをコンパイルします。

    sbt clean compile
    
  2. プロデューサーを実行して、次の引数を渡します。

    • Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
    • トピック名
    sbt "runMain io.confluent.examples.clients.scala.Producer $HOME/.confluent/java.config test1"
    
  3. プロデューサーがすべてのメッセージを送信したことを確認します。次のように表示されます。

    <snipped>
    Produced record at test1-0@120
    Produced record at test1-0@121
    Produced record at test1-0@122
    Produced record at test1-0@123
    Produced record at test1-0@124
    Produced record at test1-0@125
    Produced record at test1-0@126
    Produced record at test1-0@127
    Produced record at test1-0@128
    Produced record at test1-0@129
    Wrote ten records to test1
    [success] Total time: 6 s, completed 10-Dec-2018 16:50:13
    
  4. プロデューサーコード を表示します。

レコードの消費

  1. コンシューマーを実行して、次の引数を渡します。

    • Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
    • 前の手順で使用したトピック名
    sbt "runMain io.confluent.examples.clients.scala.Consumer $HOME/.confluent/java.config test1"
    
  2. コンシューマーがすべてのメッセージを受信したことを確認します。次のように表示されます。

    ...
    Consumed record with key alice and value {"count":1}, and updated total count to 1
    Consumed record with key alice and value {"count":2}, and updated total count to 3
    Consumed record with key alice and value {"count":3}, and updated total count to 6
    Consumed record with key alice and value {"count":4}, and updated total count to 10
    Consumed record with key alice and value {"count":5}, and updated total count to 15
    Consumed record with key alice and value {"count":6}, and updated total count to 21
    Consumed record with key alice and value {"count":7}, and updated total count to 28
    Consumed record with key alice and value {"count":8}, and updated total count to 36
    Consumed record with key alice and value {"count":9}, and updated total count to 45
    Consumed record with key alice and value {"count":10}, and updated total count to 55
    ...
    
  3. 入力し終わったら、Ctrl + C キーを押します("+" はキーを同時に押すことを意味します)。

  4. コンシューマーコード を表示します。

Kafka Streams

  1. Kafka Streams アプリケーションを実行して、次の引数を渡します。

    • Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
    • 前の手順で使用したトピック名
    sbt "runMain io.confluent.examples.clients.scala.Streams $HOME/.confluent/java.config test1"
    
  2. ストリームアプリで、次のように出力されることを確認します。

    ...
    [Consumed record]: alice, 1
    [Consumed record]: alice, 2
    [Consumed record]: alice, 3
    [Consumed record]: alice, 4
    [Consumed record]: alice, 5
    [Consumed record]: alice, 6
    [Consumed record]: alice, 7
    [Consumed record]: alice, 8
    [Consumed record]: alice, 9
    [Consumed record]: alice, 10
    [Running count]: alice, 1
    [Running count]: alice, 3
    [Running count]: alice, 6
    [Running count]: alice, 10
    [Running count]: alice, 15
    [Running count]: alice, 21
    [Running count]: alice, 28
    [Running count]: alice, 36
    [Running count]: alice, 45
    [Running count]: alice, 55
    ...
    
  3. 入力し終わったら、Ctrl + C キーを押します("+" はキーを同時に押すことを意味します)。

  4. View the Kafka Streams code.

おすすめのリソース