Tablestore SDK for Java を使用すると、配信タスクを作成して、データテーブルから同一リージョン内の OSS バケットに、完全データ、増分データ、またはその両方を配信できます。
前提条件
開始する前に、次の前提条件が満たされていることを確認してください。
Tablestore SDK for Java をインストールし、クライアントを初期化します。データ配信には、バージョン 5.10.3 以降が必要です。最新バージョンのご使用を推奨します。
Object Storage Service (OSS) を有効化し、Tablestore インスタンスと同じリージョンに OSS バケットを作成します。詳細については、「OSS の概要」をご参照ください。
サービスリンクロール
AliyunServiceRoleForOTSDataDeliveryを作成し、ロールの ARN を取得します。詳細については、「配信タスクの作成」をご参照ください。
概要
createDeliveryTask を呼び出して配信タスクを作成します。タスクでは、完全データまたは増分データのみを配信することも、最初に完全データを配信してから継続的に増分データを配信することもできます。
public CreateDeliveryTaskResponse createDeliveryTask(CreateDeliveryTaskRequest request)
throws TableStoreException, ClientException配信タスクを作成した後、タスクを初期化する必要があります。describeDeliveryTask を呼び出して、配信タスク情報をクエリできます。
次のサンプルでは、完全データと増分データを配信し、Tablestore へのデータ書き込み時間に基づいて日単位の OSS パーティションを作成するタスクを作成します。このサンプルでは、client は初期化済みのクライアントです。プレースホルダーを実際の値に置き換え、pk、event_time、active 列がそれぞれ String 型、String 型、Boolean 型であることを確認してください。
String tableName = "<TABLE_NAME>";
String taskName = "<TASK_NAME>";
OSSTaskConfig taskConfig = new OSSTaskConfig();
taskConfig.setOssPrefix("delivery/year=$yyyy/month=$MM/day=$dd");
taskConfig.setOssBucket("<OSS_BUCKET>");
taskConfig.setOssEndpoint("<OSS_ENDPOINT>");
taskConfig.setOssStsRole("<ROLE_ARN>");
taskConfig.addParquetSchema(new ParquetSchema("pk", "pk", DataType.UTF8));
taskConfig.addParquetSchema(
new ParquetSchema("event_time", "event_time", DataType.UTF8));
taskConfig.addParquetSchema(new ParquetSchema("active", "active", DataType.BOOL));
CreateDeliveryTaskRequest request =
new CreateDeliveryTaskRequest(tableName, taskName, taskConfig);
request.setTaskType(DeliveryTaskType.BASE_INC);
client.createDeliveryTask(request);パラメーター
配信リクエスト
request は CreateDeliveryTaskRequest 型で、次のパラメーターが含まれます。
|
名前 |
型 |
説明 |
|
tableName (必須) |
String |
データテーブルの名前。 |
|
taskName (必須) |
String |
配信タスクの名前。名前には、小文字、数字、ハイフン ( |
|
taskConfig (必須) |
OSSTaskConfig |
OSS 配信設定。 |
|
taskType (必須) |
DeliveryTaskType |
配信タスクのタイプ。有効な値： |
OSS 配信設定
request.taskConfig は OSSTaskConfig 型で、次のパラメーターが含まれます。
|
名前 |
型 |
説明 |
|
ossPrefix (必須) |
String |
OSS バケット内のディレクトリプレフィックス。 |
|
ossBucket (必須) |
String |
OSS バケットの名前。バケットは Tablestore インスタンスと同じリージョンにある必要があります。 |
|
ossEndpoint (必須) |
String |
OSS バケットが配置されているリージョンのエンドポイント。 |
|
ossStsRole (必須) |
String |
|
|
parquetSchema (必須) |
List<ParquetSchema> |
配信するフィールド。フィールドを選択し、OSS での名前と順序をカスタマイズできます。要素の順序によって、Parquet ファイル内のフィールドの順序が決まります。 |
|
eventTimeColumn (オプション) |
EventColumn |
イベント時間列。このパラメーターを指定すると、 |
|
format (オプション) |
OSSFileFormat |
OSS ファイル形式。デフォルト値であり、唯一サポートされている値は |
|
timeFormatter (オプション) |
TimeFormatter |
予約済みのパーティションフォーマットパラメーター。現在の SDK はリクエストにこのパラメーターを含めません。設定しないでください。 |
配信される列
request.taskConfig.parquetSchema[] の各要素は ParquetSchema 型で、次のパラメーターが含まれます。
|
名前 |
型 |
説明 |
|
columnName (必須) |
String |
Tablestore データテーブル内のソースフィールドの名前。 |
|
ossColumnName (必須) |
String |
フィールドが OSS に配信された後のフィールドの名前。 |
|
type (必須) |
DataType |
Parquet ファイル内のフィールドの送信先タイプ。このタイプは、ソースフィールドのデータの型と一致する必要があります。一致しない場合、フィールド値はダーティデータとして破棄されます。詳細については、「データ型のマッピング」をご参照ください。 |
|
encode (オプション) |
OSSFileEncoding |
Parquet エンコーディング。デフォルト値： |
|
typeExtend (オプション) |
String |
予約済みの Parquet 拡張タイプパラメーター。このパラメーターはサポートされていません。設定しないでください。 |
イベント時間列
request.taskConfig.eventTimeColumn は EventColumn 型で、次のパラメーターが含まれます。
|
名前 |
型 |
説明 |
|
columnName (必須) |
String |
イベント時間として使用されるソースフィールドの名前。 |
|
timeFormat (必須) |
EventTimeFormat |
イベント時間のフォーマット。有効な値： |
例
イベント時間によるデータパーティション
event_time 列に基づいて OSS パーティションを作成するには、リクエストを作成する前に taskConfig でイベント時間列を設定します。次のサンプルでは、列の値は RFC 3339 に準拠している必要があります。
EventColumn eventColumn =
new EventColumn("event_time", EventTimeFormat.RFC3339);
taskConfig.setEventTimeColumn(eventColumn);