Consume incremental data

Use the Tablestore SDK for Go and Stream APIs to consume incremental changes such as row writes, updates, and deletions.

Prerequisites

Install the Tablestore SDK for Go and initialize the client.
Stream is enabled for the table. You can configure StreamSpec when you create the table or call UpdateTable to update the Stream settings.

Description

Stream organizes incremental table changes into shards. Use the following workflow:

Call ListStream to obtain the Stream ID of the target table.
Call DescribeStream to obtain the Stream status and shard list.
Call GetShardIterator for each shard to obtain a read iterator.
Call GetStreamRecord to pull incremental records, and use the returned NextShardIterator to continue reading.

func (client *TableStoreClient) ListStream(req *ListStreamRequest) (*ListStreamResponse, error)
func (client *TableStoreClient) DescribeStream(req *DescribeStreamRequest) (*DescribeStreamResponse, error)
func (client *TableStoreClient) GetShardIterator(req *GetShardIteratorRequest) (*GetShardIteratorResponse, error)
func (client TableStoreClient) GetStreamRecord(req *GetStreamRecordRequest) (*GetStreamRecordResponse, error)

The following sample obtains the first shard of the example_table table and reads a batch of incremental records. In production, iterate over and continuously consume all shards.

tableName := "example_table"
listResponse, err := client.ListStream(
    &tablestore.ListStreamRequest{TableName: &tableName},
)
if err != nil {
    log.Fatal(err)
}
streamID := listResponse.Streams[0].Id

describeResponse, err := client.DescribeStream(
    &tablestore.DescribeStreamRequest{StreamId: streamID},
)
if err != nil {
    log.Fatal(err)
}
shardID := describeResponse.Shards[0].SelfShard

iteratorResponse, err := client.GetShardIterator(
    &tablestore.GetShardIteratorRequest{
        StreamId: streamID,
        ShardId:  shardID,
    },
)
if err != nil {
    log.Fatal(err)
}

recordResponse, err := client.GetStreamRecord(
    &tablestore.GetStreamRecordRequest{
        ShardIterator: iteratorResponse.ShardIterator,
    },
)
if err != nil {
    log.Fatal(err)
}

for _, record := range recordResponse.Records {
    fmt.Println(record)
}
nextIterator := recordResponse.NextShardIterator
fmt.Println(nextIterator)

Parameters

List streams

ListStreamRequest contains the following parameter.

Name	Type	Description
TableName (optional)	`*string`	The table name. If this parameter is not specified, all tables in the current instance that have Stream enabled are returned. If specified, only the target table is returned.

Describe a Stream

DescribeStreamRequest contains the following parameters.

Name	Type	Description
StreamId (required)	`*StreamId`	The Stream ID returned by `ListStream`.
InclusiveStartShardId (optional)	`*ShardId`	The first shard ID to return.
ShardLimit (optional)	`*int32`	The maximum number of shards to return.

Obtain a shard iterator

GetShardIteratorRequest contains the following parameters.

Name	Type	Description
StreamId (required)	`*StreamId`	The Stream ID.
ShardId (required)	`*ShardId`	The shard ID returned by `DescribeStream`.
Timestamp (optional)	`*int64`	The read start timestamp, in microseconds. If this parameter is not specified, reading starts from the beginning of the shard.
Token (optional)	`*string`	The token used to continue obtaining a read iterator.

Read incremental records

GetStreamRecordRequest contains the following parameters.

Name	Type	Description
ShardIterator (required)	`*ShardIterator`	The shard read iterator returned by `GetShardIterator` or the previous `GetStreamRecord` call.
Limit (optional)	`*int32`	The maximum number of incremental records to return.
TableName (optional)	`*string`	The table to which the shard belongs.

Response

Stream list

ListStreamResponse.Streams is of the []Stream type. Each element contains a Stream ID, table name, and creation time.

Stream information

DescribeStreamResponse contains the following business information.

Field	Type	Description
`StreamId`	`*StreamId`	The Stream ID.
`TableName`	`*string`	The table name.
`CreationTime`	`int64`	The Stream creation time, in microseconds.
`ExpirationTime`	`int32`	The retention period of incremental logs, in hours.
`Status`	`StreamStatus`	The Stream status.
`Shards`	`[]*StreamShard`	The shards returned by this call.
`NextShardId`	`*ShardId`	The start shard ID for the next query. A value of `nil` indicates that all shards have been returned.

Shard iterator

GetShardIteratorResponse.ShardIterator is the iterator used for the first read from the target shard.

Incremental records

GetStreamRecordResponse contains the following business information.

Field	Type	Description
`Records`	`[]*StreamRecord`	The incremental records returned by this call.
`NextShardIterator`	`*ShardIterator`	The iterator for the next read. A value of `nil` indicates that the current shard has been fully read.
`MayMoreRecord`	`*bool`	Indicates whether the current shard may contain more records.

Examples

Continuously read incremental data

The following sample uses NextShardIterator to continuously read a shard.

iterator := iteratorResponse.ShardIterator
for iterator != nil {
    response, err := client.GetStreamRecord(
        &tablestore.GetStreamRecordRequest{ShardIterator: iterator},
    )
    if err != nil {
        log.Fatal(err)
    }

    for _, record := range response.Records {
        fmt.Println(record)
    }
    iterator = response.NextShardIterator
}