在下面的方法中，我从SAP HANA中获取数据，在这里，我使用地图来保持列表对象，基于 tbl_guid 作为关键 dRListMap 再传给另一个方法 fetchUniquForTblGuid 为执行。
public static void gdprDeleteReqStatus() {
LOGGER.info("Fetching the records GDPR_DEL_REQ_STATUS in HANA");
String dbName = hanaProp.getProperty("database");
int mysqlMergeLimit=Integer.parseInt(hanaProp.getProperty("mysql.limit"));
String sql = String.format("select * from %s .GDPR_DEL_REQ_STATUS ", dbName);
Map<String, List<DeletedRecord>> dRListMap = new HashMap<>();
CommonService commonObject=new CommonService();
try (Statement stmt = hanaConnection.createStatement(); ResultSet rs = stmt.executeQuery(sql)) {
int i=0;
//Thread.sleep(10000);
while ((rs.next())) {
DeletedRecord delRecord = new DeletedRecord(rs.getString(1), rs.getString(2), rs.getString(3),
rs.getString(4), rs.getDate(5));
String key = rs.getString(2);
List<DeletedRecord> recordList = dRListMap.get(key) == null ? new ArrayList<>() : dRListMap.get(key);
recordList.add(delRecord);
dRListMap.put(key, recordList);
i++;
if (i ==mysqlMergeLimit) {
//LOGGER.info(String.format("HANA batch size %s and records %s ",i,dRListMap.toString()));
LOGGER.info("List Size "+dRListMap.values().size());
commonObject.fetchUniquForTblGuid(dRListMap);
dRListMap.clear();
i=0;
}
}if(i>0) {
//LOGGER.info(String.format("HANA batch size %s and records %s ",i,dRListMap.toString()));
commonObject.fetchUniquForTblGuid(dRListMap);
}
} catch (Exception ee) {
LOGGER.error("Exception occurred while fetching the records from GDPR_DEL_REQ_STATUS", ee);
}
}
问题说明 reqHistMap 的组合的地图。Date 和 Req_id 相隔 # 键值 List<String>迭代的同时 dRLs 列表 key 是由 Date 和 req_id 需要添加到 reqHistMap 如果密钥已经存在，那么需要用以下方法修改同一列表 tbl_guid所以在打印这张地图的时候就发现了问题。reqHistMap 每个列表的预期大小是10，但我发现它们也少于10。
public void fetchUniquForTblGuid(Map<String, List<DeletedRecord>> dRListMap) {
LOGGER.info("Fetching the unique seq for each recieved quid from delete_status_table");
List<List<DeletedRecord>> valueList = Collections.list(Collections.enumeration(dRListMap.values()));
List<String> values = valueList.stream().flatMap(Collection::stream).map(DeletedRecord::getTblGuid)
.collect(Collectors.toCollection(ArrayList::new));
Map<String, String> tblSeqMap =new HashMap<>();
tblSeqMap.put("TRNFRM_ECC_CCM.DWEPLOY_LOOKUP", "1");
tblSeqMap.put("REPLICN_DYLAN.BILLING_INFO", "3");
tblSeqMap.put("TRNFRM_SUBSCRPN.SUBSCRIPTION_FILTER", "2");
tblSeqMap.put("REPLICN_ETS.STAGE_USER_LVT_PROFILE_PARSED","4");
//getTheUniqueNoForGuids(values);
// System.out.println(dRListMap);
for (Map.Entry<String, String> map : tblSeqMap.entrySet()) {
if (dRListMap.containsKey(map.getKey())) {
dRListMap.get(map.getKey()).forEach((DeletedRecord del) -> del.setTblGuid(map.getValue()));
}
}
// System.out.println(dRListMap);
List<List<DeletedRecord>> withUpdatedGuid = Collections.list(Collections.enumeration(dRListMap.values()));
List<DeletedRecord> dRLs = withUpdatedGuid.stream().flatMap(Collection::stream)
.collect(Collectors.toCollection(ArrayList::new));
Map<String, List<String>> reqHistMap = new HashMap<>();
dRLs.parallelStream().forEach(deleteRecord -> {
String key = String.format("%s#%s", deleteRecord.getReqDts(), deleteRecord.getReqId());
List<String> value = reqHistMap.get(key) == null ? new ArrayList<>() : reqHistMap.get(key);
value.add(deleteRecord.getTblGuid());
reqHistMap.put(key, value);
});
List<RequestTableMapping> finalList = reqHistMap.entrySet().parallelStream().map(entry -> {
String[] key = entry.getKey().split("#");
return new RequestTableMapping(key[1], key[0], entry.getValue());
}).collect(Collectors.toCollection(ArrayList::new));
HbaseDao hDao=new HbaseDao();
finalList.stream().forEach(x->{
LOGGER.info(String.format("Request id %s and no. of guid's %s",x.getRequestId
(),x.getTableGuidSmallMapping().size()));
});
// hDao.insertRecords(finalList, true);
//System.out.println(finalList);
reqHistMap.clear();
}
这个POJO需要保存在Hbase中。
public class RequestTableMapping {
public String requestId;
public String date;
List<String> tableGuidSmallMapping;
public RequestTableMapping() {
super();
}
public RequestTableMapping(String requestId, String date, List<String> tableGuidSmallMapping) {
super();
this.requestId = requestId;
this.date = date;
this.tableGuidSmallMapping = tableGuidSmallMapping;
}
public String getRequestId() {
return requestId;
}
public void setRequestId(String requestId) {
this.requestId = requestId;
}
public String getDate() {
return date;
}
public void setDate(String date) {
this.date = date;
}
public List<String> getTableGuidSmallMapping() {
return tableGuidSmallMapping;
}
public void setTableGuidSmallMapping(List<String> tableGuidSmallMapping) {
this.tableGuidSmallMapping = tableGuidSmallMapping;
}
@Override
public String toString() {
return "RequestTableMapping {requestId:" + requestId + ", date:" + date + ", tableGuidSmallMapping:"
+ tableGuidSmallMapping + "}";
}
}
产出。
2020-05-18 17:56:18 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:56:18 INFO HanaService:54 - List Size 10
2020-05-18 17:56:18 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:56:18 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of `guid's 9`
2020-05-18 17:56:18 INFO HanaService:54 - List Size 10
2020-05-18 17:56:18 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:56:18 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of `guid's 7`
2020-05-18 17:56:18 INFO HanaService:54 - List Size 10
2020-05-18 17:56:18 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:56:18 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:56:18 INFO HanaService:54 - List Size 10
预期。
2020-05-18 17:52:42 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:42 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:42 INFO HanaService:54 - List Size 10
2020-05-18 17:52:42 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:42 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:42 INFO HanaService:54 - List Size 10
2020-05-18 17:52:42 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:42 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:43 INFO HanaService:54 - List Size 10
2020-05-18 17:52:43 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:43 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:43 INFO HanaService:54 - List Size 10
2020-05-18 17:52:43 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:43 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:43 INFO HanaService:54 - List Size 10
2020-05-18 17:52:43 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:43 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:43 INFO HanaService:54 - List Size 10
2020-05-18 17:52:43 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:43 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:43 INFO HanaService:54 - List Size 10
2020-05-18 17:52:43 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:43 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:43 INFO HanaService:54 - List Size 10
2020-05-18 17:52:43 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:43 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-18 17:52:43 INFO HanaService:54 - List Size 10
2020-05-18 17:52:43 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-18 17:52:43 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
我能够在上面的程序中找出问题，所以基本上我是使用并行流来处理的。dRLs 列表中，由于该元素没有按顺序处理，我在 reqHistMap 为价值。希望对大家有所帮助。
dRLs.stream().forEach(deleteRecord -> {
String key = String.format("%s#%s", deleteRecord.getReqDts(), deleteRecord.getReqId());
List<String> value = reqHistMap.get(key) == null ? new ArrayList<>() : reqHistMap.get(key);
value.add(deleteRecord.getTblGuid());
reqHistMap.put(key, value);
});
预期产出。
2020-05-21 13:21:42 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:42 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:42 INFO HanaService:54 - List Size 10
2020-05-21 13:21:42 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:42 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:42 INFO HanaService:54 - List Size 10
2020-05-21 13:21:42 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:42 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:22 INFO HanaService:54 - List Size 10
2020-05-21 13:21:22 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:22 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:22 INFO HanaService:54 - List Size 10
2020-05-21 13:21:22 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:22 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:22 INFO HanaService:54 - List Size 10
2020-05-21 13:21:22 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:22 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:22 INFO HanaService:54 - List Size 10
2020-05-21 13:21:22 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:22 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:22 INFO HanaService:54 - List Size 10
2020-05-21 13:21:22 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:22 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:22 INFO HanaService:54 - List Size 10
2020-05-21 13:21:22 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:22 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10
2020-05-21 13:21:22 INFO HanaService:54 - List Size 10
2020-05-21 13:21:22 INFO CommonService:37 - Fetching the unique seq for each recieved quid from delete_status_table
2020-05-21 13:21:22 INFO CommonService:84 - Request id 11E8D1EE51D64AB598CACF259031C1DF and no. of guid's 10