阿里OTS表格存储单挑与批量多行插入--javaSDK

你的名字 2022-06-13 11:08 82阅读 0赞

@Autowired

private SyncClient syncClient;

\----------------------------单条写入-------------------------------

/\*\*  
\* ots单条写入  
\*/  
       PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();  
        primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_CONNECTOR\_ID, PrimaryKeyValue.fromString(info.getConnectorID()));  
        primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_OPERATOR\_ID, PrimaryKeyValue.fromString(operatorId));  
        primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_STATION\_ID, PrimaryKeyValue.fromString(stationId));  
        primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_DATA\_TIME, PrimaryKeyValue.fromLong(dataTime));  
       PrimaryKey primaryKey = primaryKeyBuilder.build();  
        RowPutChange rowPutChange = new RowPutChange(AliTSConfigs.TB\_NAME\_CONNECTOR\_RT, primaryKey);  
        rowPutChange.addColumn(new Column(AliTSConfigs.TB\_COLUMN\_STATUS, ColumnValue.fromLong(info.getStatus())));  
        rowPutChange.addColumn(new Column(AliTSConfigs.TB\_COLUMN\_PARKSTATUS, ColumnValue.fromLong(info.getParkStatus())));  
        rowPutChange.addColumn(new Column(AliTSConfigs.TB\_COLUMN\_LOCKSTATUS, ColumnValue.fromLong(info.getLockStatus())));  
        PutRowRequest request = new PutRowRequest();  
        rowPutChange.setCondition(new Condition(RowExistenceExpectation.EXPECT\_NOT\_EXIST));   
        request.setRowChange(rowPutChange);  
        syncClient.putRow(request);

![Center][]

\----------------------------批量写入\-------------------------------

/\*\*  
\* ots批量操作  
\*/  
BatchWriteRowRequest request = new BatchWriteRowRequest();  
try \{   
  for (int i = 0; i < list.size(); i++) \{   
      PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();  
 primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_CONNECTOR\_ID, PrimaryKeyValue.fromString(list.get(i).getConnectorID()));  
 primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_OPERATOR\_ID, PrimaryKeyValue.fromString(operatorId));  
 primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_STATION\_ID, PrimaryKeyValue.fromString(stationId));  
   primaryKeyBuilder.addPrimaryKeyColumn(AliTSConfigs.TB\_PK\_NAME\_DATA\_TIME, PrimaryKeyValue.fromLong(dataTime));  
      PrimaryKey primaryKey = primaryKeyBuilder.build();  
  
RowPutChange rowPutChange = new RowPutChange(AliTSConfigs.TB\_NAME\_CONNECTOR\_RT, primaryKey);  
rowPutChange.addColumn(new Column(AliTSConfigs.TB\_COLUMN\_STATUS, ColumnValue.fromLong(list.get(i).getStatus())));  
        rowPutChange.addColumn(new Column(AliTSConfigs.TB\_COLUMN\_PARKSTATUS, ColumnValue.fromLong(list.get(i).getParkStatus())));  
      rowPutChange.addColumn(new Column(AliTSConfigs.TB\_COLUMN\_LOCKSTATUS, ColumnValue.fromLong(list.get(i).getLockStatus())));  
       // 构造插入数据的请求对象，  
       request.addRowChange(rowPutChange);  
   \}  
  
// batchWriteRow接口会返回一个结果集 结果集中包含的结果个数与插入的行数相同,结果集中的结果不一定都是成功，用户需要自己对不成功的操作进行重试。  
BatchWriteRowResponse  response  = syncClient.batchWriteRow(request);  
// 结果异常处理  
if (!response.isAllSucceed()) \{  
               for (BatchWriteRowResponse.RowResult rowResult : response.getFailedRows()) \{  
               log.error("失败的行:" + request.getRowChange(rowResult.getTableName(), rowResult.getIndex()).getPrimaryKey());  
               log.error("失败原因:" + rowResult.getError().toString());  
              \}\}

// 重试逻辑类似于上面的batchGetRow示例中的逻辑，这里省略

\} catch (Exception ex) \{  
   ex.printStackTrace();  
   log.error("Batch write row failed.");  
   \}

图片看着更清晰![Center 1][]

[Center]: /images/20220613/7ba0a93f4b794d78a19e4685efa68951.png
[Center 1]: /images/20220613/6c87fcc2862a40b795606c7bccd665bf.png