Get started with data reading and writing

Prerequisites

1. You have activated the DataHub service and created a project as described in Preparations.

2. Create a topic

The schema for the test topic is as follows:

Field Name	Field Type
field1	STRING
field2	SMALLINT
field3	BOOLEAN
field4	TINYINT
field5	INTEGER
field6	BIGINT
field7	FLOAT
field8	DOUBLE
field9	TIMESTAMP
field10	DECIMAL
field11	JSON

For more information, see Topic operations.

Write example

Dependencies

<!-- For zero trust credentials -->
<dependency>
	<groupId>com.aliyun</groupId>
	<artifactId>credentials-java</artifactId>
	<version>1.0.2</version>
</dependency>

<dependency>
	<groupId>com.aliyun.datahub</groupId>
	<artifactId>datahub-client-library</artifactId>
	<version>1.4.11</version>
</dependency>

Write example (asynchronous write)

Replace endpoint, projectName, and topicName with your information.

public static void main(String[] args) throws InterruptedException {
	// Obtain AccessKey information from an environment variable.
	EnvironmentVariableCredentialProvider provider = EnvironmentVariableCredentialProvider.create();

	String endpoint ="https://dh-cn-hangzhou.aliyuncs.com";
	String projectName = "test_project";
	String topicName = "test_topic";

	// Initialize the producer. The default configurations are used here.
	ProducerConfig config = new ProducerConfig(endpoint, provider);
	DatahubProducer producer = new DatahubProducer(projectName, topicName, config);

	RecordSchema schema = producer.getTopicSchema();
	// If you enable multi-version schemas, you can also obtain the schema of a specific version.
	// RecordSchema schema = producer.getTopicSchema(3);

	// For asynchronous writes, you can register a callback function as needed.
	WriteCallback callback = new WriteCallback() {
		@Override
		public void onSuccess(String shardId, List<RecordEntry> records, long elapsedTimeMs, long sendTimeMs) {
			System.out.println("write success");
		}

		@Override
		public void onFailure(String shardId, List<RecordEntry> records, long elapsedTimeMs, DatahubClientException e) {
			System.out.println("write failed");
		}
	};

	for (int i = 0; i < 10000; ++i) {
		try {
            // generate data by schema
            TupleRecordData data = new TupleRecordData(schema);
    				data.setField("field1", "hello");
    				data.setField("field2", 12);
    				data.setField("field3", Boolean.parseBoolean("true"));
    				data.setField("field4", 12);
    				data.setField("field5", 103);
    				data.setField("field6", 12);
    				data.setField("field8", 5.0);
    				data.setField("field9", Long.parseLong("124455677"));
    				data.setField("field10", BigDecimal.valueOf(Long.valueOf(240134)));
    				data.setField("field11", "    {\n"
						+ "        \"name\":\"Water Margin\",\n"
						+ "        \"price\":19.9\n"
						+ "    }");
            RecordEntry recordEntry = new RecordEntry();
            recordEntry.setRecordData(data);

            producer.sendAsync(recordEntry, callback);
            // If you do not need to check whether the data was sent successfully, do not register a callback. Just send the data directly.
            // producer.sendAsync(recordEntry, null);
        } catch (DatahubClientException e) {
            // TODO: Handle the exception. This is usually a non-retriable error or an error that occurs after the maximum number of retries is reached.
            Thread.sleep(1000);
        }
	}

	// Ensure that all data is sent before the program exits.
	producer.flush(true);
	producer.close();
}

DataHub supports multiple write methods. For more information, see SDK introduction.

View data write status

After the write operation is complete, go to the topic details page and click the Shard Data tab to view the write status for each shard.

View metrics

On the topic details page, click the Metric Statistics tab to view write-related metrics such as QPS, RPS, throughput, and network latency.

For more information, see View metrics.

Read example

Dependencies

<!-- For zero trust credentials -->
<dependency>
	<groupId>com.aliyun</groupId>
	<artifactId>credentials-java</artifactId>
	<version>1.0.2</version>
</dependency>

<dependency>
	<groupId>com.aliyun.datahub</groupId>
	<artifactId>datahub-client-library</artifactId>
	<version>1.4.11</version>
</dependency>

Read Example

Replace endpoint, projectName, topicName, and subId with the actual values.

public static void main(String[] args) throws InterruptedException {
    // Obtain AccessKey information from an environment variable.
    EnvironmentVariableCredentialProvider provider = EnvironmentVariableCredentialProvider.create();

    String endpoint = "https://dh-cn-hangzhou.aliyuncs.com";
    String projectName = "test_project";
    String topicName = "test_topic";
    String subId = "";

    ConsumerConfig config = new ConsumerConfig(endpoint, provider);
    DatahubConsumer consumer = new DatahubConsumer(projectName, topicName, subId, config);

    while (true) {
        RecordEntry recordEntry = null;
        try {
            recordEntry = consumer.read(5000);
            if (recordEntry != null) {
                TupleRecordData data = (TupleRecordData) recordEntry.getRecordData();
                // handle data
  					System.out.println("read record: " + data.getField("field1") + ", " + data.getField("field2") + ", " + 
							data.getField("field3") + ", " + data.getField("field4") +
							", " + data.getField("field5") + ", " +data.getField("field6") +"," +
							data.getField("field7") +data.getField("field8") +"," 
							+ data.getField("field9") +"," +data.getField("field10"));            }
        } catch (DatahubClientException e) {
            // TODO: Handle the exception. This is usually a non-retriable error or an error that occurs after the maximum number of retries is reached.
        }
    }
}

View checkpoints

On the topic details page, click the Subscription List tab.

You can view the consumption progress and latency of each shard.

View metrics

On the topic details page, click the Metric Statistics tab to view read-related metrics such as QPS, RPS, throughput, and network latency.

For more information, see View metrics.

DataHub supports multiple read methods. For more information, see SDK introduction.

CloudMonitor

DataHub supports alerts through CloudMonitor. You can create alert rules to monitor DataHub metrics and trigger alerts. Supported metrics include readMetric (read), writeMetric (write), and subscription consumption. You can use CloudMonitor to create alerts for read and write metrics. For more information, see CloudMonitor alerts.

​

​