DQMFBRecordMerger extends MFBRecordMerger { @Override protected Record createNewRecord(Record record1, Record record2, long mergedRecordTimestamp) { RichRecord richRecord1 = (RichRecord) record1; RichRecord richRecord2 = (RichRecord) record2; RichRecord mergedRecord = new RichRecord(record1.getId(), mergedRecordTimestamp, mergedRecordSource); DQAttribute<?>[] mergedRows = richRecord1.getOriginRow().toArray(new DQAttribute<?>[richRecord1.getOriginRow().size()]); List<Attribute> matchingAttributes = record1.getAttributes(); for (int colIdx = 0; colIdx < richRecord1.getOriginRow().size(); colIdx++) { boolean isMatchKeyIndex = false; for (Attribute attribute : matchingAttributes) { if (attribute.getColumnIndex() == colIdx) { isMatchKeyIndex = true; break; } } if (!isMatchKeyIndex) { Map<Integer, SurvivorshipFunction> defaultSurvivorshipFuncs = matchMergeParam.getDefaultSurviorshipRules(); SurvivorshipFunction survivorshipFunc = defaultSurvivorshipFuncs.get(colIdx); if (survivorshipFunc == null || survivorshipFunc.getSurvivorShipAlgoEnum() == null) { continue; } String leftValue = richRecord1.getOriginRow().get(colIdx).getValue(); String rightValue = richRecord2.getOriginRow().get(colIdx).getValue(); Integer referenceColumnIndex = survivorshipFunc.getReferenceColumnIndex(); mergedRows[colIdx] = new DQAttribute<>(StringUtils.EMPTY, colIdx, leftValue, referenceColumnIndex == null ? colIdx : referenceColumnIndex.intValue()); AttributeValues<String> leftValues = richRecord1.getOriginRow().get(colIdx).getValues(); if (leftValues.size() > 0) { mergedRows[colIdx].getValues().merge(leftValues); } else { mergedRows[colIdx].getValues().get(leftValue).increment(); } AttributeValues<String> rightValues = richRecord2.getOriginRow().get(colIdx).getValues(); if (rightValues.size() > 0) { mergedRows[colIdx].getValues().merge(rightValues); } else { mergedRows[colIdx].getValues().get(rightValue).increment(); } if (leftValue == null && rightValue == null) { mergedRows[colIdx].setValue(null); } else { SurvivorShipAlgorithmEnum survAlgo = survivorshipFunc.getSurvivorShipAlgoEnum(); String parameter = survivorshipFunc.getParameter(); String mergedValue = null; String mergedCompareValue = null; switch (survAlgo) { case MOST_RECENT: case MOST_ANCIENT: String leftCompareValue = leftValue; String rightCompareValue = rightValue; if (referenceColumnIndex != null && !referenceColumnIndex.equals(colIdx) && (datePatternMap == null || datePatternMap.get(referenceColumnIndex.toString()) != null)) { leftCompareValue = richRecord1.getOriginRow().get(colIdx).getReferenceValue(); rightCompareValue = richRecord2.getOriginRow().get(colIdx).getReferenceValue(); leftCompareValue = leftCompareValue == null ? richRecord1.getOriginRow().get(referenceColumnIndex).getValue() : leftCompareValue; rightCompareValue = rightCompareValue == null ? richRecord2.getOriginRow().get(referenceColumnIndex).getValue() : rightCompareValue; } else { referenceColumnIndex = colIdx; } mergedCompareValue = compareAsDate(leftCompareValue, rightCompareValue, survAlgo, String.valueOf(referenceColumnIndex), record1.getTimestamp(), record2.getTimestamp()); if (mergedCompareValue != null) { if (mergedCompareValue.equals(leftCompareValue)) { mergedValue = leftValue; } else { mergedValue = rightValue; } } break; default: mergedValue = createMergeValue(record1.getSource(), record2.getSource(), parameter, record1.getTimestamp(), record2.getTimestamp(), survAlgo, leftValue, rightValue, mergedRows[colIdx].getValue(), mergedRows[colIdx].getValues()); break; } if (mergedValue != null) { mergedRows[colIdx].setValue(mergedValue); mergedRows[colIdx].setReferenceValue(mergedCompareValue); } } } } List<DQAttribute<?>> originalRowList = new ArrayList<DQAttribute<?>>(); CollectionUtils.addAll(originalRowList, mergedRows); mergedRecord.setRecordSize(richRecord1.getRecordSize()); mergedRecord.setOriginRow(originalRowList); double minQuality = getGrpqualityMinValue(richRecord1, richRecord2); if (Double.compare(minQuality, 0.0) != 0) { mergedRecord.setGroupQuality(minQuality); } if (richRecord1.getGRP_SIZE() != null && richRecord2.getGRP_SIZE() != null) { int newSize = Integer.valueOf(richRecord1.getGRP_SIZE().getValue()) + Integer.valueOf(richRecord2.getGRP_SIZE().getValue()); if (newSize > 0) { mergedRecord.setGRP_SIZE(newSize); } } return mergedRecord; } DQMFBRecordMerger(String mergedRecordSource, String[] parameters,
            SurvivorShipAlgorithmEnum[] typeMergeTable);  DQMFBRecordMerger(String mergedRecordSource, String[] parameters, SurvivorShipAlgorithmEnum[] typeMergeTable,
            SurvivorShipAlgorithmParams matchMergeParam); SurvivorShipAlgorithmParams getMatchMergeParam(); @Override Record merge(Record record1, Record record2);  }
@Test public void testMerge1() throws ParseException { DQMFBRecordMerger dqMFBRecordMerger = new DQMFBRecordMerger(null, null, new SurvivorShipAlgorithmEnum[] { SurvivorShipAlgorithmEnum.MOST_RECENT }, initSurvivorShipAlgorithmParams()); Map<String, String> patternMap = new HashMap<>(); String datePattern = "dd-MM-yyyy"; patternMap.put("0", datePattern); patternMap.put("1", datePattern); dqMFBRecordMerger.setColumnDatePatternMap(patternMap); SimpleDateFormat simpleDateFormat = new SimpleDateFormat(datePattern); List<Attribute> r1Arributes = new ArrayList<>(); List<Attribute> r2Arributes = new ArrayList<>(); String referenceValue = "03-03-2003"; String inputColValue = "02-02-2000"; String colName = "HIREDATE"; Attribute attribute1 = new Attribute(colName, 0, inputColValue, 1); attribute1.setReferenceValue(referenceValue); r1Arributes.add(attribute1); referenceValue = "05-05-2005"; inputColValue = "06-06-2006"; Attribute attribute2 = new Attribute(colName, 0, inputColValue, 1); attribute2.setReferenceValue(referenceValue); r2Arributes.add(attribute2); RichRecord record1 = new RichRecord(r1Arributes, "record1", simpleDateFormat.parse("02-02-2000").getTime(), "MFB"); RichRecord record2 = new RichRecord(r2Arributes, "record2", simpleDateFormat.parse("03-03-3000").getTime(), "MFB"); record1.setOriginRow(initDQAttribute("02-02-2000", "03-03-2003", "beijing")); record2.setOriginRow(initDQAttribute("06-06-2006", "05-05-2005", "shanghai")); RichRecord createNewRecord = (RichRecord) dqMFBRecordMerger.createNewRecord(record1, record2, 0); Assert.assertEquals("Merge value should be shanghai", "shanghai", createNewRecord.getOriginRow().get(2).getValue()); Assert.assertEquals("Merge reference value should be 05-05-2005", "05-05-2005", createNewRecord.getOriginRow().get(2).getReferenceValue()); Assert.assertEquals("Merge reference column index value should be 1", 1, createNewRecord.getOriginRow().get(2).getReferenceColumnIndex()); }
@Test public void testMerge2() throws ParseException { DQMFBRecordMerger dqMFBRecordMerger = new DQMFBRecordMerger(null, null, new SurvivorShipAlgorithmEnum[] { SurvivorShipAlgorithmEnum.MOST_RECENT }, initSurvivorShipAlgorithmParams()); Map<String, String> patternMap = new HashMap<>(); String datePattern = "dd-MM-yyyy"; patternMap.put("0", datePattern); patternMap.put("1", datePattern); SimpleDateFormat simpleDateFormat = new SimpleDateFormat(datePattern); List<Attribute> r1Arributes = new ArrayList<>(); List<Attribute> r2Arributes = new ArrayList<>(); String referenceValue = "03-03-2003"; String inputColValue = "02-02-2000"; String colName = "HIREDATE"; Attribute attribute1 = new Attribute(colName, 0, inputColValue, 1); attribute1.setReferenceValue(referenceValue); r1Arributes.add(attribute1); referenceValue = "05-05-2005"; inputColValue = "06-06-2006"; Attribute attribute2 = new Attribute(colName, 0, inputColValue, 1); attribute2.setReferenceValue(referenceValue); r2Arributes.add(attribute2); RichRecord record1 = new RichRecord(r1Arributes, "record1", simpleDateFormat.parse("02-02-2000").getTime(), "MFB"); RichRecord record2 = new RichRecord(r2Arributes, "record2", simpleDateFormat.parse("03-03-3000").getTime(), "MFB"); record1.setOriginRow(initDQAttribute("02-02-2000", "03-03-2003", "beijing")); record2.setOriginRow(initDQAttribute("06-06-2006", "05-05-2005", "shanghai")); RichRecord createNewRecord = (RichRecord) dqMFBRecordMerger.createNewRecord(record1, record2, 0); Assert.assertEquals("Merge value should be shanghai", "shanghai", createNewRecord.getOriginRow().get(2).getValue()); Assert.assertEquals("Merge reference value should be 05-05-2005", "05-05-2005", createNewRecord.getOriginRow().get(2).getReferenceValue()); Assert.assertEquals("Merge reference column index value should be 1", 1, createNewRecord.getOriginRow().get(2).getReferenceColumnIndex()); }
@Test public void testMerge3() throws ParseException { DQMFBRecordMerger dqMFBRecordMerger = new DQMFBRecordMerger(null, null, new SurvivorShipAlgorithmEnum[] { SurvivorShipAlgorithmEnum.MOST_RECENT }, initSurvivorShipAlgorithmParams()); Map<String, String> patternMap = new HashMap<>(); String datePattern = "dd-MM-yyyy"; patternMap.put("0", datePattern); dqMFBRecordMerger.setColumnDatePatternMap(patternMap); SimpleDateFormat simpleDateFormat = new SimpleDateFormat(datePattern); List<Attribute> r1Arributes = new ArrayList<>(); List<Attribute> r2Arributes = new ArrayList<>(); String referenceValue = "03-03-2003"; String inputColValue = "02-02-2000"; String colName = "HIREDATE"; Attribute attribute1 = new Attribute(colName, 0, inputColValue, 1); attribute1.setReferenceValue(referenceValue); r1Arributes.add(attribute1); referenceValue = "05-05-2005"; inputColValue = "06-06-2006"; Attribute attribute2 = new Attribute(colName, 0, inputColValue, 1); attribute2.setReferenceValue(referenceValue); r2Arributes.add(attribute2); RichRecord record1 = new RichRecord(r1Arributes, "record1", simpleDateFormat.parse("02-02-2000").getTime(), "MFB"); RichRecord record2 = new RichRecord(r2Arributes, "record2", simpleDateFormat.parse("03-03-3000").getTime(), "MFB"); record1.setOriginRow(initDQAttribute("02-02-2000", "03-03-2003", "beijing")); record2.setOriginRow(initDQAttribute("06-06-2006", "05-05-2005", "shanghai")); RichRecord createNewRecord = (RichRecord) dqMFBRecordMerger.createNewRecord(record1, record2, 0); Assert.assertEquals("Merge value should be shanghai", "shanghai", createNewRecord.getOriginRow().get(2).getValue()); Assert.assertEquals("Merge reference value should be shanghai", "shanghai", createNewRecord.getOriginRow().get(2).getReferenceValue()); Assert.assertEquals("Merge reference column index value should be 1", 1, createNewRecord.getOriginRow().get(2).getReferenceColumnIndex()); }