通過Java SDK建立投遞任務 - Tablestore

如果要Table Store資料表中的資料投遞到OSS Bucket中儲存，您可以使用CreateDeliveryTask介面建立一個投遞任務。

重要

Table StoreJava SDK從5.10.3版本開始支援資料湖投遞功能，請確保已安裝正確的Table StoreJava SDK。關於Table StoreJava SDK版本的更多資訊，請參見Java SDK歷史迭代版本。

前提條件

已開通OSS服務且在Table Store執行個體所在地區建立Bucket。具體操作，請參見開通OSS服務。
已通過控制台建立Table Store服務關聯角色並記錄角色的ARN。具體操作，請參見建立投遞任務。
服務關聯角色的ARN請通過RAM控制台擷取，具體操作如下：
在RAM存取控制台的角色介面，搜尋AliyunServiceRoleForOTSDataDelivery後，單擊角色名稱，在角色詳情介面，可以查看和複製角色的ARN資訊。
已初始化OTSClient。具體操作，請參見初始化Tablestore Client。
已建立資料表並寫入資料。

參數

參數	說明
tableName	資料表名稱。
taskName	投遞任務名稱。名稱只能包含英文小寫字母（a~z）、數字和虛線（-），開頭和結尾必須為英文小寫字母或數字，且長度為3~16字元。
taskConfig	投遞任務配置，包括如下選項： ossPrefix：OSS Bucket中的目錄首碼，將Table Store的資料投遞到該OSS Bucket目錄中。投遞路徑中支援引用$yyyy、$MM、$dd、$HH、$mm五種時間變數。當投遞路徑中引用時間變數時，可以按資料的寫入時間動態產生OSS目錄，實現hive partition naming style的資料時間分區，從而按照時間分區組織OSS中的檔案分布。當投遞路徑中不引用時間變數時，所有檔案會被投遞到固定的OSS首碼目錄中。 ossBucket：OSS Bucket名稱。 ossEndpoint：OSS Bucket所在地區的服務地址。 ossStsRole：Table Store服務關聯角色的ARN資訊。 format：投遞的資料的儲存以Parquet列存格式儲存，資料湖投遞預設使用PLAIN編碼方式，PLAIN編碼方式支援任意類型資料。 eventTimeColumn：事件時間列，用於指定按某一列資料的時間進行分區。如果不設定此參數，則按資料寫入Table Store的時間進行分區。 parquetSchema：指定需要投遞的資料列，必須手動設定投遞欄位的源表欄位、目標欄位和目標欄位類型。您可以選擇任意欄位以任意順序、名稱寫入列存檔案，OSS的列存資料會按Schema數組中的資料列先後順序分布。重要投遞資料的欄位類型必須與資料來源的欄位類型匹配，否則會作為髒資料丟棄。欄位類型映射詳情請參見資料格式映射。
taskType	投遞任務的類型，包括如下選項： INC：表示增量資料投遞模式，只同步增量資料。 BASE：表示全量資料投遞模式，一次性全表掃描資料同步。 BASE_INC（預設）：表示全量&增量資料投遞模式，全量資料同步完成後，再同步增量資料其中增量資料同步時可以擷取最新投遞時間和瞭解當前投遞狀態。

樣本

以下樣本用於為資料表建立投遞任務。

private static void createDeliveryTask(SyncClient client) {
    String tableName = "<TABLE_NANE>";
    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"); //eventColumn為可選配置，指定按某一列資料的時間進行分區。如果不設定此參數，則按資料寫入Table Store的時間進行分區。
    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("resquestID: "+ response.getRequestId());
    System.out.println("traceID: " + response.getTraceId());
    System.out.println("create delivery task success");
}