DataHub:Pengurangan biaya yang dibawa oleh DataHub

Mengaktifkan fitur skema multi-versi di konsol

Informasi topik yang telah dibuat di konsol tidak dapat dimodifikasi. Untuk menggunakan serialisasi batch, aktifkan opsi Enable Multi-version saat membuat topik. Untuk detail lebih lanjut tentang pembuatan topik, lihat Mengelola topik.

Mengaktifkan fitur skema multi-versi menggunakan 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();
       // Aktifkan fitur skema multi-versi.   
      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;
    }
  }

Tambahkan dependensi 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>

Konfigurasikan serialisasi batch

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

RecordSchema schema = producer.getTopicSchema();
List<RecordEntry> recordList = new ArrayList<>();
// Untuk mencapai performa yang lebih baik, kami sarankan Anda menambahkan sebanyak mungkin catatan ke recordList.
// Atur ukuran recordList dalam rentang 512 KB hingga 1 MB jika memungkinkan.
for (int i = 0; i < 1000; ++i) {
    RecordEntry record = new RecordEntry();
    TupleRecordData data = new TupleRecordData(schema);
    // Anggap skema berikut digunakan: {"fields":[{"name":"f1", "type":"STRING"},{"name":"f2", "type":"BIGINT"}]}
    data.setField("f1", "value" + i);
    data.setField("f2", i);
    record.setRecordData(data);

    // Opsional. Tambahkan atribut kustom.
    record.addAttribute("key1", "value1");
    recordList.add(record);
}

try {
      // Tulis data berulang kali sebanyak 1.000 kali.
      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;
          }
          // Jalankan pernyataan sleep untuk mencoba menulis ulang data.
          Thread.sleep(1000);
        }
      }
    } finally {
      // Nonaktifkan sumber daya terkait produser.
      datahubProducer.close();
    }