TGIP-CN-009:Pulsar performance tuning

展开查看详情

1.TGIP-CN 009 Apache Pulsar performance tuning Penghui Li

2.Pulsar components Client Broker Bookie Pulsar SQL Function IO Connector Pulsar Proxy Metadata Store Tiered Storage

3.Outline 1.Basic concepts 2.Message writing 3.Message reading

4.Basic Concepts

5.Basic concepts Topic - Producer - Consumer 5 4 3 2 1 0 Producer Topic Consumer

6.Basic concepts Topic - Managed Ledger Topic Managed Ledger Ledger Ledger Ledger ō Ledger Ledger Ledger Cursor Cursor Cursor ō Cursor Cursor Cursor

7.Basic concepts Managed Ledger Cache Read from Cache Read from Bookies Topic Tailing Cache

8.Basic concepts Ledger - Entry Topic Ledger - 2 (open) Ledger - 1 (closed) Ledger - 0 (closed) Ledger rollover Fragment F - 1 F - 0 Fragment Bookie failover E-3 E-2 E-1 E-0 E-3 E-2 E-1 E-0 E-3 E-2 E-1 E-0

9.Basic concepts Journal - Entry log L-0 E-0 L-1 E-0 L-2 E-0 L-0 E-0 L-0 E-0 L-1 E-0 L-1 E-1 Checkpoint L-1 E-0 L-1 E-1 L-1 E-2 L-1 E-1 L-2 E-0 L-0 E-1 L-2 E-0 Journal Entry Log - 0 Index DB Entry Log - 1

10.Basic concepts Message Single Message Batched Message Client 2 Client 6 7 8 Broker 1 Broker 3 4 5 Bookie 0 Bookie 0 1 2

11.Basic concepts Message - Entry Entry-3 Entry-2 Entry-1 Entry-0 8 7 6 5 4 3 2 1 0 11 10 9

12.Message writing

13.Topic Producer 0 1 2 3 4 Message route mode Partition switch P-0 P-1 P-2 P-3 P-4 Partitioned Topic

14.Producer pending queue Producer Broker . 7 6 5 4 3 2 1 0 4 3 2 1 0 Send response to the client . 12 11 10 9 8 7 6 5

15.Producer pending queue Producer Broker 8 7 6 5 4 3 2 1 0 2 1 0 Send response to the client ….. 8 7 6 5 4 3

16.Publish batch message Producer<byte[]> producer = client.newProducer() .topic("my-topic") .enableBatching(true) .batchingMaxBytes(128 * 1024 * 1024) .batchingMaxMessages(1000) .batchingMaxPublishDelay(2, TimeUnit.MILLISECONDS) .blockIfQueueFull(true) .roundRobinRouterBatchingPartitionSwitchFrequency(10) .batcherBuilder(BatcherBuilder.DEFAULT) .create(); producer.newMessage().value(new byte[0]).send(); producer.newMessage().value(new byte[0]).sendAsync(); producer.send(new byte[0]); producer.sendAsync(new byte[0]);

17.Message compression Producer<byte[]> producer = client.newProducer() .topic("my-topic") .compressionType(CompressionType.LZ4) .create(); LZ4 ZLIB ZSTD SNAPPY All versions >= 2.3.0 >= 2.4.0

18.Message writing overview Topic ō Brokers ō Bookies ō

19.E Qw Qa Broker Broker 0 0 0 0 0 0 1 1 1 1 1 1 2 2 2 2 2 2 3 3 3 3 3 3 4 4 4 4 4 4 Bookie-0 Bookie-1 Bookie-2 Bookie-0 Bookie-1 Bookie-2 Bookie-3 Bookie-4 3-3-2 5-3-2

20.Bookie handle entry add Request processor 1.Append into Journal 3.Complete add request Journal Memory Table 2.Add into Ledger Storage 4.Periodically flush Entry Log Ledger Storage

21.Broker configurations 1. managedLedgerDefaultEnsembleSize 2. managedLedgerDefaultWriteQuorum 3. managedLedgerDefaultAckQuorum 4. managedLedgerNumWorkerThreads 5. numIOThreads 6. -Dorg.apache.bookkeeper.conf.readsystemproperties=true -DnumIOThreads=8

22.Bookie configurations 1. Journal Directories 2. Ledger Directories 3. Journal sync data 4. Journal group commit 5. Write cache 6. Flush interval 7. Add worker threads and max pending add requests 8. Journal pagecache flush interval

23.Message reading

24.Consumer receiver queue 9 8 7 6 5 4 3 2 1 0 1 2 9 8 7 6 5 14 13 12 11 10 9 8 3 4

25.Message reading overview Topic ō Brokers ō Bookies ō

26.Message dispatcher Consumer Managed ledger Dispatcher Consumer Consumer

27.Key_Shared dispatcher 65535 Consumer - 0 Consumer - 3 1.Peek key 2.Get key hash 3.Group by hash % slots 3 2 0 29093 Consumer - 2 Consumer - 5 6 5 4 3 2 1 0 6 4 10297 Consumer - 4 5 1 507 Consumer - 6 Consumer - 1 0

28.Key based batcher <k0-v1> <k0-v0> <k0-v1> <k1-v1> <k1-v0> <k0-v0> <k1-v1> <k1-v0> partition_key ordering_key

29.Bookie configurations 1. dbStorage_rocksDB_blockCacheSize 2. dbStorage_readAheadCacheMaxSizeMb 3. dbStorage_readAheadCacheBatchSize 4. Read worker threads

StreamNative 是一家围绕 Apache Pulsar 和 Apache BookKeeper 打造下一代流数据平台的开源基础软件公司。秉承 Event Streaming 是大数据的未来基石、开源是基础软件的未来这两个理念,专注于开源生态和社区的构建,致力于前沿技术。