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

#目的

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

#はじめに

#必要条件

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

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

#コンセプトの概要

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

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

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

データを読み込んだ後、アクションが実行されている限り、メッセージは自動的にLakehouse Managerテーブルに読み込まれます。アクションはデフォルトでServerless実行モードで実行されることに注意してください。そのため、Kafkaコネクタを使用している場合は、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コネクタ記事を参照してください。

#スキーマを構成する

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

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

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

#テーブルを作成して構築する

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

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

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

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

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

#実行モード

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

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

If you are using the Serverless execution mode, the action will stop running once timeout is reached (by default 2 hours). Given that new data can arrive anytime in your Kafka Broker, this means that you need to re-run this action after it stopped if you want to keep feeding data from your Kafka to Data Platform. That is why we recommend the use of the Always-up execution mode, specially for production environments.

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

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

#カスタムオフセット

One other option available to you is to start reading your messages from a custom offset instead of configuring your topics to reading from the earliest or latest message.

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

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

#技術的な考慮事項

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

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

#さらに進む

トレーニングや技術的なサポートが必要な場合は、営業担当者にお問い合わせください。または、Professional Servicesページをご覧になり、見積もりをリクエストして、専門家によるプロジェクト分析を受けてください。

質問やフィードバックを送信し、Data Platformを構築するチームと直接交流するには、専用のDiscordチャネルをご利用ください。

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

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