DataHub:DataHub によるコスト削減

コンソールでマルチバージョンスキーマ機能を有効にする

コンソールで作成済み Topic の情報を変更することはできません。バッチシリアル化を使用する場合は、Topic の作成時に [複数バージョンを有効にする] をオンにする必要があります。Topic の作成方法の詳細については、「Topic を管理する」をご参照ください。

SDK を使用してマルチバージョンスキーマ機能を有効にする

  public static void createTopicWithOption() {
    try {RecordSchema recordSchema = new RecordSchema() {{
        this.addField(new Field("field1", FieldType.STRING));
        this.addField(new Field("field2", FieldType.BIGINT));
      }};
      TopicOption option = new TopicOption();
       // マルチバージョンスキーマ機能を有効にします。   
      option.setEnableSchemaRegistry(true);
     
      option.setComment(Constant.TOPIC_COMMENT);
      option.setExpandMode(ExpandMode.ONLY_EXTEND);
      option.setLifeCycle(Constant.LIFE_CYCLE);
      option.setRecordType(RecordType.TUPLE);
      option.setRecordSchema(recordSchema);
      option.setShardCount(Constant.SHARD_COUNT);
      datahubClient.createTopic(Constant.PROJECT_NAME, Constant.TOPIC_NAME, option);
      LOGGER.info("create topic successful");
    } catch (ResourceAlreadyExistException e) {
      LOGGER.info("topic already exists, please check if it is consistent");
    } catch (ResourceNotFoundException e) {
      // project not found
      e.printStackTrace();
      throw e;
    } catch (DatahubClientException e) {
      // other error
      e.printStackTrace();
      throw e;
    }
  }

Maven 依存関係を追加する

<dependency>
  <groupId>com.aliyun.datahub</groupId>
  <artifactId>aliyun-sdk-datahub</artifactId>
  <version>2.25.3</version>
</dependency>
<dependency>
  <groupId>com.aliyun.datahub</groupId>
  <artifactId>datahub-client-library</artifactId>
  <version>1.4.3</version>
</dependency>

バッチシリアル化を構成する

ProducerConfig config = new ProducerConfig(endpoint, accessId, accessKey);
DatahubProducer producer = new DatahubProducer(projectName, topicName, config);

RecordSchema schema = producer.getTopicSchema();
List<RecordEntry> recordList = new ArrayList<>();
// パフォーマンスを向上させるために、できるだけ多くのレコードを recordList に追加することをお勧めします。
// 可能であれば、recordList のサイズを 512 KB から 1 MB の範囲に設定します。
for (int i = 0; i < 1000; ++i) {
    RecordEntry record = new RecordEntry();
    TupleRecordData data = new TupleRecordData(schema);
    // 次のスキーマが使用されていると仮定します：{"fields":[{"name":"f1", "type":"STRING"},{"name":"f2", "type":"BIGINT"}]}
    data.setField("f1", "value" + i);
    data.setField("f2", i);
    record.setRecordData(data);

    // オプション。カスタム属性を追加します。
    record.addAttribute("key1", "value1");
    recordList.add(record);
}

try {
      // データを 1,000 回繰り返し書き込みます。
      for (int i = 0; i < 1000; ++i) {
        try {
          String shardId = datahubProducer.send(recordList);
          LOGGER.info("Write shard {} success, record count:{}", shardId, recordList.size());
        } catch (DatahubClientException e) {
          if (!ExceptionChecker.isRetryableException(e)) {
            LOGGER.info("Write data fail", e);
            break;
          }
          // sleep 文を実行してデータ書き込みを再試行します。
          Thread.sleep(1000);
        }
      }
    } finally {
      // プロデューサー関連のリソースを無効にします。
      datahubProducer.close();
    }