Todos los productos
Search
Centro de documentos

DataHub:DataHub cost saving guide

Última actualización: Sep 22, 2026

Este tema describe la nueva función de procesamiento por lotes en DataHub. Se explica su funcionamiento y cómo implementarla. Cambiar al modo por lotes mejora el rendimiento y reduce los costos, ya que disminuye el consumo de recursos del servidor y reduce significativamente las tarifas de uso.

Detalles de la actualización

Compatibilidad con la compresión zstd

DataHub ahora es compatible con el algoritmo de compresión Zstandard (zstd). En comparación con los algoritmos lz4 y deflate, que DataHub también admite, zstd ofrece un rendimiento superior.

Nota

Zstandard (zstd) es un algoritmo de compresión de alto rendimiento desarrollado por Facebook y liberado como código abierto en 2016. Este algoritmo destaca tanto en velocidad como en relación de compresión, lo que lo hace ideal para los escenarios de DataHub.

Transformación de serialización

DataHub introduce la serialización por lotes, un método para organizar los datos antes de transmitirlos. El término «lote» no se refiere a un método de serialización específico, sino a una encapsulación secundaria de los datos ya serializados. Por ejemplo, para enviar 100 registros a la vez, primero se serializan en un búfer. Luego se aplica un algoritmo de compresión a ese búfer para generar un búfer comprimido. Por último, se agrega un encabezado al búfer comprimido para almacenar metadatos, como su tamaño, la cantidad de registros de datos, el algoritmo de compresión y una suma de comprobación CRC. Este proceso genera un búfer de lote completo.

Esto resuelve los siguientes problemas:

  • Evita de manera eficaz los datos corruptos en la capa de negocio.

  • Reduce la sobrecarga de CPU del servidor y mejora el rendimiento del procesamiento de datos.

  • Disminuye la latencia en las operaciones simultáneas de lectura y escritura.

image
Nota

El cliente comprueba exhaustivamente la validez de los datos antes de enviar un búfer de lote. Por ello, el servidor solo necesita verificar la suma de comprobación CRC para confirmar que el búfer es válido y luego puede escribirlo directamente en el disco. Este proceso elimina las operaciones de serialización, deserialización, compresión, descompresión y validación en el servidor. Como resultado, el rendimiento del servidor mejora más de un 80 %. Además, dado que varios registros de datos se comprimen juntos, la relación de compresión también aumenta, lo que reduce los costos de almacenamiento.

Comparación de costos

The following test was conducted to verify the effects of batch processing. The test scenario is as follows:
  • The test data consists of ad delivery information with approximately 200 columns. Approximately 20% to 30% of the data values are null.
  • 1,000 data records per batch.
  • Avro is used for serialization within the batch.
  • The zstd compression algorithm is used to replace lz4, which is the default compression algorithm in previous versions.
The test produced the following results:

Data source size (bytes)

lz4 compressed size (bytes)

zstd compressed size (bytes)

protobuf serialization

11,506,677

3,050,640

1,158,868

batch serialization

11,154,596

2,931,729

1,112,693

DataHub billing is based on two main dimensions: storage and traffic. Other billable items are primarily intended as penalties to prevent resource abuse. Therefore, the test results are analyzed from the perspectives of storage and traffic.

  • Storage costs: When using protobuf serialization, DataHub does not compress data for storage. Data is compressed only during HTTP transmission. If you replace this method with batch processing and zstd compression, the storage size decreases from 11,506 KB to 1,112 KB. This reduces storage costs by approximately 90%.
  • Traffic costs: When using protobuf and lz4, the data size is 3,050 KB. When using batch processing and zstd, the data size is 1,112 KB. This reduces traffic costs by approximately 60%.
Importante The preceding results are based on sample data. Your actual results may vary depending on your data. We recommend that you run your own tests as needed.

Uso del procesamiento por lotes

Nota

Notas

  • The primary benefit of batch writing is maximizing the number of records in each batch. If the client cannot batch records, or if the number of records per batch is small, the performance improvement may not be significant.
  • To ensure a smooth transition, batch mode is compatible with the original read and write methods. Data written in batch mode can be read in the original mode, and vice versa. However, for optimal performance, you should write and consume data in the same mode. Mismatched writing and consumption degrades performance.

Importante

La versión más reciente de DataHub ya no requiere habilitar el esquema multiversión de un topic para usar el protocolo por lotes. Sin embargo, debe actualizar su cliente a una versión compatible. A continuación se muestran los clientes compatibles y sus versiones requeridas.

Lenguaje

Estado de compatibilidad

Versión requerida

Java SDK

Compatible

1.5.1 o posterior

Go SDK

Compatible

1.1.0 o posterior

Python

No compatible

-

C++

No compatible

-

Java example

En las versiones 1.5.1 y posteriores, el protocolo de serialización predeterminado es Batch. No se requiere ninguna configuración especial. Puede escribir datos como de costumbre, ya sea de forma síncrona o asíncrona.

Dependencia de Maven
<dependency>
  <groupId>com.aliyun.datahub</groupId>
  <artifactId>datahub-client-library</artifactId>
  <version>1.5.1</version>
</dependency>
Ejemplo de código
public static void main(String[] args) throws InterruptedException {
	// Obtain AccessKey information from an environment variable.
	EnvironmentVariableCredentialProvider provider = EnvironmentVariableCredentialProvider.create();

	String endpoint ="https://dh-cn-hangzhou.aliyuncs.com";
	String projectName = "test_project";
	String topicName = "test_topic";

	// Initialize the producer. The default configurations are used.
	ProducerConfig config = new ProducerConfig(endpoint, provider);
	DatahubProducer producer = new DatahubProducer(projectName, topicName, config);

	RecordSchema schema = producer.getTopicSchema();
	// If a multi-version schema is enabled, you can also get the schema of a specified version.
	// RecordSchema schema = producer.getTopicSchema(3);

	// For an asynchronous write, register a callback function as needed.
	WriteCallback callback = new WriteCallback() {
		@Override
		public void onSuccess(String shardId, List<RecordEntry> records, long elapsedTimeMs, long sendTimeMs) {
			System.out.println("write success");
		}

		@Override
		public void onFailure(String shardId, List<RecordEntry> records, long elapsedTimeMs, DatahubClientException e) {
			System.out.println("write failed");
		}
	};

	for (int i = 0; i < 10000; ++i) {
		try {
            // Generate data by schema.
            TupleRecordData data = new TupleRecordData(schema);
            data.setField("field1", "hello");
            data.setField("field2", 1234);
            RecordEntry recordEntry = new RecordEntry();
            recordEntry.setRecordData(data);

            producer.sendAsync(recordEntry, callback);
            // If you do not need to confirm whether the data was sent successfully, do not register a callback. Send the data directly.
            // producer.sendAsync(recordEntry, null);
        } catch (DatahubClientException e) {
            // TODO: Handle exceptions, which are typically non-retriable errors or errors that occur after the maximum number of retries.
            Thread.sleep(1000);
        }
	}

	// Ensure that all data is sent before the program exits.
	producer.flush(true);
	producer.close();
}

Ejemplo en Go

Dependencia en go.mod
require (
	github.com/aliyun/aliyun-datahub-sdk-go v1.1.0
)
Ejemplo de código

En las versiones 1.1.0 y posteriores, el protocolo de serialización predeterminado es Batch. No se requiere ninguna configuración especial. Puede escribir datos como de costumbre, ya sea de forma síncrona o asíncrona.

func handleSuccessRun(producer datahub.AsyncProducer) {
	for suc := range producer.Successes() {
		// handle request success
		fmt.Printf("shard:%s, rid:%s, records:%d, latency:%v\n",
			suc.ShardId, suc.RequestId, len(suc.Records), suc.Latency)
	}
}

func handleFailedRun(producer datahub.AsyncProducer) {
	// handle request failed
	for err := range producer.Errors() {
		fmt.Printf("shard:%s, records:%d, latency:%v, error:%v\n",
			err.ShardId, len(err.Records), err.Latency, err.Err)
	}
}

func main() {
	cfg := datahub.NewProducerConfig()

    // Obtain AccessKey information from an environment variable.
	credential, err := credentials.NewCredential(nil)
	if err != nil {
		fmt.Println(err)
		// TODO: handle error
	}
	cfg.Account = datahub.NewCredentialAccount(credential)
	cfg.Endpoint = "https://dh-cn-hangzhou.aliyuncs.com"
	cfg.Project = "test_project"
	cfg.Topic = "test_topic"

	producer := datahub.NewAsyncProducer(cfg)
	err = producer.Init()

	if err != nil {
		// TODO: handle error
		fmt.Println(err)
	}

	schema, err := producer.GetSchema()
	if err != nil {
		// TODO: handle error
		fmt.Println(err)
	}

	// Handle the success channel.
	go handleSuccessRun(producer)

	// Handle the error channel.
	go handleFailedRun(producer)

	// Generate data in a loop.
	for i := 0; i < 1000; i++ {
		record := datahub.NewTupleRecord(schema)
        // Each time you set a value, check if the operation was successful.
		err = record.SetValueByName("f1", "val1")
		if err != nil {
			fmt.Println(err)
			return
		}

		err = record.SetValueByName("f2", 1234)
		if err != nil {
			fmt.Println(err)
			return
		}

		producer.Input() <- record
	}

	err = producer.Close()
	if err != nil {
		fmt.Println(err)
	}
}

Soporte

If you have any questions or issues, you can submit un ticket for support or join our user group. The group number is 33517130.