向 HStreamDB 中的 Stream 写入 Records

本文档提供了关于如何通过 hstreamdb-java 等客户端向 HStreamDB 中的 Stream 写入数据的相关教程。

同时还可参考其他的相关教程：

为了向 HStreamDB 写数据，我们需要将消息打包成 HStream Record，以及一个创建和发送 消息到服务器的 Producer。

HStream Record

Stream 中的所有数据都是以 HStream Record 的形式存在，HStreamDB 支持以下两种 HStream Record：

  • HRecord: 可以看作是一段 JSON 数据，就像一些 NoSQL 数据库中的 document。
  • Raw Record: 二进制数据。

端到端压缩

为了降低传输开销，最大化带宽利用率，HStreamDB 支持对写入的 HStream Record 进行压缩。 用户在创建 BufferedProducer 时可以设置压缩算法。当前可选的压缩算法有 gzip 和 zstd。客户端从 HStreamDB 中消费数据时会自动完成解压缩操作。

写入 HStream Record

有两种方法可以把 records 写入 HStreamDB。从简单易用的角度，你可以从 client.newProducer() 的Producer 入手。这个 Producer 没有提供任何配置项，它 只会即刻将收到的每个 record 并行发送到 HServer，这意味着它并不能保证这些 records 的顺序。在生产环境中， client.newBufferedProducer() 中的 BufferedProducer 将 是更好的选择，BufferedProducer 将按顺序缓存打包 records 成一个 batch，并将该 batch 发送到服务器。每一条 record 被写入 stream 时，HServer 将为该 record 生成一 个相应的 record ID，并将其发回给客户端。这个 record ID 在 stream 中是唯一的。

使用 Producer

  1. // WriteDataSimpleExample.java
  3. package docs.code.examples;
  5. import io.hstream.*;
  6. import io.hstream.Record;
  7. import java.nio.charset.StandardCharsets;
  8. import java.util.Arrays;
  9. import java.util.List;
  10. import java.util.concurrent.CompletableFuture;
  12. public class WriteDataSimpleExample {
  13.   public static void main(String[] args) throws Exception {
  14.     // TODO (developers): Replace these variables for your own use cases.
  15.     String serviceUrl = "127.0.0.1:6570";
  16.     if (System.getenv("serviceUrl") != null) {
  17.       serviceUrl = System.getenv("serviceUrl");
  18.     }
  20.     String streamName1 = "your_h_records_stream_name";
  21.     String streamName2 = "your_raw_records_stream_name";
  23.     // We do not recommend write both raw data and HRecord data into the same stream.
  24.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  26.     writeHRecordData(client, streamName1);
  27.     writeRawData(client, streamName2);
  28.     client.close();
  29.   }
  31.   public static void writeHRecordData(HStreamClient client, String streamName) {
  32.     // Create a basic producer for low latency scenarios
  33.     // For high throughput scenarios, please see the next section "Using `BufferedProducer`s"
  34.     Producer producer = client.newProducer().stream(streamName).build();
  36.     HRecord hRecord =
  37.         HRecord.newBuilder()
  38.             // Number
  39.             .put("id", 10)
  40.             // Boolean
  41.             .put("isReady", true)
  42.             // List
  43.             .put("targets", HArray.newBuilder().add(1).add(2).add(3).build())
  44.             // String
  45.             .put("name", "hRecord-example")
  46.             .build();
  48.     for (int i = 0; i <= 3000; i++) {
  49.       Record record = Record.newBuilder().hRecord(hRecord).build();
  50.       // If the data is written successfully, returns a server-assigned record id
  51.       CompletableFuture<String> recordId = producer.write(record);
  52.       System.out.println("Wrote message ID: " + recordId.join());
  53.     }
  54.   }
  56.   private static void writeRawData(HStreamClient client, String streamName) {
  57.     Producer producer = client.newProducer().stream(streamName).build();
  58.     List<String> messages = Arrays.asList("first", "second");
  59.     for (final String message : messages) {
  60.       Record record =
  61.           Record.newBuilder().rawRecord(message.getBytes(StandardCharsets.UTF_8)).build();
  62.       CompletableFuture<String> recordId = producer.write(record);
  63.       System.out.println("Published message ID: " + recordId.join());
  64.     }
  65.   }
  66. }
  1. // ExampleWriteProducer.go
  3. package examples
  5. import (
  6.     "github.com/hstreamdb/hstreamdb-go/hstream"
  7.     "github.com/hstreamdb/hstreamdb-go/hstream/Record"
  8.     "log"
  9. )
  11. func ExampleWriteProducer() error {
  12.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  13.     if err != nil {
  14.         log.Fatalf("Creating client error: %s", err)
  15.     }
  16.     defer client.Close()
  18.     producer, err := client.NewProducer("testStream")
  19.     if err != nil {
  20.         log.Fatalf("Creating producer error: %s", err)
  21.     }
  23.     defer producer.Stop()
  25.     payload := []byte("Hello HStreamDB")
  27.     rawRecord, err := Record.NewHStreamRawRecord("testStream", payload)
  28.     if err != nil {
  29.         log.Fatalf("Creating raw record error: %s", err)
  30.     }
  32.     for i := 0; i < 100; i++ {
  33.         appendRes := producer.Append(rawRecord)
  34.         if resp, err := appendRes.Ready(); err != nil {
  35.             log.Printf("Append error: %s", err)
  36.         } else {
  37.             log.Printf("Append response: %s", resp)
  38.         }
  39.     }
  41.     return nil
  42. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. async def append_records(client):
  21.     payloads = [b"some_raw_binary_bytes", {"msg": "hi"}]
  22.     rs = await client.append(stream_name, payloads)
  23.     for r in rs:
  24.         print("Append done, ", r)

使用 BufferedProducer

在几乎所有情况下，我们更推荐使用 BufferedProducer。不仅因为它能提供更大的吞吐 量，它还提供了更加灵活的配置去调整，用户可以根据需求去在吞吐量和时延之间做出调整 。你可以配置 BufferedProducer 的以下两个设置来控制和设置触发器和缓存区大小。通 过 BatchSetting，你可以根据 batch 的最大 record 数、batch 的总字节数和 batch 存在的最大时限来决定何时发送。通过配置 FlowControlSetting，你可以为所有的缓存 的 records 设置缓存大小和策略。下面的代码示例展示了如何使用 BatchSetting 来设置 响应的 trigger，以通知 producers 何时应该刷新，以及 FlowControlSetting 来限制 BufferedProducer 中的 buffer 的最大字节数。

  1. // WriteDataBufferedExample.java
  3. package docs.code.examples;
  5. import io.hstream.*;
  6. import io.hstream.Record;
  7. import java.util.ArrayList;
  8. import java.util.List;
  9. import java.util.Random;
  10. import java.util.concurrent.CompletableFuture;
  12. public class WriteDataBufferedExample {
  13.   public static void main(String[] args) throws Exception {
  14.     // TODO (developers): Replace these variables for your own use cases.
  15.     String serviceUrl = "127.0.0.1:6570";
  16.     if (System.getenv("serviceUrl") != null) {
  17.       serviceUrl = System.getenv("serviceUrl");
  18.     }
  20.     String streamName = "your_h_records_stream_name";
  21.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  22.     writeHRecordDataWithBufferedProducers(client, streamName);
  23.     client.close();
  24.   }
  26.   public static void writeHRecordDataWithBufferedProducers(
  27.       HStreamClient client, String streamName) {
  28.     BatchSetting batchSetting =
  29.         BatchSetting.newBuilder()
  30.             // optional, default: 100, the max records count of a batch,
  31.             // disable the trigger if the value <= 0.
  32.             .recordCountLimit(100)
  34.             // optional, default: 4096(4KB), the max bytes size of a batch,
  35.             // disable the trigger if the value <= 0.
  36.             .bytesLimit(4096)
  38.             // optional, default: 100(ms), the max age of a buffering batch,
  39.             // disable the trigger if the value <= 0.
  40.             .ageLimit(100)
  41.             .build();
  43.     // FlowControlSetting is to control total records,
  44.     // including buffered batch records and sending records
  45.     FlowControlSetting flowControlSetting =
  46.         FlowControlSetting.newBuilder()
  47.             // Optional, the default: 104857600(100MB), total bytes limit, including buffered batch
  48.             // records and
  49.             // sending records, the value must be greater than batchSetting.bytesLimit
  50.             .bytesLimit(40960)
  51.             .build();
  52.     BufferedProducer producer =
  53.         client.newBufferedProducer().stream(streamName)
  54.             .batchSetting(batchSetting)
  55.             .flowControlSetting(flowControlSetting)
  56.             .build();
  58.     List<CompletableFuture<String>> recordIds = new ArrayList<>();
  59.     Random random = new Random();
  61.     for (int i = 0; i < 100; i++) {
  62.       double temp = random.nextInt(100) / 10.0 + 15;
  63.       HRecord hRecord = HRecord.newBuilder().put("temperature", temp).build();
  64.       Record record = Record.newBuilder().hRecord(hRecord).build();
  65.       CompletableFuture<String> recordId = producer.write(record);
  66.       recordIds.add(recordId);
  67.     }
  69.     // close a producer, it will call flush() first
  70.     producer.close();
  71.     System.out.println("Wrote message IDs: " + recordIds.stream().map(CompletableFuture::join));
  72.   }
  73. }
  1. // ExampleWriteBatchProducer.go
  3. package examples
  5. import (
  6.     "github.com/hstreamdb/hstreamdb-go/hstream"
  7.     "github.com/hstreamdb/hstreamdb-go/hstream/Record"
  8.     "log"
  9. )
  11. func ExampleWriteBatchProducer() error {
  12.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  13.     if err != nil {
  14.         log.Fatalf("Creating client error: %s", err)
  15.     }
  16.     defer client.Close()
  18.     producer, err := client.NewBatchProducer("testDefaultStream", hstream.WithBatch(10, 500))
  19.     if err != nil {
  20.         log.Fatalf("Creating producer error: %s", err)
  21.     }
  22.     defer producer.Stop()
  24.     result := make([]hstream.AppendResult, 0, 100)
  25.     for i := 0; i < 100; i++ {
  26.         rawRecord, _ := Record.NewHStreamHRecord("", map[string]interface{}{
  27.             "id":      i,
  28.             "isReady": true,
  29.             "name":    "hRecord-example",
  30.         })
  31.         r := producer.Append(rawRecord)
  32.         result = append(result, r)
  33.     }
  35.     for i, res := range result {
  36.         resp, err := res.Ready()
  37.         if err != nil {
  38.             log.Printf("write error: %s\n", err.Error())
  39.         }
  40.         log.Printf("record[%d]=%s\n", i, resp.String())
  41.     }
  42.     return nil
  43. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. class AppendCallback(hstreamdb.BufferedProducer.AppendCallback):
  21.     count = 0
  23.     def on_success(self, stream_name, payloads, stream_keyid):
  24.         self.count += 1
  25.         print(f"Batch {self.count}: Append success with {len(payloads)} payloads.")
  27.     def on_fail(self, stream_name, payloads, stream_keyid, e):
  28.         print("Append failed!")
  29.         print(e)
  32. async def buffered_append_records(client):
  33.     p = client.new_producer(
  34.         append_callback=AppendCallback(),
  35.         size_trigger=10240,
  36.         time_trigger=0.5,
  37.         retry_count=2,
  38.     )
  40.     for i in range(50):
  41.         await p.append(stream_name, b"some_raw_binary_bytes")
  42.         await p.append(stream_name, {"msg": "hello"})
  44.     await p.wait_and_close()

使用分区键（Partition Key）

具有相同分区键的 records 可以在 BufferedProducer 中被保证能有序地写入。HStreamDB 的另一个重要功能，分区，也使用这些分区键来决定 records 将被分配到哪个分区， 以此提高写/读性能。更详细的解释请看管理 Stream 的分区。

参考下面的例子，你可以很容易地写入带有分区键的 records。

  1. // WriteDataWithKeyExample.java
  3. package docs.code.examples;
  5. import io.hstream.*;
  6. import io.hstream.Record;
  7. import java.util.ArrayList;
  8. import java.util.List;
  9. import java.util.Random;
  10. import java.util.concurrent.CompletableFuture;
  12. public class WriteDataWithKeyExample {
  13.   public static void main(String[] args) throws Exception {
  14.     // TODO (developers): Replace these variables for your own use cases.
  15.     String serviceUrl = "127.0.0.1:6570";
  16.     if (System.getenv("serviceUrl") != null) {
  17.       serviceUrl = System.getenv("serviceUrl");
  18.     }
  20.     String streamName = "your_h_records_stream_name";
  22.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  23.     writeHRecordDataWithKey(client, streamName);
  24.     client.close();
  25.   }
  27.   public static void writeHRecordDataWithKey(HStreamClient client, String streamName) {
  28.     // For demonstrations, we would use the following as our partition keys for the records.
  29.     // As the documentations mentioned, if we do not give any partition key, it will get a default
  30.     // key and be mapped to some default shard.
  31.     String key1 = "South";
  32.     String key2 = "North";
  34.     // Create a buffered producer with default BatchSetting and FlowControlSetting.
  35.     BufferedProducer producer = client.newBufferedProducer().stream(streamName).build();
  37.     List<CompletableFuture<String>> recordIds = new ArrayList<>();
  38.     Random random = new Random();
  40.     for (int i = 0; i < 100; i++) {
  41.       double temp = random.nextInt(100) / 10.0 + 15;
  42.       Record record;
  43.       if ((i % 3) == 0) {
  44.         HRecord hRecord = HRecord.newBuilder().put("temperature", temp).put("withKey", 1).build();
  45.         record = Record.newBuilder().hRecord(hRecord).partitionKey(key1).build();
  46.       } else {
  47.         HRecord hRecord = HRecord.newBuilder().put("temperature", temp).put("withKey", 2).build();
  48.         record = Record.newBuilder().hRecord(hRecord).partitionKey(key2).build();
  49.       }
  51.       CompletableFuture<String> recordId = producer.write(record);
  52.       recordIds.add(recordId);
  53.     }
  54.     System.out.println("Wrote message IDs: " + recordIds.stream().map(CompletableFuture::join));
  55.     producer.close();
  56.   }
  57. }
  1. // ExampleWriteBatchProducerMultiKey.go
  3. package examples
  5. import (
  6.     "fmt"
  7.     "github.com/hstreamdb/hstreamdb-go/hstream"
  8.     "github.com/hstreamdb/hstreamdb-go/hstream/Record"
  9.     "github.com/hstreamdb/hstreamdb-go/hstream/compression"
  10.     "log"
  11.     "math/rand"
  12.     "sync"
  13. )
  15. func ExampleWriteBatchProducerMultiKey() error {
  16.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  17.     if err != nil {
  18.         log.Fatalf("Creating client error: %s", err)
  19.     }
  20.     defer client.Close()
  22.     producer, err := client.NewBatchProducer("testStream",
  23.         // optional: set record count and max batch bytes trigger
  24.         hstream.WithBatch(10, 500),
  25.         // optional: set timeout trigger
  26.         hstream.TimeOut(1000),
  27.         // optional: set client compression
  28.         hstream.WithCompression(compression.Zstd),
  29.         // optional: set flow control
  30.         hstream.WithFlowControl(81920000))
  31.     if err != nil {
  32.         log.Fatalf("Creating producer error: %s", err)
  33.     }
  34.     defer producer.Stop()
  36.     keys := []string{"sensor1", "sensor2", "sensor3", "sensor4", "sensor5"}
  37.     rids := sync.Map{}
  38.     wg := sync.WaitGroup{}
  39.     wg.Add(5)
  41.     for _, key := range keys {
  42.         go func(key string) {
  43.             result := make([]hstream.AppendResult, 0, 100)
  44.             for i := 0; i < 100; i++ {
  45.                 temp := rand.Intn(100)/10.0 + 15
  46.                 rawRecord, _ := Record.NewHStreamHRecord(key, map[string]interface{}{
  47.                     key: fmt.Sprintf("temperature=%d", temp),
  48.                 })
  49.                 r := producer.Append(rawRecord)
  50.                 result = append(result, r)
  51.             }
  52.             rids.Store(key, result)
  53.             wg.Done()
  54.         }(key)
  55.     }
  57.     wg.Wait()
  58.     rids.Range(func(key, value interface{}) bool {
  59.         k := key.(string)
  60.         res := value.([]hstream.AppendResult)
  61.         for i := 0; i < 100; i++ {
  62.             resp, err := res[i].Ready()
  63.             if err != nil {
  64.                 log.Printf("write error: %s\n", err.Error())
  65.             }
  66.             log.Printf("[key: %s]: record[%d]=%s\n", k, i, resp.String())
  67.         }
  68.         return true
  69.     })
  71.     return nil
  72. }