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.
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.
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.
|
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%.
Uso del procesamiento por lotes
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.
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.