DataHub：DataHub成本節省攻略

Maven依賴

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

程式碼範例：

public static void main(String[] args) throws InterruptedException {
	// 通過環境變數擷取AK資訊
	EnvironmentVariableCredentialProvider provider = EnvironmentVariableCredentialProvider.create();

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

	// 初始化Producer，這裡直接使用預設配置
	ProducerConfig config = new ProducerConfig(endpoint, provider);
	DatahubProducer producer = new DatahubProducer(projectName, topicName, config);

	RecordSchema schema = producer.getTopicSchema();
	// 如果開啟了多version schema，這裡也可以擷取指定version的schema
	// RecordSchema schema = producer.getTopicSchema(3);

	// 對於非同步寫入，可以根據需要來選擇是否註冊回呼函數
	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);
            // 如果不需要關心資料是否發送成功，那麼就不需要註冊回調，直接發送
            // producer.sendAsync(recordEntry, null);
        } catch (DatahubClientException e) {
            // TODO 處理異常，一般是不可重試錯誤或者超過重試次數;
            Thread.sleep(1000);
        }
	}

	// 保證退出前，資料全部被發送完
	producer.flush(true);
	producer.close();
}

go.mod依賴

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

程式碼範例：

在 1.1.0 及以後的版本中，預設序列化協議已經設定為 Batch，無需使用者做任何特殊配置，只需要正常的寫入即可，同步和非同步寫入均可以。

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()

    // 通過環境變數擷取AK資訊
	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)
	}

	// 處理success channle
	go handleSuccessRun(producer)

	// 處理error channle
	go handleFailedRun(producer)

	// 迴圈產生資料
	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
		}

		producer.Input() <- record
	}

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