Bu konuda, DataHub'daki yeni toplu işleme özelliğinin çalışma şekli ve nasıl uygulanacağı açıklanmaktadır. Toplu moda geçiş yapmak, sunucu tarafındaki kaynak tüketimini azaltarak ve kullanım ücretlerini önemli ölçüde düşürerek performansı artırır ve maliyetleri azaltır.
Yükseltme ayrıntıları
zstd sıkıştırma desteği
DataHub artık Zstandard (zstd) sıkıştırma algoritmasını destekler. DataHub'ın desteklediği diğer lz4 ve deflate algoritmalarıyla karşılaştırıldığında zstd daha üstün performans sunar.
Zstandard (zstd), Facebook tarafından geliştirilen ve 2016 yılında açık kaynak hale getirilen yüksek performanslı bir sıkıştırma algoritmasıdır. Hem sıkıştırma hızı hem de sıkıştırma oranı açısından mükemmel sonuçlar veren zstd, DataHub senaryoları için son derece uygundur.
Serileştirme dönüşümü
DataHub, verileri iletim için düzenleyen bir yöntem olan toplu serileştirmeyi tanıtır. 'Toplu' terimi belirli bir serileştirme yöntemini ifade etmez; bunun yerine serileştirilmiş verilerin ikinci kez kapsüllenmesini ifade eder. Örneğin, tek seferde 100 kayıt göndermek için önce bu 100 kayıt bir arabellekte serileştirilir. Ardından bu arabelleğe bir sıkıştırma algoritması uygulanarak sıkıştırılmış bir arabellek oluşturulur. Son olarak, boyutu, veri kaydı sayısı, kullanılan sıkıştırma algoritması ve CRC sağlama toplamı gibi meta verileri depolamak üzere sıkıştırılmış arabelleğe bir başlık eklenir. Bu işlem eksiksiz bir toplu arabellek oluşturur.
Bu yaklaşım aşağıdaki sorunları çözer:
İş katmanındaki hatalı verileri etkin bir şekilde önler.
Sunucu tarafındaki CPU yükünü azaltır ve veri işleme performansını artırır.
Eşzamanlı okuma ve yazma gecikme süresini düşürür.
İstemci, toplu arabelleği göndermeden önce kapsamlı veri geçerlilik kontrolleri yaptığından, sunucunun arabelleğin geçerli olduğunu doğrulamak için yalnızca CRC sağlama toplamını kontrol etmesi yeterlidir. Arabellek daha sonra doğrudan diske yazılabilir. Bu süreç, sunucu tarafında serileştirme, serileştirmeyi geri alma, sıkıştırma, açma ve doğrulama işlemlerine duyulan ihtiyacı ortadan kaldırır. Sonuç olarak, sunucu tarafındaki performans %80'in üzerinde artar. Birden fazla veri kaydının birlikte sıkıştırılması nedeniyle sıkıştırma oranı da iyileşir ve bu durum depolama maliyetlerini düşürür.
Maliyet karşılaştırması
Toplu işlemenin etkilerini doğrulamak amacıyla aşağıdaki test gerçekleştirilmiştir. Test senaryosu şu şekildedir:
Test verileri yaklaşık 200 sütundan oluşan reklam teslimat bilgilerini içerir. Veri değerlerinin yaklaşık %20 ila %30'u boştur.
Her toplu işlemde 1.000 veri kaydı bulunur.
Toplu işlem içinde serileştirme için Avro kullanılır.
Önceki sürümlerdeki varsayılan sıkıştırma algoritması olan lz4 yerine zstd sıkıştırma algoritması kullanılır.
Testten elde edilen sonuçlar şunlardır:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
DataHub faturalaması temel olarak depolama ve trafik olmak üzere iki ana boyuta dayanır. Diğer faturalandırılan öğeler öncelikli olarak kaynak kötüye kullanımını önlemeye yönelik cezalandırma amaçlıdır. Bu nedenle test sonuçları depolama ve trafik perspektiflerinden analiz edilmiştir.
Depolama maliyetleri: protobuf serileştirme kullanıldığında DataHub verileri depolama için sıkıştırmaz. Veriler yalnızca HTTP iletimi sırasında sıkıştırılır. Bu yöntem yerine toplu işleme ve zstd sıkıştırma kullanılırsa depolama boyutu 11.506 KB'tan 1.112 KB'a düşer. Bu durum depolama maliyetlerini yaklaşık %90 oranında azaltır.
Trafik maliyetleri: protobuf ve lz4 kullanıldığında veri boyutu 3.050 KB'tır. Toplu işleme ve zstd kullanıldığında ise veri boyutu 1.112 KB olur. Bu durum trafik maliyetlerini yaklaşık %60 oranında azaltır.
Yukarıdaki sonuçlar örnek verilere dayanmaktadır. Gerçek sonuçlarınız kullandığınız verilere göre değişiklik gösterebilir. Gerektiğinde kendi testlerinizi çalıştırmanızı öneririz.
Toplu işlemeyi kullanma
Notlar
Toplu yazmanın temel faydası her toplu işlemin içindeki kayıt sayısını en üst düzeye çıkarmaktır. İstemci kayıtları toplu işleyemiyorsa veya toplu işlem başına düşen kayıt sayısı azsa performans artışı önemli olmayabilir.
Sorunsuz bir geçiş sağlamak için toplu mod orijinal okuma ve yazma yöntemleriyle uyumludur. Toplu modda yazılan veriler orijinal modda okunabilir ve tam tersi de geçerlidir. Ancak en iyi performans için verileri aynı modda yazmalı ve tüketmelisiniz. Mismatched writing and consumption degrades performance.
DataHub'ın en son sürümünde toplu protokolü kullanmak için bir konu için çok sürümlü şemayı etkinleştirmeniz gerekmez. Ancak istemcinizi desteklenen bir sürüme güncellemeniz gerekir. Aşağıdaki tabloda desteklenen istemciler ve gerekli sürümleri listelenmiştir.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Java örneği
1.5.1 ve sonraki sürümlerde varsayılan serileştirme protokolü Batch'tir. Özel bir yapılandırma gerekmez. Senkron veya asenkron yazma kullanarak verileri her zamanki gibi yazabilirsiniz.
Maven bağımlılığı
<dependency>
<groupId>com.aliyun.datahub</groupId>
<artifactId>datahub-client-library</artifactId>
<version>1.5.1</version>
</dependency>
Kod örneği
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();
}
Go örneği
go.mod bağımlılığı
require (
github.com/aliyun/aliyun-datahub-sdk-go v1.1.0
)
Kod örneği
1.1.0 ve sonraki sürümlerde varsayılan serileştirme protokolü Batch'tir. Özel bir yapılandırma gerekmez. Senkron veya asenkron yazma kullanarak verileri her zamanki gibi yazabilirsiniz.
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)
}
}
Destek
Sorularınız veya sorunlarınız varsa destek almak için bir destek bileti oluşturun veya kullanıcı grubumuza katılın. Grup numarası 33517130'dur.