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