DataHub:DataHub Kafka 互換モード

DataHub Kafka 互換モード

DataHub は Apache Kafka プロトコルと互換性があります。標準の Kafka SDK を使用して DataHub に接続し、データを発行およびサブスクライブすることができます。

DataHub と Kafka の概念マッピング

Kafka	DataHub
トピック	Project.topic
パーティション	シャード
オフセット	シーケンス

Kafka トピック

DataHub にもトピックはありますが、DataHub には Kafka には存在しないプロジェクトという追加のリソースレイヤーが含まれています。互換性を確保するため、Kafka のトピックは、DataHub のプロジェクトとトピックをピリオド (.) で結合したものに対応します。たとえば、DataHub に test_project という名前のプロジェクトと test_topic という名前のトピックがある場合、対応する Kafka のトピック名は test_project.test_topic となります。

Kafka パーティション

Kafka のパーティションは、DataHub のシャードに直接相当します。どちらも順序付けられたデータのキューを表します。

Kafka コンシューマーグループ

DataHub のコンシューマーグループは、Kafka のコンシューマーグループと同様に動作します。プロジェクト内にコンシューマーグループを作成し、サブスクライブしたいトピックに関連付けることができます。これにより、グループはそのプロジェクト内の複数のトピックをサブスクライブできます。コンシューマーグループがトピックにバインドされると、サブスクリプションが自動的に作成され、トピックのサブスクリプションリストページで確認できます。このサブスクリプションを削除すると、グループはそのトピックをサブスクライブできなくなり、以前のすべてのコンシューマーオフセットが消去されます。

コンシューマーグループはプロジェクトのサブリソースであるため、プロジェクトと一緒に指定する必要があります。たとえば、DataHub のプロジェクトが test_project で、コンシューマーグループが test_group の場合、Kafka のグループ名は test_project.test_group となります。

[コンシューマーグループの作成] ダイアログボックスで、[名前] と [説明] を入力し、[トピック] エリアでサブスクライブするトピックを選択し、[>] ボタンをクリックして右側の選択済みリストに追加し、[作成] をクリックします。

各グループは最大 50 のトピックをサブスクライブできます。それ以上をサブスクライブする必要がある場合は、チケットを起票してください。

Kafka レコード

Kafka のレコードはキーと値のペア形式です。対照的に、DataHub のレコードにはタプルと BLOB の 2 つのフォーマットがあります。タプルレコードには厳密に構造化されたデータが含まれ、BLOB レコードにはバイナリデータが含まれます。Kafka のレコードは通常、ヘッダーとキーと値のペアの 2 つの部分で構成されます。

Kafka ヘッダー

Kafka のヘッダーはデータに追加情報を付加できます。これは DataHub の属性と同じ目的を果たします。Kafka クライアントを使用してヘッダーを含むデータを書き込むと、ヘッダー情報は DataHub の属性として保存されます。Kafka ヘッダーの値が NULL の場合、対応するヘッダーは無視されます。ヘッダーキーとして "__kafka_key__" を使用しないでください。これは予約済みの内部キーです。

Kafka のキーと値のデータ

• DataHub のトピックがタプル型の場合、そのキーは最初の String 列に対応し、値は 2 番目の String 列に対応します。

• DataHub のトピックが BLOB 型の場合、その値はデータコンテンツに対応し、キーは <"__kafka_key__", key> の形式で属性に格納されます。

Kafka オフセット

Kafka のオフセットは 64 ビットの整数です。パーティションの場合、オフセットは 0 から始まり、自動的にインクリメントされます。これにより、各データレコードが一意のオフセットを持つことが保証され、これは主にコンシューマーのオフセットを記録するために使用されます。DataHub のシーケンスは、Kafka のオフセットと同じ意味を持ちます。したがって、DataHub のシーケンスは Kafka のオフセットに直接相当します。

制限事項

DataHub は、Kafka のトランザクション、べき等性、SchemaRegistry、またはログコンパクションをサポートしていません。

クイックスタート

1. 推奨される Kafka クライアントバージョン 2.4.0 を使用してください。バージョン 0.10.0 から 4.0 までと互換性があります。

2. DataHub で対応するプロジェクト、トピック、およびグループのリソースを作成します。

3. セキュリティ認証方法を SASL/SSL に変更し、SASL には PLAIN メカニズムを使用し、Alibaba Cloud の AccessKey ペア (AK/SK) を設定します。

4. Kafka ブローカー情報を DataHub エンドポイントに変更します (サービスドメインのリストをご参照ください)。

リソースの作成

まず、DataHub コンソールにログインし、プロジェクトを作成します。

[新しいプロジェクト] ダイアログボックスで、[名前] を test_kafka_project に、[説明] を test kafka に設定します。

トピックの作成

DataHub コンソールの [新しいトピック] ページで、作成方法として [直接作成] を選択し、名前として test_kafka を入力し、タイプとして [TUPLE] を選択します。[スキーマの詳細] セクションで、key と value という名前の 2 つのフィールドを追加し、両方のタイプを [STRING] に設定し、[NULLを許可] を選択します。[シャード数] を 1 に、[ライフサイクル] を 3 に設定し、[シャードスケーリングモード] を有効にし、[複数バージョン] を無効にします。説明を入力して [作成] をクリックします。

グループを作成し、コンシュームする必要があるトピックをバインドします。グループ作成後にバインドされたトピックのリストを変更することもできます。メッセージをコンシュームする必要がない場合は、この手順をスキップできます。

[新しいグループ] パネルで [名前] と [説明] を入力し、コンシュームするトピックを左側の利用可能なリストから右側の選択済みリストに移動させ、[作成] をクリックします。

認証方法

kafka_client_producer_jaas.conf という名前のファイルを作成し、任意のパスに保存します。ファイルの内容は次のとおりです。

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="accessId"
  password="accessKey";
};

kafka_client_producer_jaas.conf という名前のファイルを作成し、任意のディレクトリに保存します。ファイルには、次の内容を含める必要があります。

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="YOUR_ACCESS_ID"
  password="YOUR_ACCESS_KEY";
};

プロデューサーの例

kafka クライアント pom ファイル

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.4.0</version>
</dependency>

Kafka クライアントの pom.xml ファイル：

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.4.0</version>
</dependency>

ProducerExample：

コンシューマーの例

コンシューマーの例

package com.aliyun.datahub.kafka.demo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample2 {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("group.id", "test_project.test_kafka_group");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("ssl.endpoint.identification.algorithm", "");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        List<String> topicList = new ArrayList<>();
        topicList.add("test_project.test_topic1");
        topicList.add("test_project.test_topic2");
        topicList.add("test_project.test_topic3");
        kafkaConsumer.subscribe(topicList);
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.toString());
            }
        }
    }
}

ConsumerExample：

package com.aliyun.datahub.kafka.demo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("group.id", "test_project.test_kafka_group");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("ssl.endpoint.identification.algorithm", "");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        List<String> topicList = new ArrayList<>();
        topicList.add("test_project.test_topic1");
        topicList.add("test_project.test_topic2");
        topicList.add("test_project.test_topic3");
        kafkaConsumer.subscribe(topicList);
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.toString());
            }
        }
    }
}

Streams の例

このコードは、test_project の input からデータを読み取り、キーと値の文字列を小文字に変換し、データを output に書き換えます。

public class StreamExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(final String[] args) {
        final String input = "test_project.input";
        final String output = "test_project.output";
        final Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("application.id", "test_project.test_kafka_group");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("auto.offset.reset", "earliest");
        final StreamsBuilder builder = new StreamsBuilder();
        TestMapper testMapper = new TestMapper();
        builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
                .map(testMapper)
                .to(output, Produced.with(Serdes.String(), Serdes.String()));
        final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });
        try {
            streams.start();
            latch.await();
        } catch (final Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
    static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
        @Override
        public KeyValue<String, String> apply(String s, String s2) {
            return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
        }
    }
}

この例では、test_project プロジェクトの入力トピックからデータを読み取り、キーと値の文字列を小文字に変換し、結果を出力トピックに書き込みます。

public class StreamExample {
    static {
        System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
    }
    public static void main(final String[] args) {
        final String input = "test_project.input";
        final String output = "test_project.output";
        final Properties properties = new Properties();
        properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
        properties.put("application.id", "test_project.test_kafka_group");
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "PLAIN");
        properties.put("session.timeout.ms", "60000");
        properties.put("heartbeat.interval.ms", "40000");
        properties.put("auto.offset.reset", "earliest");
        final StreamsBuilder builder = new StreamsBuilder();
        TestMapper testMapper = new TestMapper();
        builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
                .map(testMapper)
                .to(output, Produced.with(Serdes.String(), Serdes.String()));
        final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });
        try {
            streams.start();
            latch.await();
        } catch (final Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }
    static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
        @Override
        public KeyValue<String, String> apply(String s, String s2) {
            return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
        }
    }
}

Streams ジョブを開始すると、シャードの割り当てに約 1 分かかります。その後、コンソールで現在のタスク数を確認できます。タスク数は入力トピックのシャード数と一致します。この例では、入力トピックには 3 つのシャードがあります。

出力サンプリングの結果は、データが正しく書き込まれたことを示しています。Streams タスクの TestMapper によって処理された後、大文字の入力は小文字の出力に変換されました。キーは aaaa、cccc、eeee で、対応する値は bbbb、dddd、ffff です。

セルフマネージド Kafka から DataHub へのデータ移行

1. ブローカーアドレスを切り替えます。詳細については、以下のドメイン名リストをご参照ください。

2. DataHub でリソースを作成し、コード内のリソース名を変更します。詳細については、上記のリソース作成セクションをご参照ください。

3. 認証方法を SASL_SSL に切り替えます。認証メカニズムは PLAIN です。ユーザー名に Alibaba Cloud の AccessKey ID (AK) を、パスワードに AccessKey Secret (SK) を設定します。

付録

設定の概要

C=コンシューマー、P=プロデューサー、S=Streams

パラメーター	C/P/S	推奨値	必須	説明
bootstrap.servers	*	ドメイン名のリストをご参照ください。	はい
security.protocol	*	SASL_SSL	はい	安全なデータ転送を確保するため、Kafka から DataHub へのデータ書き込みはデフォルトで SSL 暗号化を使用します。
sasl.mechanism	*	PLAIN	はい	AccessKey 認証。PLAIN のみがサポートされています。
compression.type	P	LZ4	いいえ	データ転送で圧縮を有効にするかどうかを指定します。現在、LZ4 のみがサポートされています。
enable.idempotence	P	false	いいえ	Kafka クライアントバージョン 3.0.1 以降では、べき等性がデフォルトで有効になっています。DataHub はべき等性をサポートしていないため、この機能を手動で無効にする必要があります。この設定は、バージョン 3.0.1 より前のクライアントでは不要です。
group.id	C	project.group	はい
partition.assignment.strategy	C	org.apache.kafka.clients.consumer.RangeAssignor	いいえ	Kafka のデフォルト値は RangeAssignor です。DataHub は現在 RangeAssignor のみをサポートしています。この設定は変更しないでください。
session.timeout.ms	C/S	[60000, 180000]	いいえ	Kafka のデフォルト値は 10000 です。ただし、DataHub は最小値 60000 を要求するため、値はデフォルトで 60000 になります。
heartbeat.interval.ms	C/S	session.timeout.ms 値の 3 分の 2。	いいえ	Kafka でのデフォルト値は 3000 です。session.timeout.ms のデフォルト値は 60000 であるため、頻繁なハートビートリクエストを防ぐには、この値を明示的に 40000 に設定してください。
application.id	S	project.topic:subId または project.group	はい	project.topic:subId を使用する場合、サブスクライブするトピックと一致させる必要があります。そうしないと、データを読み取れません。project.group の使用を推奨します。

サービスドメイン名リスト

リージョン名	リージョン	パブリックエンドポイント	VPC ECS エンドポイント
中国東部 1 (杭州)	cn-hangzhou	dh-cn-hangzhou.aliyuncs.com:9092	dh-cn-hangzhou-int-vpc.aliyuncs.com:9094
中国東部 2 (上海)	cn-shanghai	dh-cn-shanghai.aliyuncs.com:9092	dh-cn-shanghai-int-vpc.aliyuncs.com:9094
中国北部 2 (北京)	cn-beijing	dh-cn-beijing.aliyuncs.com:9092	dh-cn-beijing-int-vpc.aliyuncs.com:9094
中国 (ウランチャブ)	cn-wulanchabu	dh-cn-wulanchabu.aliyuncs.com:9092	dh-cn-wulanchabu-int-vpc.aliyuncs.com:9094
中国南部 1 (深圳)	cn-shenzhen	dh-cn-shenzhen.aliyuncs.com:9092	dh-cn-shenzhen-int-vpc.aliyuncs.com:9094
中国北部 3 (張家口)	cn-zhangjiakou	dh-cn-zhangjiakou.aliyuncs.com:9092	dh-cn-zhangjiakou-int-vpc.aliyuncs.com:9094
アジアパシフィック SE 1 (シンガポール)	ap-southeast-1	dh-ap-southeast-1.aliyuncs.com:9092	dh-ap-southeast-1-int-vpc.aliyuncs.com:9094
アジアパシフィック SE 3 (クアラルンプール)	ap-southeast-3	dh-ap-southeast-3.aliyuncs.com:9092	dh-ap-southeast-3-int-vpc.aliyuncs.com:9094
ヨーロッパ中央 1 (フランクフルト)	eu-central-1	dh-eu-central-1.aliyuncs.com:9092	dh-eu-central-1-int-vpc.aliyuncs.com:9094
中国東部 2 (上海) 金融	cn-shanghai-finance-1	dh-cn-shanghai-finance-1.aliyuncs.com:9092	dh-cn-shanghai-finance-1-int-vpc.aliyuncs.com:9094
中国 (香港)	cn-hongkong	dh-cn-hongkong.aliyuncs.com:9092	dh-cn-hongkong-int-vpc.aliyuncs.com:9094

Kafka API 互換性

Kafka の公式ドキュメントには、利用可能なすべての API がリストされています。DataHub の互換性を理解しやすくするために、互換性のある API のリストも提供しています。

API	説明
Produce	データを書き込みます。
Fetch	データを読み取ります。
ListOffsets	時間に基づいてオフセットを取得します。古いバージョンはオフセットのリストを返し、新しいバージョンは単一のオフセットを返します。
Metadata	読み取りおよび書き込み操作のメタデータを取得します。
OffsetCommit	オフセットをコミットします。
OffsetFetch	オフセットを取得します。
FindCoordinator	コーディネーターをホストするブローカーを検索し、その仮想 IP (VIP) アドレスを返します。
JoinGroup	グループに参加します。
Heartbeat	ハートビートを送信します。
LeaveGroup	グループから離脱します。
SyncGroup	グループリーダーがパーティション割り当てプランを送信し、すべてのメンバーがそれを受信するために使用されます。
SaslHandshake	認証のために Simple Authentication and Security Layer (SASL) ハンドシェイクを処理します。
ApiVersions	利用可能なすべての API を取得します。
CreateTopics	トピックを作成します。
DeleteTopics	トピックを削除します。
OffsetForLeaderEpoch	最新のオフセットを取得します。
SaslAuthenticate	SASL を使用してクライアントを認証します。
CreatePartitions	パーティションを追加します。
DeleteGroups	グループを削除します。
OffsetDelete	オフセットを削除します。