@Test void testGenerateSpecifiedPartition() { SalesforceSource.Histogram histogram = new SalesforceSource.Histogram(); for (String group: HISTOGRAM.split(", ")) { String[] groupInfo = group.split("::"); histogram.add(new SalesforceSource.HistogramGroup(groupInfo[0], Integer.parseInt(groupInfo[1]))); } SalesforceSource source = new SalesforceSource(); long expectedHighWatermark = 20170407152123L; long lowWatermark = 20140213000000L; int maxPartitions = 5; String expectedPartitions = "20140213000000,20170224000000,20170228000000,20170301000000,20170407000000,20170407152123"; String actualPartitions = source.generateSpecifiedPartitions(histogram, 1, maxPartitions, lowWatermark, expectedHighWatermark); Assert.assertEquals(actualPartitions, expectedPartitions); }
String generateSpecifiedPartitions(Histogram histogram, int minTargetPartitionSize, int maxPartitions, long lowWatermark, long expectedHighWatermark) { int interval = computeTargetPartitionSize(histogram, minTargetPartitionSize, maxPartitions); int totalGroups = histogram.getGroups().size(); log.info("Histogram total record count: " + histogram.totalRecordCount); log.info("Histogram total groups: " + totalGroups); log.info("maxPartitions: " + maxPartitions); log.info("interval: " + interval); List<HistogramGroup> groups = histogram.getGroups(); List<String> partitionPoints = new ArrayList<>(); DescriptiveStatistics statistics = new DescriptiveStatistics(); int count = 0; HistogramGroup group; Iterator<HistogramGroup> it = groups.iterator(); while (it.hasNext()) { group = it.next(); if (count == 0) { partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); } if (count != 0 && count + group.count >= 2 * interval) { statistics.addValue(count); partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); count = group.count; } else { count += group.count; } if (count >= interval) { statistics.addValue(count); count = 0; } } if (partitionPoints.isEmpty()) { throw new RuntimeException("Unexpected empty partition list"); } if (count > 0) { statistics.addValue(count); } partitionPoints.add(Long.toString(expectedHighWatermark)); log.info("Dynamic partitioning statistics: "); log.info("data: " + Arrays.toString(statistics.getValues())); log.info(statistics.toString()); String specifiedPartitions = Joiner.on(",").join(partitionPoints); log.info("Calculated specified partitions: " + specifiedPartitions); return specifiedPartitions; }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { String generateSpecifiedPartitions(Histogram histogram, int minTargetPartitionSize, int maxPartitions, long lowWatermark, long expectedHighWatermark) { int interval = computeTargetPartitionSize(histogram, minTargetPartitionSize, maxPartitions); int totalGroups = histogram.getGroups().size(); log.info("Histogram total record count: " + histogram.totalRecordCount); log.info("Histogram total groups: " + totalGroups); log.info("maxPartitions: " + maxPartitions); log.info("interval: " + interval); List<HistogramGroup> groups = histogram.getGroups(); List<String> partitionPoints = new ArrayList<>(); DescriptiveStatistics statistics = new DescriptiveStatistics(); int count = 0; HistogramGroup group; Iterator<HistogramGroup> it = groups.iterator(); while (it.hasNext()) { group = it.next(); if (count == 0) { partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); } if (count != 0 && count + group.count >= 2 * interval) { statistics.addValue(count); partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); count = group.count; } else { count += group.count; } if (count >= interval) { statistics.addValue(count); count = 0; } } if (partitionPoints.isEmpty()) { throw new RuntimeException("Unexpected empty partition list"); } if (count > 0) { statistics.addValue(count); } partitionPoints.add(Long.toString(expectedHighWatermark)); log.info("Dynamic partitioning statistics: "); log.info("data: " + Arrays.toString(statistics.getValues())); log.info(statistics.toString()); String specifiedPartitions = Joiner.on(",").join(partitionPoints); log.info("Calculated specified partitions: " + specifiedPartitions); return specifiedPartitions; } }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { String generateSpecifiedPartitions(Histogram histogram, int minTargetPartitionSize, int maxPartitions, long lowWatermark, long expectedHighWatermark) { int interval = computeTargetPartitionSize(histogram, minTargetPartitionSize, maxPartitions); int totalGroups = histogram.getGroups().size(); log.info("Histogram total record count: " + histogram.totalRecordCount); log.info("Histogram total groups: " + totalGroups); log.info("maxPartitions: " + maxPartitions); log.info("interval: " + interval); List<HistogramGroup> groups = histogram.getGroups(); List<String> partitionPoints = new ArrayList<>(); DescriptiveStatistics statistics = new DescriptiveStatistics(); int count = 0; HistogramGroup group; Iterator<HistogramGroup> it = groups.iterator(); while (it.hasNext()) { group = it.next(); if (count == 0) { partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); } if (count != 0 && count + group.count >= 2 * interval) { statistics.addValue(count); partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); count = group.count; } else { count += group.count; } if (count >= interval) { statistics.addValue(count); count = 0; } } if (partitionPoints.isEmpty()) { throw new RuntimeException("Unexpected empty partition list"); } if (count > 0) { statistics.addValue(count); } partitionPoints.add(Long.toString(expectedHighWatermark)); log.info("Dynamic partitioning statistics: "); log.info("data: " + Arrays.toString(statistics.getValues())); log.info(statistics.toString()); String specifiedPartitions = Joiner.on(",").join(partitionPoints); log.info("Calculated specified partitions: " + specifiedPartitions); return specifiedPartitions; } SalesforceSource(); @VisibleForTesting  SalesforceSource(LineageInfo lineageInfo); }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { String generateSpecifiedPartitions(Histogram histogram, int minTargetPartitionSize, int maxPartitions, long lowWatermark, long expectedHighWatermark) { int interval = computeTargetPartitionSize(histogram, minTargetPartitionSize, maxPartitions); int totalGroups = histogram.getGroups().size(); log.info("Histogram total record count: " + histogram.totalRecordCount); log.info("Histogram total groups: " + totalGroups); log.info("maxPartitions: " + maxPartitions); log.info("interval: " + interval); List<HistogramGroup> groups = histogram.getGroups(); List<String> partitionPoints = new ArrayList<>(); DescriptiveStatistics statistics = new DescriptiveStatistics(); int count = 0; HistogramGroup group; Iterator<HistogramGroup> it = groups.iterator(); while (it.hasNext()) { group = it.next(); if (count == 0) { partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); } if (count != 0 && count + group.count >= 2 * interval) { statistics.addValue(count); partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); count = group.count; } else { count += group.count; } if (count >= interval) { statistics.addValue(count); count = 0; } } if (partitionPoints.isEmpty()) { throw new RuntimeException("Unexpected empty partition list"); } if (count > 0) { statistics.addValue(count); } partitionPoints.add(Long.toString(expectedHighWatermark)); log.info("Dynamic partitioning statistics: "); log.info("data: " + Arrays.toString(statistics.getValues())); log.info(statistics.toString()); String specifiedPartitions = Joiner.on(",").join(partitionPoints); log.info("Calculated specified partitions: " + specifiedPartitions); return specifiedPartitions; } SalesforceSource(); @VisibleForTesting  SalesforceSource(LineageInfo lineageInfo); @Override Extractor<JsonArray, JsonElement> getExtractor(WorkUnitState state); @Override boolean isEarlyStopped(); }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { String generateSpecifiedPartitions(Histogram histogram, int minTargetPartitionSize, int maxPartitions, long lowWatermark, long expectedHighWatermark) { int interval = computeTargetPartitionSize(histogram, minTargetPartitionSize, maxPartitions); int totalGroups = histogram.getGroups().size(); log.info("Histogram total record count: " + histogram.totalRecordCount); log.info("Histogram total groups: " + totalGroups); log.info("maxPartitions: " + maxPartitions); log.info("interval: " + interval); List<HistogramGroup> groups = histogram.getGroups(); List<String> partitionPoints = new ArrayList<>(); DescriptiveStatistics statistics = new DescriptiveStatistics(); int count = 0; HistogramGroup group; Iterator<HistogramGroup> it = groups.iterator(); while (it.hasNext()) { group = it.next(); if (count == 0) { partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); } if (count != 0 && count + group.count >= 2 * interval) { statistics.addValue(count); partitionPoints.add(Utils.toDateTimeFormat(group.getKey(), SECONDS_FORMAT, Partitioner.WATERMARKTIMEFORMAT)); count = group.count; } else { count += group.count; } if (count >= interval) { statistics.addValue(count); count = 0; } } if (partitionPoints.isEmpty()) { throw new RuntimeException("Unexpected empty partition list"); } if (count > 0) { statistics.addValue(count); } partitionPoints.add(Long.toString(expectedHighWatermark)); log.info("Dynamic partitioning statistics: "); log.info("data: " + Arrays.toString(statistics.getValues())); log.info(statistics.toString()); String specifiedPartitions = Joiner.on(",").join(partitionPoints); log.info("Calculated specified partitions: " + specifiedPartitions); return specifiedPartitions; } SalesforceSource(); @VisibleForTesting  SalesforceSource(LineageInfo lineageInfo); @Override Extractor<JsonArray, JsonElement> getExtractor(WorkUnitState state); @Override boolean isEarlyStopped(); static final String USE_ALL_OBJECTS; static final boolean DEFAULT_USE_ALL_OBJECTS; }