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 {
	// 環境変数から AccessKey 情報を取得します。
	EnvironmentVariableCredentialProvider provider = EnvironmentVariableCredentialProvider.create();

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

	// プロデューサーを初期化します。デフォルト設定が使用されます。
	ProducerConfig config = new ProducerConfig(endpoint, provider);
	DatahubProducer producer = new DatahubProducer(projectName, topicName, config);

	RecordSchema schema = producer.getTopicSchema();
	// マルチバージョン スキーマが有効な場合は、指定したバージョンのスキーマも取得できます。
	// 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 {
            // スキーマに従ってデータを生成します。
            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() {
		// リクエスト成功の処理
		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) {
	// リクエスト失敗の処理
	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()

    // 環境変数から AccessKey 情報を取得します。
	credential, err := credentials.NewCredential(nil)
	if err != nil {
		fmt.Println(err)
		// TODO: エラー処理
	}
	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: エラー処理
		fmt.Println(err)
	}

	schema, err := producer.GetSchema()
	if err != nil {
		// TODO: エラー処理
		fmt.Println(err)
	}

	// 成功チャネルの処理
	go handleSuccessRun(producer)

	// エラーチャネルの処理
	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)
	}
}