Manage topics-DataHub(DataHub)-阿里云帮助中心

This topic describes how to perform topic-related operations using the DataHub Java SDK.

Topic overview

A topic is the smallest unit for data subscription and publishing in DataHub. You can use a topic to represent a type or category of streaming data. DataHub currently supports two types of topics: Blob and Tuple.

A Blob topic allows you to write a block of binary data as a record.
A Tuple topic supports structured records similar to database rows. Each record consists of multiple columns. A record schema must be specified because data is transmitted over the network in string format, and the schema is required to parse it into the corresponding data types.

Supported Data Types:

Type	Description	Range
BIGINT	8-byte signed integer	`-9223372036854775808 – 9223372036854775807`
DOUBLE	8-byte double-precision float	`-1.0 _10^308 – 1.0 _10^308`
BOOLEAN	Boolean type	Can be any of the following pairs: True/False true/false 0/1
TIMESTAMP	Timestamp with microsecond precision	-
STRING	UTF-8 encoded string	Max 2 MB per column
TINYINT	1-byte signed integer	-128 – 127
SMALLINT	2-byte signed integer	`-32768 – 32767`
INTEGER	4-byte signed integer	`-2147483648 – 2147483647`
FLOAT	4-byte single-precision float	`-3.40292347_10^38 – 3.40292347_10^38`

Note

TINYINT, SMALLINT, INTEGER, and FLOAT are supported starting from Java SDK version 2.16.1-public.

Create a topic

Create a Tuple topic

Parameters

Name	Type	Description
projectName	String	The project name.
topicName	String	The topic name.
shardCount	int	The initial number of shards.
lifeCycle	int	Data retention period (in days).
recordType	RecordType	Type of records. Only `TUPLE` and `BLOB` are supported.
recordSchema	RecordSchema	Schema definition for the Topic.
comment	String	Description of the Topic.

Exceptions

Class name	Error code	Description
DatahubClientException	-	The base class for all exceptions.

Code example

 public static void createTupleTopic(String projectName, String topicName, int shardCount, int lifeCycle,  String topicComment) {
   RecordSchema schema = new RecordSchema();
   schema.addField(new Field("bigint_field", FieldType.BIGINT));
   schema.addField(new Field("double_field", FieldType.DOUBLE));
   schema.addField(new Field("boolean_field", FieldType.BOOLEAN));
   schema.addField(new Field("timestamp_field", FieldType.TIMESTAMP));
   schema.addField(new Field("tinyint_field", FieldType.TINYINT));
   schema.addField(new Field("smallint_field", FieldType.SMALLINT));
   schema.addField(new Field("integer_field", FieldType.INTEGER));
   schema.addField(new Field("float_field", FieldType.FLOAT));
   schema.addField(new Field("decimal_field", FieldType.DECIMAL));
   schema.addField(new Field("string_field", FieldType.STRING));
   try {
       datahubClient.createTopic(projectName,topicName, shardCount, lifeCycle, RecordType.TUPLE, schema, topicComment);
       System.out.println("create topic successful");
   } catch (DatahubClientException e) {
        System.out.println(e.getErrorMessage());
   }
 }

Create a Blob Topic

Parameters

Name	Type	Description
projectName	String	The project name.
topicName	String	The topic name.
shardCount	int	The initial number of shards.
lifeCycle	int	Data retention period (in days).
recordType	RecordType	Type of records. Only `TUPLE` and `BLOB` are supported.
comment	String	Description of the Topic.

Exceptions

Class name	Error code	Description
DatahubClientException	-	The base class for all exceptions.
InvalidParameterException	`InvalidParameter` `InvalidCursor`	Invalid parameters.
AuthorizationFailureException	`Unauthorized`	Authorization signature parsing failed. Check AccessKey.
ResourceNotFoundException	`ResourceNotFound` `NoSuchProject` `NoSuchTopic` `NoSuchShard` `NoSuchSubscription` `NoSuchConnector` `NoSuchMeteringInfo`	The requested resource does not exist. Note After performing a Split or Merge operation, this exception might be thrown if other requests are sent immediately.
ResourceAlreadyExistException	`ResourceAlreadyExist` `ProjectAlreadyExist` `TopicAlreadyExist` `ConnectorAlreadyExist`	The resource already exists. This exception is thrown if the resource already exists when you attempt to create it.

Code example

public static void createBlobTopic(String projectName, String topicName, int shardCount, int lifeCycle,  String topicComment) {
  try {
      datahubClient.createTopic(projectName, blobTopicName, shardCount, lifeCycle, RecordType.BLOB, topicComment);
      System.out.println("create topic successful");
  } catch (DatahubClientException e) {
        System.out.println(e.getErrorMessage());
  }

Delete a topic

Before deleting a Topic, ensure that no subscriptions or connectors are associated with it. Otherwise, a NoPermissionException will be thrown.

Parameters

Name	Type	Description
projectName	String	The project name.
topicName	String	The topic name.

Exception description

Class name

Error code

Description

DatahubClientException

The base class for all exceptions.

NoPermissionException

NoPermission

OperationDenied

No permission. This is usually caused by incorrect RAM configuration or authorization.

Code example

public static void deleteTopic(String projectName, String topicName) {
  try {
      datahubClient.deleteTopic(projectName, topicName);
      System.out.println("delete topic successful");
  } catch (DatahubClientException e) {
        System.out.println(e.getErrorMessage());

  }
}

List topics

Retrieves all topics under the specified project.

Parameters

Name	Type	Description
projectName	String	The project name.

Code example

   public static void listTopic(String projectName ) {
      try {
          ListTopicResult listTopicResult = datahubClient.listTopic(projectName);
          if (listTopicResult.getTopicNames().size() > 0) {
              for (String tName : listTopicResult.getTopicNames()) {
                  System.out.println(tName);
              }
          }
      } catch (DatahubClientException e) {
        System.out.println(e.getErrorMessage());          
      }
  }

Update a topic

Updates the lifecycle or description of a Topic.

Parameters

Name	Type	Description
projectName	String	The project name.
topicName	String	The topic name.
comment	String	The topic description.
lifeCycle	int	The topic lifecycle.

Exceptions

Class name	Error code	Description
DatahubClientException	-	The base class for all exceptions.

Code example

   public static void updateTopic(String projectName, String topicName, int lifeCycle, String comment) {
        try {
            comment = "new topic comment";
             lifeCycle = 1;
            datahubClient.updateTopic(projectName, Constant.topicName,lifeCycle, comment);
            System.out.println("update topic successful");
            //View the updated result
            GetTopicResult getTopicResult = datahubClient.getTopic(projectName, topicName);
            System.out.println(getTopicResult.getComment());
        } catch (DatahubClientException e) {
            System.out.println(e.getErrorMessage());
        }
    }

Query a topic

Retrieves topic metadata by project name and topic name.

Parameters

Name	Type	Description
projectName	String	The project name.
topicName	String	The topic name.

Exceptions

Class name	Error code	Description
DatahubClientException	-	The base class for all exceptions.

Code example

   public static void getTopic(String projectName, String topicName) {
        try {
            GetTopicResult getTopicResult = datahubClient.getTopic(projectName, topicName);
            System.out.println(getTopicResult.getShardCount() + "\t"
                    + getTopicResult.getLifeCycle() + "\t"
                    + getTopicResult.getRecordType() + "\t"
                    + getTopicResult.getComment());
        } catch (DatahubClientException e) {
            System.out.println(e.getErrorMessage());

        }
    }

More operations

Add a field to a Tuple topic

You can add one or more fields to a Tuple topic.

Parameters

Name	Type	Description
projectName	String	The project name.
topicName	String	The Topic name.
fields	Field	The new field to add. Null values are not allowed.

Exceptions

Class name	Error code	Description
DatahubClientException	-	The base class for all exceptions.

Code example

public static void appendNewField(String projectName,String topicName) {
    try {
        Field newField = new Field("newField", FieldType.STRING, true,"comment");
        datahubClient.appendField(projectName, topicName, newField);
        System.out.println("append field successful");
    } catch (DatahubClientException e) {
        System.out.println(e.getErrorMessage());

    }
}