向 HStreamDB 中的 Stream 写入 Records

向 HStreamDB 中的 Stream 写入 Records

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

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

如何创建和管理 Stream.
如何通过 Subscription 消费写入 Stream 中的 Records.

为了向 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. }