Antes de usar un kit de desarrollo de software (SDK) para el envío de datos, revise las notas, las interfaces y demás información sobre la función de entrega de data lake. Una vez que cree una tarea de entrega, Tablestore envía automáticamente los datos de la tabla de datos a un bucket de Object Storage Service (OSS) para su almacenamiento.
Notas de uso
El envío de data lake está disponible en las siguientes regiones: China (Hangzhou), China (Shanghai), China (Beijing), China (Zhangjiakou) y China (Shenzhen).
La entrega de datos ignora las operaciones de eliminación. Los datos que elimine en Tablestore no se eliminan del bucket de OSS de destino.
Inicializar una nueva tarea de entrega puede tardar hasta un minuto.
-
Con una tasa de escritura estable, la latencia de sincronización suele ser inferior a 3 minutos. La latencia P99 es menor a 10 minutos.
NotaLa latencia P99 corresponde al percentil 99 de la latencia. Esto significa que el 99 % de las solicitudes son más rápidas que este valor.
Requisitos previos
-
Complete las siguientes operaciones en Object Storage Service:
Active Object Storage Service (OSS) y cree un bucket en la misma región que su instancia de Tablestore. Para obtener más información, consulte Activate OSS.
NotaLa entrega de data lake admite el envío de datos a cualquier bucket de OSS ubicado en la misma región que la instancia de Tablestore. Si necesita enviar datos a otros almacenes de datos, como MaxCompute, submit a ticket.
-
A continuación se detallan las operaciones que debe completar en el servicio Tablestore:
Obtenga el endpoint de la instancia desde la página de detalles de la instancia. Para obtener más información, consulte Endpoints.
Cree una tabla de datos de Tablestore. Para obtener más información, consulte Create a data table.
-
Ejecute las siguientes operaciones en el servicio Resource Access Management (RAM):
-
Cree un usuario de Resource Access Management (RAM) y otórguele el permiso AliyunOTSFullAccess para gestionar Tablestore. Para obtener más información, consulte Create a RAM user y Manage the permissions of a RAM user.
AvisoSi el AccessKey de su cuenta de Alibaba Cloud se filtra, todos sus recursos en la nube quedan expuestos. Le recomendamos usar el AccessKey de un usuario de RAM para reducir los riesgos de seguridad.
Cree un AccessKey para el usuario de RAM. Para obtener más información, consulte Create an AccessKey.
-
Configure las credenciales de acceso. Para obtener más información, consulte Configure access credentials.
Interfaces
|
Interfaz |
Descripción |
|
CreateDeliveryTask |
Crea una tarea de entrega. |
|
ListDeliveryTask |
Enumera todas las tareas de entrega de una tabla de datos. |
|
DescribeDeliveryTask |
Consulta la descripción de una tarea de entrega. |
|
DeleteDeliveryTask |
Elimina una tarea de entrega. |
Parámetros
|
Parámetro |
Descripción |
|
tableName |
Nombre de la tabla de datos. |
|
taskName |
Nombre de la tarea de entrega. El nombre solo puede contener letras minúsculas (a-z), dígitos y guiones (-). Debe comenzar y terminar con una letra minúscula o un dígito. Su longitud debe estar entre 3 y 16 caracteres. |
|
taskConfig |
Configuración de la tarea de entrega. Incluye las siguientes opciones:
|
|
taskType |
Tipo de la tarea de entrega. Incluye las siguientes opciones:
|
Uso
Puede usar Java SDK o Go SDK para implementar la función de entrega de data lake. En esta sección se toma el SDK de Java como ejemplo para describir las operaciones de entrega de data lake.
En el siguiente ejemplo se muestra cómo crear una tarea de entrega para una tabla de datos.
import com.alicloud.openservices.tablestore.ClientException;
import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.TableStoreException;
import com.alicloud.openservices.tablestore.model.delivery.*;
public class DeliveryTask {
public static void main(String[] args) {
// Replace yourInstanceName with the name of your instance.
final String instanceName = "yourInstanceName";
// Replace yourEndpoint with the endpoint of your instance.
final String endPoint = "yourEndpoint";
// Get the AccessKey ID and AccessKey secret from environment variables.
final String accessKeyId = System.getenv("TABLESTORE_ACCESS_KEY_ID");
final String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
SyncClient client = new SyncClient(endPoint, accessKeyId, accessKeySecret, instanceName);
try {
createDeliveryTask(client);
System.out.println("end");
} catch (TableStoreException e) {
System.err.println("Operation failed. Details: " + e.getMessage() + e.getErrorCode() + e.toString());
System.err.println("Request ID:" + e.getRequestId());
} catch (ClientException e) {
System.err.println("Request failed. Details: " + e.getMessage());
} finally {
client.shutdown();
}
}
private static void createDeliveryTask(SyncClient client){
String tableName = "sampleTable";
String taskName = "sampledeliverytask";
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 is an optional configuration. It specifies partitioning based on the time in a specific data column. If this parameter is not set, data is partitioned based on the time it is written to Tablestore.
EventColumn eventColumn = new EventColumn("Col1", 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");
}
}