Go high-level SDK

Introduction

The Go high-level software development kit (SDK) currently supports only producers. Support for consumers is not yet available. This document will be updated when consumer support becomes available.

Identity authentication

An AccessKey pair is an access credential provided by Alibaba Cloud for identity verification when you call Alibaba Cloud OpenAPI. An AccessKey pair consists of an AccessKey ID and an AccessKey secret. You must keep your AccessKey pair secure. A leaked AccessKey pair threatens the security of all cloud resources under your account. If you hard-code a plaintext AccessKey pair in your code, the AccessKey pair can be leaked because of improper permission management of your code repository.

Alibaba Cloud Credentials is an identity credential management tool for Alibaba Cloud developers. After you configure the default credential chain, you no longer need to hard-code a plaintext AccessKey pair in your code to call Alibaba Cloud OpenAPI. This effectively secures the cloud resources under your account.

Prerequisites

• Obtain the AccessKey ID and AccessKey secret for a Resource Access Management (RAM) user account. For more information, see View the AccessKey information of a RAM user.

Configuration methods

All examples in this topic retrieve AccessKey pair information by configuring environment variables. For more information about other methods, see Manage access credentials.

Configuration

Configure the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables.

Configure on Linux and macOS

Run the following commands:

export ALIBABA_CLOUD_ACCESS_KEY_ID=<access_key_id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<access_key_secret>

Replace <access_key_id> with your AccessKey ID and <access_key_secret> with your AccessKey secret.

Configure on Windows

1. Create an environment variable file. In the file, add the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables and set their values to your AccessKey ID and AccessKey secret.

2. Restart your Windows operating system.

Code example

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

Producer overview

The Go SDK supports synchronous and asynchronous write methods. The following sections describe both methods.

The dependencies are listed below. For the latest version, visit GitHub.

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

Parameters

Asynchronous write (recommended)

Asynchronous writes allow you to delegate data batching to the SDK. You only need to monitor the callback channel to handle successful and failed writes. This greatly simplifies the process.

func handleSuccessRun(producer datahub.AsyncProducer) {
	for suc := range producer.Successes() {
		// handle request success
		fmt.Printf("Shard: %s, Request ID: %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()
	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)
        // Check whether the value is set successfully each time
		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)
	}
}

Hash-based write

To preserve the order of your data, you must hash it based on a consistent key. This ensures that data with the same hash value is written to the same shard, where its order is guaranteed. For these hash-based writes, we recommend using the asynchronous method.

func handleSuccessRun(producer datahub.AsyncProducer) {
	for suc := range producer.Successes() {
		// handle request success
		fmt.Printf("Shard: %s, Request ID: %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()
	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)
        // Check whether the value is set successfully each time
		err = record.SetValueByName("f1", "val1")
		if err != nil {
			fmt.Println(err)
			return
		}

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

        // Set a hash value for the record
        record.SetPartitionKey("pk-val")
		producer.Input() <- record
	}

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

Synchronous write

For synchronous writes, data must be batched. A batch size of 512 KB to 1 MB is recommended.

func main() {
	cfg := datahub.NewProducerConfig()
	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.NewProducer(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)
	}

	// Generate data in a loop. We recommend a batch size of 512 KB to 1 MB.
	records := make([]datahub.IRecord, 0)
	for i := 0; i < 1000; i++ {
		record := datahub.NewTupleRecord(schema)
		err = record.SetValueByName("f1", "val1")
		if err != nil {
			fmt.Println(err)
			return
		}

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

		records = append(records, record)
	}

	shardId, err := producer.Send(records)
	if err != nil {
		fmt.Printf("Write failed. Error: %v\n", err)
	} else {
		fmt.Printf("Write successful. Shard ID: %v\n", shardId)
	}

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