创建和管理 Stream

创建和管理 Stream

命名资源准则

一个HStream资源的名称可以唯一地识别一个 HStream 资源，如一个 stream、 subscription 或 reader。资源名称必须符合以下要求:

以一个字母开头
长度必须不超过255个字符
只包含以下字符。字母[A-Za-z]，数字[0-9]。破折号-，下划线_。

*用于资源名称作为SQL语句的一部分的情况。例如在 HStream SQL Shell 中或者用 SQL 创建 IO 任务时，将会出现资源名称无法被正确解析的情况（如与关键词冲突），此时需要用户用反斜线 ` ，括住资源名称。这个限制或将会在日后的版本中被改进移除。

Stream 的属性

Replication factor

为了容错性和更高的可用性，每个 Stream 都可以在集群中的节点之间进行复制。一个常用的生产环境 Replication factor 配置是为 3，也就是说，你的数据总是有三个副本，这在出错或你想对 Server 进行维护时将会很有帮助。这种复制是以 Stream 为单位上进行的。 - Backlog Retention

该配置控制 HStreamDB 的 Stream 中的 records 被写入后保留的时间。当超过 retention 保留的时间后，HStreamDB 将会清理这些 records，不管它是否被消费过。

默认值=7 天
最小值=1 秒
最大值=21 天

创建一个 stream

在你写入 records 或者创建一个订阅之前先创建一个 stream。

1. // CreateStreamExample.java

3. package docs.code.examples;

5. import io.hstream.HStreamClient;

7. public class CreateStreamExample {
8. public static void main(String[] args) throws Exception {
9. // TODO(developer): Replace these variables before running the sample.
10. String serviceUrl = "127.0.0.1:6570";
11. if (System.getenv("serviceUrl") != null) {
12. serviceUrl = System.getenv("serviceUrl");
13. }

15. String streamName1 = "your_h_records_stream_name";
16. String streamName2 = "your_raw_records_stream_name";

18. HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
19. createStreamExample(client, streamName1);
20. createStreamWithAttrsExample(client, streamName2);
21. client.close();
22. }

24. public static void createStreamExample(HStreamClient client, String streamName) {
25. client.createStream(streamName);
26. }

28. public static void createStreamWithAttrsExample(HStreamClient client, String streamName) {
29. client.createStream(
30. streamName,
31. (short) 1 // replication factor
32. ,
33. 10 // Number of shards
34. ,
35. 7 * 24 * 3600 // backlog retention time in seconds
36. );
37. }
38. }

1. // ExampleCreateStream.go

3. package examples

5. import (
6. "log"

8. "github.com/hstreamdb/hstreamdb-go/hstream"
9. )

11. func ExampleCreateStream() 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. // Create a stream, only specific streamName
19. if err = client.CreateStream("testDefaultStream"); err != nil {
20. log.Fatalf("Creating stream error: %s", err)
21. }

23. // Create a new stream with 1 replica, 5 shards, set the data retention to 1800s.
24. err = client.CreateStream("testStream",
25. hstream.WithReplicationFactor(1),
26. hstream.EnableBacklog(1800),
27. hstream.WithShardCount(5))
28. if err != nil {
29. log.Fatalf("Creating stream error: %s", err)
30. }

32. return nil
33. }

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 create_stream(client):
21. await client.create_stream(
22. stream_name, replication_factor=1, backlog=24 * 60 * 60, shard_count=1
23. )

删除一个 Stream

只有当一个 Stream 没有所属的订阅时才允许被删除，除非传一个强制标删除的 flag 。

强制删除一个 Stream

如果你需要删除一个有订阅的 stream 时，请启用强制删除。在强制删除一个 stream 后，原来 stream 的订阅仍然可以从 backlog 中读取数据。这些订阅的 stream 名字会变成 __deleted_stream__。同时，我们并不允许在被删除的 stream 上创建新的订阅，也不允许向该 stream 写入新的 record。

1. // DeleteStreamExample.java

3. package docs.code.examples;

5. import io.hstream.HStreamClient;

7. public class DeleteStreamExample {
8. public static void main(String[] args) throws Exception {
9. // TODO(developer): Replace these variables before running the sample.
10. // String serviceUrl = "your-service-url-address";
11. String serviceUrl = "127.0.0.1:6570";
12. if (System.getenv("serviceUrl") != null) {
13. serviceUrl = System.getenv("serviceUrl");
14. }

16. String streamName1 = "your_h_records_stream_name";
17. String streamName2 = "your_raw_records_stream_name";

19. HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
20. deleteStreamExample(client, streamName1);
21. deleteStreamForceExample(client, streamName2);
22. client.close();
23. }

25. public static void deleteStreamExample(HStreamClient client, String streamName) {
26. client.deleteStream(streamName);
27. }

29. public static void deleteStreamForceExample(HStreamClient client, String streamName) {
30. client.deleteStream(streamName, true);
31. }
32. }

1. // ExampleDeleteStream.go

3. package examples

5. import (
6. "github.com/hstreamdb/hstreamdb-go/hstream"
7. "log"
8. )

10. func ExampleDeleteStream() error {
11. client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
12. if err != nil {
13. log.Fatalf("Creating client error: %s", err)
14. }
15. defer client.Close()

17. // force delete stream and ignore none exist stream
18. if err := client.DeleteStream("testStream",
19. hstream.EnableForceDelete,
20. hstream.EnableIgnoreNoneExist); err != nil {
21. log.Fatalf("Deleting stream error: %s", err)
22. }

24. if err := client.DeleteStream("testDefaultStream"); err != nil {
25. log.Fatalf("Deleting stream error: %s", err)
26. }

28. return nil
29. }

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 delete_stream(client):
21. await client.delete_stream(stream_name, ignore_non_exist=True, force=True)

列出所有 stream 信息

可以如下拿到所有 HStream 中的 stream:

1. // ListStreamsExample.java

3. package docs.code.examples;

5. import io.hstream.HStreamClient;
6. import io.hstream.Stream;
7. import java.util.List;

9. public class ListStreamsExample {
10. public static void main(String[] args) throws Exception {
11. // TODO(developer): Replace these variables before running the sample.
12. String serviceUrl = "127.0.0.1:6570";
13. if (System.getenv("serviceUrl") != null) {
14. serviceUrl = System.getenv("serviceUrl");
15. }

17. HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
18. listStreamExample(client);
19. client.close();
20. }

22. public static void listStreamExample(HStreamClient client) {
23. List<Stream> streams = client.listStreams();
24. for (Stream stream : streams) {
25. System.out.println(stream.getStreamName());
26. }
27. }
28. }

1. // ExampleListStreams.go

3. package examples

5. import (
6. "fmt"
7. "github.com/hstreamdb/hstreamdb-go/hstream"
8. "log"
9. )

11. func ExampleListStreams() 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. streams, err := client.ListStreams()
19. if err != nil {
20. log.Fatalf("Listing streams error: %s", err)
21. }

23. for _, stream := range streams {
24. fmt.Printf("%+v\n", stream)
25. }

27. return nil
28. }

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 list_streams(client):
21. ss = await client.list_streams()
22. for s in ss:
23. print(s)