Write Records to Streams

This document provides information about how to write data to streams in HStreamDB using hstreamdb-java or clients implemented in other languages.

You can also read the following pages to get a more thorough understanding:

To write data to HStreamDB, we need to pack messages as HStream Records and a producer that creates and sends messages to servers.

HStream Record

All data in streams are in the form of an HStream Record. There are two kinds of HStream Record:

  • HRecord: You can think of an hrecord as a piece of JSON data, just like the document in some NoSQL databases.
  • Raw Record: Arbitrary binary data.

End-to-End Compression

To reduce transfer overhead and maximize bandwidth utilization, HStreamDB supports the compression of written HStream records. Users can set the compression algorithm when creating a BufferedProducer. Currently, HStreamDB supports both gzip and zstd compression algorithms. Compressed records are automatically decompressed by the client when they are consumed from HStreamDB.

Write HStream Records

There are two ways to write records to servers. For simplicity, you can use Producer from client.newProducer() to start with. Producers do not provide any configure options, it simply sends records to servers as soon as possible, and all these records are sent in parallel, which means they are unordered. In practice, BufferedProducer from the client.newBufferedProducer() would always be better. BufferedProducer will buffer records in order as a batch and send the batch to servers. When a record is written to the stream, HStream Server will generate a corresponding record id for the record and send it back to the client. The record id is unique in the stream.

Write Records Using a 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)

Write Records Using a Buffered Producer

In almost all scenarios, we would recommend using BufferedProducer whenever possible because it offers higher throughput and provides a very flexible configuration that allows you to adjust between throughput and latency as needed. You can configure the following two settings of BufferedProducer to control and set the trigger and the buffer size. With BatchSetting, you can determine when to send the batch based on the maximum number of records, byte size in the batch and the maximum age of the batch. By configuring FlowControlSetting, you can set the buffer for all records. The following code example shows how you can use BatchSetting to set responding triggers to notify when the producer should flush and FlowControlSetting to limit maximum bytes in a BufferedProducer.

  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()

Write Records with Partition Keys

Partition keys are optional, and if not given, the server will automatically assign a default key. Records with the same partition key can be guaranteed to be written orderly in BufferedProducer.

Another important feature of HStreamDB, sharding, uses these partition keys to decide which shards the record will be allocated to and improve write/read performance. See Manage Shards of a Stream for a more detailed explanation.

You can easily write records with keys using the following example:

  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. }