Lindorm:プルモードでのデータサブスクリプションチャネルの作成

前提条件

クライアント IP を IP ホワイトリストに追加済みであること。詳細については、「IP ホワイトリストの設定」をご参照ください。

データサブスクリプション機能が有効化済みであること。詳細については、「データサブスクリプションの有効化」をご参照ください。

操作手順

1. LTS ページに移動します。左側のナビゲーションウィンドウで、Change Tracking > [プルモード] を選択します。

   これにより、Lindorm CDC の [サブスクリプションチャネルリスト] ページに移動します。[データサブスクリプションチャネルの作成] ボタンは右上隅にあります。リストには、既存のチャネルのサブスクリプションチャネル ID、Lindorm テーブル名、トピック名、および操作 ([詳細] と [削除]) が表示されます。

2. [データサブスクリプションチャネルの作成] をクリックし、次のパラメーターを設定します。

   パラメーター	説明
   ソースクラスター	Lindorm インスタンスの ID を入力します。
   Lindorm テーブル名	サブスクライブする Lindorm テーブルを選択します。各チャネルは 1 つのテーブルにのみサブスクライブできます。
   トピック名	データ消費トピックの名前です。
   [データ保持期間 (日)]	データを保持する日数です。デフォルト値は 7 です。
   [トピックパーティション数]	トピックのパーティション数です。複数のパーティションにより、同時データ消費が可能になります。デフォルト値は 4 です。

3. 送信 をクリックします。

4. (任意) チャネルの詳細を表示するには、リストで対象のチャネルを見つけ、操作 列の 詳細 をクリックします。チャネル詳細、消費詳細、およびストレージ詳細を表示できます。

5. (任意) Kafka クライアントで次のコード例を使用して、サブスクライブしたデータを消費します。

   import org.apache.hadoop.hbase.util.Bytes;
   import org.apache.kafka.clients.admin.AdminClientConfig;
   import org.apache.kafka.clients.consumer.ConsumerConfig;
   import org.apache.kafka.clients.consumer.ConsumerRecord;
   import org.apache.kafka.clients.consumer.ConsumerRecords;
   import org.apache.kafka.clients.consumer.KafkaConsumer;
   import org.apache.kafka.common.serialization.ByteArrayDeserializer;
   import java.time.Duration;
   import java.util.Arrays;
   import java.util.Properties;
   public class TestConsume {
     public static void main(String[] args) throws Exception {
       // データサブスクリプションチャネルの作成時に指定したトピック名。
       String topic = "test-topic";
       // エンドポイントに接続するためのプロパティ。
       Properties props = new Properties();
       // エンドポイントアドレスを指定します。
       props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "ld-xxx:9092");
       // キーのデシリアライザー。この値は変更しないでください。
       props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // 値のデシリアライザー。この値は変更しないでください。
       props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // コンシューマーグループの名前。グループは消費中に自動的に作成されます。
       props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id-0");
       // コンシューマーを作成します。
       KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
       // トピックにサブスクライブします。
       consumer.subscribe(Arrays.asList(topic));
       // コンシューマーを使用してデータをプルします。
       ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(10000));
       for (ConsumerRecord<byte[], byte[]> record : records) {
         // データの内容を表示します。
         System.out.println("key: " + Bytes.toString(record.key()));
         System.out.println("value: " + Bytes.toString(record.value()));
       }
       // 現在のコンシューマーオフセットをコミットします。
       consumer.commitSync();
       // コンシューマーを閉じます。
       consumer.close();
     }
   }

   説明

   データ消費フォーマットの詳細については、「データ消費フォーマット」をご参照ください。