Importación masiva de datos sin conexión - Tablestore - Centro de documentación de Alibaba Cloud

El Tablestore SDK for Java importa datos sin conexión mediante la inserción, actualización o eliminación de varias filas, y devuelve el resultado de cada operación por fila.

Requisitos previos

Instale Tablestore SDK for Java e inicialice el cliente.

Descripción

Llame a bulkImport para enviar varias operaciones de fila en una sola solicitud. El servidor procesa cada fila de forma independiente y devuelve resultados individuales, por lo que un error en una fila no afecta las operaciones de las demás. Este método está diseñado para importaciones masivas sin conexión en escenarios de big data.

public BulkImportResponse bulkImport(BulkImportRequest bulkImportRequest) throws TableStoreException, ClientException

A continuación se muestra un ejemplo que inserta cinco filas en la tabla bulk_import_demo mediante una única solicitud masiva e imprime el recuento de filas exitosas y con errores.

String tableName = "bulk_import_demo";

BulkImportRequest request = new BulkImportRequest(tableName);

List<RowChange> rowChanges = new ArrayList<RowChange>();
for (int i = 0; i < 5; i++) {
    PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
            .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("row" + i))
            .build();
    RowPutChange put = new RowPutChange(tableName, pk);
    put.addColumn(new Column("col1", ColumnValue.fromString("v" + i)));
    rowChanges.add(put);
}
request.addRowChanges(rowChanges);

BulkImportResponse response = client.bulkImport(request);

// Retrieve per-row results into succeedRows and failedRows
List<BulkImportResponse.RowResult> succeedRows = new ArrayList<BulkImportResponse.RowResult>();
List<BulkImportResponse.RowResult> failedRows = new ArrayList<BulkImportResponse.RowResult>();
response.getResult(succeedRows, failedRows);

System.out.println("All succeed: " + response.isAllSucceed());
System.out.println("Succeed: " + succeedRows.size() + ", Failed: " + failedRows.size());

Parámetros

Nombre

Tipo

Descripción

tableName (obligatorio)

String

El nombre de la tabla.

rowChanges (obligatorio)

List<RowChange>

La lista de operaciones de fila. Puede combinar las siguientes operaciones en una sola solicitud:

RowPutChange: Escribe una fila. Para más detalles sobre la configuración, consulte Write a row.
RowUpdateChange: Actualiza una fila. Para más detalles sobre la configuración, consulte Update a row.
RowDeleteChange: Elimina una fila. Para más detalles sobre la configuración, consulte Delete a row.

Respuesta

Resultado de la importación

BulkImportResponse contiene el siguiente campo específico de la operación.

Nombre	Tipo	Descripción
rowResults	List<BulkImportResponse.RowResult>	El resultado de cada operación de fila, en el mismo orden que `rowChanges` en la solicitud. Llame a `getRowResults()` para obtener la lista.

Resultado por fila

Cada elemento en BulkImportResponse.rowResults[] es del tipo BulkImportResponse.RowResult.

Nombre	Tipo	Descripción
succeed	boolean	Indica si la operación de fila fue exitosa. Llame a `isSucceed()` para obtener el valor.
error	Error	El error correspondiente a una operación de fila con errores. El valor es `null` cuando la operación se completa correctamente.
index	int	El índice de la operación de fila dentro de `rowChanges` de la solicitud.

Ejemplos

Combinar inserción, actualización y eliminación en una sola solicitud

Utilice un solo objeto BulkImportRequest para enviar operaciones de inserción, actualización y eliminación de forma conjunta.

String tableName = "bulk_import_demo";

BulkImportRequest request = new BulkImportRequest(tableName);

// Insert a new row
PrimaryKey pkPut = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("mixed_put"))
        .build();
RowPutChange put = new RowPutChange(tableName, pkPut);
put.addColumn(new Column("col1", ColumnValue.fromString("put_value")));
request.addRowChange(put);

// Update an existing row (append a column)
PrimaryKey pkUpdate = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("row0"))
        .build();
RowUpdateChange update = new RowUpdateChange(tableName, pkUpdate);
update.put(new Column("col2", ColumnValue.fromLong(100)));
request.addRowChange(update);

// Delete an existing row
PrimaryKey pkDelete = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("row1"))
        .build();
RowDeleteChange delete = new RowDeleteChange(tableName, pkDelete);
request.addRowChange(delete);

BulkImportResponse response = client.bulkImport(request);
System.out.println("Mixed all succeed: " + response.isAllSucceed());

Reintentar filas con errores

Utilice createRequestForRetry(failedRows) para crear una nueva solicitud a partir de las filas con errores. El reintento incluye únicamente dichas filas, por lo que no se vuelven a aplicar las que ya se procesaron correctamente.

String tableName = "bulk_import_demo";

BulkImportRequest request = new BulkImportRequest(tableName);

PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("good"))
        .build();
RowPutChange row = new RowPutChange(tableName, pk);
row.addColumn(new Column("col1", ColumnValue.fromString("ok")));
request.addRowChange(row);

BulkImportResponse response = client.bulkImport(request);

List<BulkImportResponse.RowResult> succeedRows = new ArrayList<BulkImportResponse.RowResult>();
List<BulkImportResponse.RowResult> failedRows = new ArrayList<BulkImportResponse.RowResult>();
response.getResult(succeedRows, failedRows);

// If any rows failed, build a retry request that contains only the failed rows
if (!failedRows.isEmpty()) {
    BulkImportRequest retryRequest = request.createRequestForRetry(failedRows);
    BulkImportResponse retryResponse = client.bulkImport(retryRequest);
    System.out.println("Retry all succeed: " + retryResponse.isAllSucceed());
}