DataHub:DataHub cost saving guide

Dependência do Maven

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

Exemplo de código

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

Dependência do go.mod

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

Exemplo de código

Nas versões 1.1.0 e posteriores, o protocolo de serialização padrão é o Batch. Nenhuma configuração especial é necessária. Você pode gravar dados normalmente, usando gravações síncronas ou assíncronas.

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