shard操作_数据总线 DataHub(DataHub)-阿里云帮助中心

本文为您展示DataHub的 GO SDK的Shard操作。

Shard说明

Shard表示对一个Topic进行数据传输的并发通道，每个Shard会有对应的ID。每个Shard会有多种状态：

Opening：为启动中状态。
Active：为启动完成可服务状态。

每个Shard启用以后会占用一定的服务端资源，建议按需申请Shard数量。

说明

Shard可以进行合并和分裂，当数据量增大时，可以采用分裂Shard来增加数据通道，提高数据写入的并发量，当数据量减小时，应该合并shard减少服务器资源浪费。例如淘宝在双11期间，数据量骤增，这个时候每个Shard的写入压力过大，便可以增加Shard提高写入效率，在双11过后，数据量明显降低，则需要合并Shard。

列出Shard

参数说明

参数名	参数类型	参数说明
projectName	String	项目名称。
topicName	string	Topic名称

返回示例

1. type SplitShardResult struct {
2.     NewShards []ShardEntry `json:"NewShards"`
3. }

错误说明

错误类名	错误码	错误说明
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	访问的资源不存在（注：进行Split/Merge操作后，立即发送其他请求，有可能会抛出该异常）。
AuthorizationFailedError	`Unauthorized`	Authorization 签名解析异常，检查AK是否填写正确。
DatahubClientError	-	其他所有，并且是所有异常的基类。
InvalidParameterError	`InvalidParameter` `InvalidCursor`	非法参数。

代码示例

func ExampleDataHub_ListShard() {
    ls, err := dh.ListShard(projectName, topicName)
    if err != nil {
        fmt.Println("get shard list failed")
        fmt.Println(err)
        return
    }
    fmt.Println("get shard list successful")
    for _, shard := range ls.Shards {
        fmt.Println(shard)
    }
}

分裂Shard

参数说明

参数名	参数类型	参数说明
projectName	String	项目名称。
topicName	string	Topic名称。
shardId	string	The shard which to split
splitKey	string	The split key which is used to split shard

返回示例

type SplitShardResult struct {
    NewShards []ShardEntry `json:"NewShards"`
}

错误说明

错误类名	错误码	错误说明
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	访问的资源不存在（注：进行Split/Merge操作后，立即发送其他请求，有可能会抛出该异常）。
AuthorizationFailedError	`Unauthorized`	Authorization 签名解析异常，检查AK是否填写正确。
DatahubClientError	-	其他所有，并且是所有异常的基类。
InvalidParameterError	`InvalidParameter` `InvalidCursor`	非法参数。

代码示例

func ExampleDataHub_SplitShard() {
    // the shardId of you want to split
    shardId := "0"
    ss, err := dh.SplitShard(projectName, topicName, shardId)
    if err != nil {
        fmt.Println("split shard failed")
        fmt.Println(err)
        return
    }
    fmt.Println("split shard successful")
    fmt.Println(ss)
    // After splitting, you need to wait for all shard states to be ready
    // before you can perform related operations.
    dh.WaitAllShardsReady(projectName, topicName)
}

合并Shard

参数说明

参数名	参数类型	参数说明
projectName	String	项目名称。
topicName	string	Topic名称。
shardId	string	The shard which will be merged
adjacentShardId	string	The adjacent shard of the specified shard.

返回示例

type SplitShardResult struct {
    NewShards []ShardEntry `json:"NewShards"`
}

错误说明

错误类名	错误码	错误说明
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	访问的资源不存在（注：进行Split/Merge操作后，立即发送其他请求，有可能会抛出该异常）。
InvalidOperationError	-	-
AuthorizationFailedError	`Unauthorized`	Authorization 签名解析异常，检查AK是否填写正确。
InvalidParameterError	`InvalidParameter` `InvalidCursor`	非法参数
ShardSealedError	-	-

代码示例

func ExampleDataHub_MergeShard() {
    shardId := "3"
    adjacentShardId := "4"
    ms, err := dh.MergeShard(projectName, topicName, shardId, adjacentShardId)
    if err != nil {
        fmt.Println("merge shard failed")
        fmt.Println(err)
        return
    }
    fmt.Println("merge shard successful")
    fmt.Println(ms)
    // After splitting, you need to wait for all shard states to be ready
    // before you can perform related operations.
    dh.WaitAllShardsReady(projectName, topicName)
}