Apache Kafkaからデータをプラットフォームにストリーミングする#

#目的

このチュートリアルでは、Apache Kafkaブローカーのデータをプラットフォームで使用する方法を説明します。 最初のセクションでは、Kafkaサーバーにテストデータを作成する方法を説明します。ブローカーに既にメッセージがある場合は、このステップをスキップできます。

#はじめに

#必要条件

このチュートリアルを実行するには、Kafkaブローカーが動作している必要があります。提供されたサンプルコードは、特定の認証を必要としないサーバー（すなわち、IPアドレスを持つ誰でもメッセージを読み取れる）用に書かれています。Kafkaサーバーの認証が異なるように設定されている場合は、ここで使用されているコードをその認証に合わせて変更する必要があります。

さらに、少なくとも最初のGetting Startedチュートリアルを完了していることをお勧めします。ここでは、Data Platformの使用に慣れており、プラットフォームの主要コンポーネントを理解していることを前提としています。

#コンセプトの概要

Data Platformは、ConnectorsのKafka Connectorを介してApache Kafkaと統合されています。このコネクタを使用すると、同じサーバーの1つ以上のトピックからData Platformにデータを取得できます。

通常、データはLakehouse Managerテーブルに取り込まれます。1つのテーブルは、1つのトピックのデータを格納し、トピック内の各メッセージが対応するテーブルのデータの行になります。

Kafkaトピックへの接続をConnectorsで設定し、Lakehouse Managerテーブルを設定した後、Data Processing Engineを使用してLoad actionを実行することで、トピックからテーブルへのデータを読み込む必要があります。

データを読み込んだ後、アクションが実行されている限り、メッセージは自動的にLakehouse Managerテーブルに読み込まれます。アクションはデフォルトでServerless実行モードで実行されることに注意してください。そのため、Kafka Connectorを使用している場合は、Always-up実行モードを使用することをお勧めします。

これで、実際にどのように動作するかを見てみましょう！

#テストデータの設定（オプション）

テスト目的でメッセージをKafkaブローカーに送信するには、Data PlatformでDPEカスタムアクションを使用してプロデューサーを設定できます。Data Processing Engineでカスタムアクションを作成し、ボイラープレートから開始を選択し、以下のボイラープレートコードを次のコードに置き換えます（簡単な名前当てゲーム）：

from forepaas.dwh.connect import connect
import logging, time, json, random
from kafka import KafkaProducer # kafka-python

logger = logging.getLogger(__name__)

TOPIC = "sample"
#   The next two lines MUST be replaced by your own Kafka address and port
KAFKA_PORT="9092"
KAFKA_ADDRESS=["10.152.1.186","10.152.1.187","10.152.7.65"]

def generate_bootstrap():
    servers = [f"{x}:{KAFKA_PORT}" for x in KAFKA_ADDRESS]
    return ",".join(servers)

def customfunc(event):
    logger.info("Begin function customfunc")
    i = 0

    #  Name guessing game: 
    #   - Win 5 to 15 points for guessing the correct name
    #   - Lose 5 to 15 points for guessing the bad name
    #   - No points for guessing the other names

    # generates list of names
    nameslist = ["helene", "francoise", "lea", "lorene", "claire", "lise", "karen", "elise", "elia", "annabele"]

    # select a "bad" and "correct" name
    correct_name = nameslist[0]
    bad_name = nameslist[1]
    logger.info("Correct name:" + correct_name)
    logger.info("Bad name:" + bad_name)

    try:
        producer = KafkaProducer(bootstrap_servers=generate_bootstrap())
        while True:
            i +=1

            name = random.choice(nameslist)
            # distinguish the bad and good names, against the rest
            if name == correct_name:
                points = random.randint(5,15)
            elif name == bad_name:
                points = random.randint(-15,-5)
            else:
                points = 0

            value = {
                "index":i,
                "points":points,
                "name":name,
                }

            # Kafka connection, and publish to broker

            producer.send(TOPIC, json.dumps(value).encode("utf-8"))
            if i % 1000 == 0:
                logger.info("SENT 1000 records")

                # Sleep for 1 second every 1000 messages
                time.sleep(1)
            if i ==10000:
                return
        logger.info("END function customfunc")
    except Exception as err:
        logger.critical(err)

上記のコードは名前当てゲームを表しており、単に推測を表すメッセージを送信します。 各メッセージにはインデックス、名前（推測）、推測のポイントが含まれます。10,000のメッセージが送信されると、アクションは停止し、ブローカーにいくつかのメッセージがあるはずです。

次に、アクションを実行してテストデータでトピックをポップレートし、数千件のレコードが送信されたらアクションの実行を停止します。

#KafkaサーバーをData Platformに接続する

#Kafka接続の設定

最初に行う必要があるのは、Kafkaサーバーへの接続を設定し、データを読み取るトピックを選択することです。必要に応じて、専用のApache Kafka Connector記事を参照してください。

#スキーマの設定

接続とトピックが適切に設定されたら、メッセージにアクセスする準備が整いました。そのためには、Analyzerタブに移動し、接続のトピックのメタデータを抽出する必要があります。

接続は左サイドバーに表示され、接続をクリックするとトピックが表示されます。トピックを選択し、メタデータを抽出ボタンをクリックします。

メタデータの抽出後、メッセージはプレビューパネルに表示され、各行はメッセージに対応します。メタデータパネルのチェックボックスを確認して、メッセージをData Platformで使用する際にどのフィールドが含まれるかを設定します。

#テーブルの作成とビルド

データをData Platformに読み込む前に、そのデータを格納するテーブルを作成し、ビルドする必要があります。これらの概念に慣れていない場合は、Tables記事を参照してください。おそらく新しいテーブルの作成とすべてのテーブルのビルドのセクションを読みたいと思います。

#Data Platformへのデータの読み込み

#ロードアクションの設定

他のコネクタと比較して、Apache Kafka のようなストリーミングソースを使用して Load Action を作成する際には、いくつかの違いがあります。

はじめに、アクションを設定する際に、トピックに関連するテーブルをソースとして選択してください（テーブルを作成する際にアクションを生成している場合は、テーブルを選択する必要はありません）。

#実行モード

Kafkaソースに接続されたLoad Actionを実行する際には、Always-up実行モードを選択することを強くお勧めします。

実行モードに関係なく、アクションはデータが到着するまで実行されます。データが到着すると、Data Platformに読み込まれます。

サーバーレス実行モードを使用している場合、アクションはタイムアウト（デフォルトでは2時間）に達すると停止します。Kafka Broker には新しいデータがいつでも到着し得るため、Kafka から Data Platform へのデータ供給を続けるには、停止したアクションを再実行する必要があります。そのため、特にプロダクション環境では、常時実行モードの使用をお勧めします。

#自動セグメンテーション

Kafka コネクタを使用する場合、Automatic Segmentation機能を使用することで、より高速な実行時間を実現できます。このオプションは、アクションの設定で利用可能です。大規模なワークロードの場合は、ぜひご利用ください！

#カスタムオフセット

もう1つの選択肢として、トピックの設定で最も古いメッセージまたは最新のメッセージから読み取るのではなく、カスタムオフセットからメッセージの読み取りを開始することもできます。

トピックで設定したLatestまたはEarliestオフセットポリシーを上書きし、定義されたオフセットからメッセージの読み取りを開始するには、ActionsのAdvanced modeを使用する必要があります。次の例のように、paras.load_fromフィールド内にフィールドを追加します（オフセット7から読み取りを開始します）：

"params": {
    "load_from": [
      {
        "offset_number": 7,
        ...
      }
    ],
...

#技術的な考慮事項

Earliest モードの場合、最も古いメッセージを読み取った後に、目的のテーブルを変更した場合、Load アクションを再度実行すると、トピック内のすべてのメッセージが再度読み取られます。

これは、OVHcloud Data Platformが消費した最後のメッセージのオフセットが、Data Platformデータベースの宛先テーブルのメタデータに保存されるために起こります。新しい宛先テーブルを使用する場合、オフセットは最初から再開されます。

#さらに進む

もしトレーニングや技術的なサポートが必要な場合は、営業担当者にお問い合わせください、またはこのリンクをクリックして見積もりを依頼し、プロフェッショナルサービスの専門家にプロジェクトのカスタム分析を依頼してください。

質問をする、フィードバックを提供する、または直接データプラットフォームを構築するチームと交流するには、専用のDiscordチャネルをご利用ください。

サポートが必要な場合は、ヘルプセンターでリクエストを作成してください。

当社のユーザーコミュニティに参加してください。