Java SDK を使用した配信タスクの作成 - Tablestore - Alibaba Cloud ドキュメントセンター

Tablestore データテーブルから OSS バケットへデータを配信するには、CreateDeliveryTask 操作を使用して配信タスクを作成します。

重要

データレイク配信機能には、Tablestore SDK for Java 5.10.3 以降が必要です。正しい SDK バージョンがインストールされていることを確認してください。詳細については、「Java SDK リリースノート」をご参照ください。

前提条件

OSS を有効化し、ご利用の Tablestore インスタンスと同じリージョンにバケットを作成します。詳細については、「OSS の有効化」をご参照ください。
コンソールで Tablestore のサービスリンクロールを作成し、その ARN を記録します。詳細については、「配信タスクの作成」をご参照ください。

サービスリンクロールの ARN を取得するには、RAM コンソールで以下の手順を実行します。

Resource Access Management (RAM) コンソールの ロール ページで、AliyunServiceRoleForOTSDataDelivery を検索し、ロール名をクリックします。ロールの詳細ページで、ARN を表示してコピーします。
OTSClient を初期化します。詳細については、「Tablestore クライアントの初期化」をご参照ください。
データテーブルを作成し、データを書き込みます。

パラメーター

パラメーター	説明
tableName	テーブルの名前です。
taskName	配信タスクの名前です。名前は 3～16 文字で、小文字、数字、ハイフン (-) のみを使用できます。先頭および末尾は小文字または数字である必要があります。
taskConfig	配信タスクの構成です。有効な値は次のとおりです。 ossPrefix：バケット内のフォルダのプレフィックスです。Tablestore からこのフォルダにデータが配信されます。送信先フォルダのパスでは、次の時間変数がサポートされています：$yyyy、$MM、$dd、$HH、$mm。パスで時間変数を使用して配信を行う場合、データの書き込み時刻に基づいて OSS フォルダが動的に生成されます。これにより、Hive がデータをパーティション分割する際の命名規則に従ってデータがパーティション分割されます。OSS 内のオブジェクトは、時間に基づいて整理・パーティション分割・分散されます。パスで時間変数を使用しない場合、すべてのファイルは特定のプレフィックスを含む名前の OSS フォルダに配信されます。 ossBucket：OSS バケットの名前です。 ossEndpoint：OSS バケットがデプロイされているリージョンのサービスアドレスです。 ossStsRole：Tablestore サービスリンクロールの Alibaba Cloud リソースネーム (ARN) です。 format：配信データのフォーマットです。配信データは Parquet フォーマットで保存されます。デフォルトでは、配信サービスはすべての型のデータを PLAIN でエンコードします。 eventTimeColumn：イベント時間カラムです。このパラメーターは、カラムの時刻に基づいてデータをパーティション分割することを指定します。このパラメーターを指定しない場合、データは Tablestore への書き込み時刻に基づいてパーティション分割されます。 parquetSchema：配信対象のカラムを指定します。配信するには、ソースフィールド、送信先フィールド、送信先フィールドのデータの型を設定する必要があります。 Tablestore 内でのフィールドの並び順は、スキーマ内のフィールドの並び順と異なる場合があります。OSS に保存される Parquet データは、スキーマ内のフィールドの並び順に基づいて分散されます。重要ソースフィールドと送信先フィールドのデータの型は一致している必要があります。フィールド間でデータの型が一致しない場合、そのフィールドはダーティデータとして破棄されます。フィールドの型マッピングの詳細については、「クイックスタート」トピックの「データの型マッピング」セクションをご参照ください。
taskType	データ配信モードです。デフォルト値は BASE_INC です。有効な値は次のとおりです： INC：増分データ配信モードです。増分データのみが同期されます。 BASE：完全データ配信モードです。テーブル内のすべてのデータがスキャンされ、同期されます。 BASE_INC：差分データ配信モードです。完全データの同期後、Tablestore は増分データを同期します。増分モードでデータを同期する場合、前回のデータ配信時刻および現在の配信タスクのステータスを確認できます。

例

以下の例は、データテーブルの配信タスクを作成する方法を示しています。

private static void createDeliveryTask(SyncClient client) {
    String tableName = "<TABLE_NAME>";
    String taskName = "<TASK_NAME>";
    OSSTaskConfig taskConfig = new OSSTaskConfig();
    taskConfig.setOssPrefix("sampledeliverytask/year=$yyyy/month=$MM");
    taskConfig.setOssBucket("datadeliverytest");
    taskConfig.setOssEndpoint("oss-cn-hangzhou.aliyuncs.com");
    taskConfig.setOssStsRole("acs:ram::17************45:role/aliyunserviceroleforotsdatadelivery");
    // eventTimeColumn パラメーターはオプションです。イベント時間パーティション分割に使用するカラムを指定します。
    // このパラメーターを設定しない場合、データは Tablestore への書き込み時刻に基づいてパーティション分割されます。
    EventColumn eventColumn = new EventColumn("PK1", EventTimeFormat.RFC1123);
    taskConfig.setEventTimeColumn(eventColumn);
    taskConfig.addParquetSchema(new ParquetSchema("PK1", "PK1", DataType.UTF8));
    taskConfig.addParquetSchema(new ParquetSchema("PK2", "PK2", DataType.BOOL));
    taskConfig.addParquetSchema(new ParquetSchema("Col1", "Col1", DataType.UTF8));
    CreateDeliveryTaskRequest request = new CreateDeliveryTaskRequest();
    request.setTableName(tableName);
    request.setTaskName(taskName);
    request.setTaskConfig(taskConfig);
    request.setTaskType(DeliveryTaskType.BASE_INC);
    CreateDeliveryTaskResponse response = client.createDeliveryTask(request);
    System.out.println("requestID: "+ response.getRequestId());
    System.out.println("traceID: " + response.getTraceId());
    System.out.println("create delivery task success");
}