Topic operations - DataHub - Alibaba Cloud Documentation Center

This topic describes the topic operations in the DataHub GO SDK.

Creating a topic

Creating a tuple topic

Data written to a tuple topic has a specific format. You need to specify the record schema. The following data types are supported:

Type	Description	Range
BIGINT	8-byte signed integer	`-9223372036854775807 ~ 9223372036854775807`
DOUBLE	8-byte double-precision floating-point number	`-1.0 _10^308 ~ 1.0 _10^308`
BOOLEAN	The Boolean data type	Valid values: `True/False`. `true/false`. `0/1`.
TIMESTAMP	Timestamp type	Represents a timestamp accurate to microseconds.
STRING	String, only UTF-8 encoding is supported	A single STRING column can be up to 2MB in length.
DECIMAL		s

Parameters

Parameter	Type	Description
projectName	String	The name of the project.
topicName	string	The name of the topic.
comment	string	The description of the topic.
lifeCycle	int	The expire time of the data (Unit: DAY). The data written before that time is not accessible.
recordSchema	RecordSchema	The records schema of this topic.

Errors

Error class	Error code	Description
ResourceExistError	`ResourceAlreadyExist` `ProjectAlreadyExist` `TopicAlreadyExist` `ConnectorAlreadyExist`	The resource already exists (this exception is thrown when you try to create a resource that already exists).
AuthorizationFailedError	`Unauthorized`	Authorization signature parsing exception. Check whether the AccessKey is correct.
DatahubClientError	-	If the above exceptions are excluded, you can usually retry the operation. However, you should limit the number of retries.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	A parameter is set to an invalid value.

Sample code

func Example_CreateTupleTopic(dh datahub.DataHub, projectName, topicName string) {
    recordSchema := datahub.NewRecordSchema()
    recordSchema.AddField(datahub.Field{Name: "bigint_field", Type: datahub.BIGINT, AllowNull: true}).
        AddField(datahub.Field{Name: "timestamp_field", Type: datahub.TIMESTAMP, AllowNull: false}).
        AddField(datahub.Field{Name: "string_field", Type: datahub.STRING}).
        AddField(datahub.Field{Name: "double_field", Type: datahub.DOUBLE}).
        AddField(datahub.Field{Name: "boolean_field", Type: datahub.BOOLEAN})
    if err := dh.CreateTupleTopic(projectName, topicName, "topic comment", 5, 7, recordSchema); err != nil {
        fmt.Println("create topic failed")
        fmt.Println(err)
        return
    }
    fmt.Println("create topic successful")
}

Creating a blob topic

Parameters

Parameter	Type	Description
projectName	String	The name of the project.
topicName	string	The name of the topic.
comment	string	The description of the topic.
lifeCycle	int	The expire time of the data (Unit: DAY). The data written before that time is not accessible.

Errors

Error class	Error code	Description
ResourceExistError	`ResourceAlreadyExist` `ProjectAlreadyExist` `TopicAlreadyExist` `ConnectorAlreadyExist`	The resource already exists (this exception is thrown when you try to create a resource that already exists).
AuthorizationFailedError	`Unauthorized`	Authorization signature parsing exception. Check whether the AccessKey is correct.
DatahubClientError	-	Other errors. This is the base class for all exceptions.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	The error message returned because a parameter is set to an invalid value.

Sample code

func Example_CreateBlobTopic(dh datahub.DataHub, projectName, topicName string) {
    if err := dh.CreateBlobTopic(projectName, topicName, "topic comment", 5, 7); err != nil {
        fmt.Println("create topic failed")
        fmt.Println(err)
        return
    }
    fmt.Println("create topic successful")
}

Deleting a topic

Parameters

Parameter	Type	Description
projectName	String	The name of the project.
topicName	string	The name of the topic.

Errors

Error class	Error code	Description
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The accessed resource does not exist (Note: If you send other requests immediately after performing Split/Merge operations, this exception may be thrown).
AuthorizationFailedError	`Unauthorized`	Authorization signature parsing exception. Check whether the AccessKey is correct.
DatahubClientError		If the above exceptions are excluded, you can usually retry the operation. However, you should limit the number of retries.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	The error message returned because a parameter is set to an invalid value.

Sample code

func ExampleDataHub_DeleteTopic(dh datahub.DataHub, projectName, topicName string) {
    if err := dh.DeleteTopic(projectName, topicName); err != nil {
        fmt.Println("delete failed")
        fmt.Println(err)
        return
    }
    fmt.Println("delete successful")
}

Listing topics

Parameters

Parameter	Type	Description
projectName	String	The name of the project.

Response example

type ListTopicResult struct {
    TopicNames [] string `json:"TopicNames"`
}

Errors

Error class	Error code	Description
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The accessed resource does not exist (Note: If you send other requests immediately after performing Split/Merge operations, this exception may be thrown).
AuthorizationFailedError	`Unauthorized`	Authorization signature parsing exception. Check whether the AccessKey is correct.
DatahubClientError		If the above exceptions are excluded, you can usually retry the operation. However, you should limit the number of retries.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	The error message returned because a parameter is set to an invalid value.

Sample code

func ExampleDataHub_ListTopic(dh datahub.DataHub, projectName, topicName string) {
    lt, err := dh.ListTopic(projectName)
    if err != nil {
        fmt.Println("get topic list failed")
        fmt.Println(err)
        return
    }
    fmt.Println("get topic list successful")
    fmt.Println(lt)
}

Updating a topic

Parameters

Parameter	Type	Description
projectName	String	The name of the project.
topicName	string	The name of the topic.
comment	string	The description of the topic.

Errors

Error class	Error code	Description
ResourceNotFoundError	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The accessed resource does not exist (Note: If you send other requests immediately after performing Split/Merge operations, this exception may be thrown).
AuthorizationFailedError	`Unauthorized`	Authorization signature parsing exception. Check whether the AccessKey is correct.
DatahubClientError		If the above exceptions are excluded, you can usually retry the operation. However, you should limit the number of retries.
InvalidParameterError	`InvalidParameter` `InvalidCursor`	The error message returned because a parameter is set to an invalid value.

Sample code

func ExampleDataHub_UpdateTopic(dh datahub.DataHub, projectName, topicName string) {
    if err := dh.UpdateTopic(projectName, topicName, "new topic comment"); err != nil {
        fmt.Println("update topic comment failed")
        fmt.Println(err)
        return
    }
    fmt.Println("update topic comment successful")
}

More operations

When creating a tuple topic and reading/writing records, you need to use a schema to specify the name and corresponding type of the data storage.

Note

During network transmission, data is sent in string format and needs to be converted to the corresponding type using the schema. A schema is a slice of Field objects. Each Field contains three parameters: the first parameter is the field name, the second parameter is the field type, and the third parameter is a Boolean value. True indicates that the field value can be null, and False indicates that the field value cannot be null.

Getting a schema

For an existing tuple topic, you can use the get_topic interface to obtain schema information. The following is a code example:

func getSchema(dh datahub.DataHub, projectName, topicName string) {
    gt, err := dh.GetTopic(projectName, "topic_test")
    if err != nil {
        fmt.Println("get topic failed")
        fmt.Println(err)
        return
    } else {
        schema := gt.RecordSchema
        fmt.Println(schema)
    }
}

Defining a schema

To create a new tuple topic, you need to define a schema. A schema can be initialized in the following three ways:

Creating a schema directly

func createSchema1() {
    fields := []datahub.Field{
        {"bigint_field", datahub.BIGINT, true},
        {"timestamp_field", datahub.TIMESTAMP, false},
        {"string_field", datahub.STRING, false},
        {"double_field", datahub.DOUBLE, false},
        {"boolean_field", datahub.BOOLEAN, true},
        {"decimal_field", datahub.DECIMAL, false},
    }
    schema := datahub.RecordSchema{
        fields,
    }
    fmt.Println(schema)
}

Initializing a schema step by step

func createSchema2() {
    recordSchema := datahub.NewRecordSchema()
    recordSchema.AddField(datahub.Field{Name: "bigint_field", Type: datahub.BIGINT, AllowNull: true}).
        AddField(datahub.Field{Name: "timestamp_field", Type: datahub.TIMESTAMP, AllowNull: false}).
        AddField(datahub.Field{Name: "string_field", Type: datahub.STRING}).
        AddField(datahub.Field{Name: "double_field", Type: datahub.DOUBLE}).
        AddField(datahub.Field{Name: "boolean_field", Type: datahub.BOOLEAN}).
        AddField(datahub.Field{Name: "decimal_field", Type: datahub.DECIMAL})
}

Defining a schema using a JSON string

func createSchema3() {
    str := ""
    schema, err := datahub.NewRecordSchemaFromJson(str)
    if err != nil {
        fmt.Println("create recordSchema failed")
        fmt.Println(err)
        return
    }
    fmt.Println("create recordSchema successful")
    fmt.Println(schema)
}

Note

The format of the JSON string is: {"fields":[{"type":"BIGINT","name":"a"},{"type":"STRING","name":"b"}]}