DataHub cost saving guide

Maven dependency

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

Code example

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.
	ProducerConfig config = new ProducerConfig(endpoint, provider);
	DatahubProducer producer = new DatahubProducer(projectName, topicName, config);

	RecordSchema schema = producer.getTopicSchema();
	// If a multi-version schema is enabled, you can also get the schema of a specified version.
	// RecordSchema schema = producer.getTopicSchema(3);

	// For an asynchronous write, 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", 1234);
            RecordEntry recordEntry = new RecordEntry();
            recordEntry.setRecordData(data);

            producer.sendAsync(recordEntry, callback);
            // If you do not need to confirm whether the data was sent successfully, do not register a callback. Send the data directly.
            // producer.sendAsync(recordEntry, null);
        } catch (DatahubClientException e) {
            // TODO: Handle exceptions, which are typically non-retriable errors or errors that occur after the maximum number of retries.
            Thread.sleep(1000);
        }
	}

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

go.mod dependency

require (
	github.com/aliyun/aliyun-datahub-sdk-go v1.1.0
)

Code example

In versions 1.1.0 and later, the default serialization protocol is Batch. No special configuration is needed. You can write data as usual, using either synchronous or asynchronous writes.

func handleSuccessRun(producer datahub.AsyncProducer) {
	for suc := range producer.Successes() {
		// handle request success
		fmt.Printf("shard:%s, rid:%s, records:%d, latency:%v\n",
			suc.ShardId, suc.RequestId, len(suc.Records), suc.Latency)
	}
}

func handleFailedRun(producer datahub.AsyncProducer) {
	// handle request failed
	for err := range producer.Errors() {
		fmt.Printf("shard:%s, records:%d, latency:%v, error:%v\n",
			err.ShardId, len(err.Records), err.Latency, err.Err)
	}
}

func main() {
	cfg := datahub.NewProducerConfig()

    // Obtain AccessKey information from an environment variable.
	credential, err := credentials.NewCredential(nil)
	if err != nil {
		fmt.Println(err)
		// TODO: handle error
	}
	cfg.Account = datahub.NewCredentialAccount(credential)
	cfg.Endpoint = "https://dh-cn-hangzhou.aliyuncs.com"
	cfg.Project = "test_project"
	cfg.Topic = "test_topic"

	producer := datahub.NewAsyncProducer(cfg)
	err = producer.Init()

	if err != nil {
		// TODO: handle error
		fmt.Println(err)
	}

	schema, err := producer.GetSchema()
	if err != nil {
		// TODO: handle error
		fmt.Println(err)
	}

	// Handle the success channel.
	go handleSuccessRun(producer)

	// Handle the error channel.
	go handleFailedRun(producer)

	// Generate data in a loop.
	for i := 0; i < 1000; i++ {
		record := datahub.NewTupleRecord(schema)
        // Each time you set a value, check if the operation was successful.
		err = record.SetValueByName("f1", "val1")
		if err != nil {
			fmt.Println(err)
			return
		}

		err = record.SetValueByName("f2", 1234)
		if err != nil {
			fmt.Println(err)
			return
		}

		producer.Input() <- record
	}

	err = producer.Close()
	if err != nil {
		fmt.Println(err)
	}
}