@Test public void testGetIntervalsOnMaxInterval() { HourWatermark hourwm = new HourWatermark("Hourwatermark", "test"); long lwm = 20150201011111l; long hwm = 20150202011111l; int partition = 2; int maxInterval = 2; Map<Long, Long> results = hourwm.getIntervals(lwm, hwm, partition, maxInterval); Map<Long, Long> expected = Maps.newHashMap(); expected.put(20150201010000l, 20150201130000l); expected.put(20150201130000l, 20150202010000l); Assert.assertEquals(results, expected); }
@Override synchronized public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionIntervalInHours, int maxIntervals) { Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); Preconditions .checkArgument(partitionIntervalInHours > 0, "Invalid value for partitionInterval, should be at least 1."); HashMap<Long, Long> intervalMap = Maps.newHashMap(); if (lowWatermarkValue > highWatermarkValue) { LOG.warn("The low water mark is greater than the high water mark, empty intervals are returned"); return intervalMap; } final Calendar calendar = Calendar.getInstance(); Date nextTime; Date lowWatermarkDate = extractFromTimestamp(Long.toString(lowWatermarkValue)); Date highWatermarkDate = extractFromTimestamp(Long.toString(highWatermarkValue)); final long lowWatermark = lowWatermarkDate.getTime(); final long highWatermark = highWatermarkDate.getTime(); int interval = getInterval(highWatermark - lowWatermark, partitionIntervalInHours, maxIntervals); LOG.info("Recalculated partition interval:" + interval + " hours"); Date startTime = new Date(lowWatermark); Date endTime = new Date(highWatermark); LOG.debug("Start time:" + startTime + "; End time:" + endTime); long lwm; long hwm; if (startTime.getTime() == endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); hwm = lwm; intervalMap.put(lwm, hwm); return intervalMap; } while (startTime.getTime() < endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); calendar.setTime(startTime); calendar.add(Calendar.HOUR, interval); nextTime = calendar.getTime(); hwm = Long.parseLong(inputFormatParser.format(nextTime.getTime() <= endTime.getTime() ? nextTime : endTime)); intervalMap.put(lwm, hwm); LOG.debug("Partition - low:" + lwm + "; high:" + hwm); startTime = nextTime; } return intervalMap; }
HourWatermark implements Watermark { @Override synchronized public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionIntervalInHours, int maxIntervals) { Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); Preconditions .checkArgument(partitionIntervalInHours > 0, "Invalid value for partitionInterval, should be at least 1."); HashMap<Long, Long> intervalMap = Maps.newHashMap(); if (lowWatermarkValue > highWatermarkValue) { LOG.warn("The low water mark is greater than the high water mark, empty intervals are returned"); return intervalMap; } final Calendar calendar = Calendar.getInstance(); Date nextTime; Date lowWatermarkDate = extractFromTimestamp(Long.toString(lowWatermarkValue)); Date highWatermarkDate = extractFromTimestamp(Long.toString(highWatermarkValue)); final long lowWatermark = lowWatermarkDate.getTime(); final long highWatermark = highWatermarkDate.getTime(); int interval = getInterval(highWatermark - lowWatermark, partitionIntervalInHours, maxIntervals); LOG.info("Recalculated partition interval:" + interval + " hours"); Date startTime = new Date(lowWatermark); Date endTime = new Date(highWatermark); LOG.debug("Start time:" + startTime + "; End time:" + endTime); long lwm; long hwm; if (startTime.getTime() == endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); hwm = lwm; intervalMap.put(lwm, hwm); return intervalMap; } while (startTime.getTime() < endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); calendar.setTime(startTime); calendar.add(Calendar.HOUR, interval); nextTime = calendar.getTime(); hwm = Long.parseLong(inputFormatParser.format(nextTime.getTime() <= endTime.getTime() ? nextTime : endTime)); intervalMap.put(lwm, hwm); LOG.debug("Partition - low:" + lwm + "; high:" + hwm); startTime = nextTime; } return intervalMap; } }
HourWatermark implements Watermark { @Override synchronized public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionIntervalInHours, int maxIntervals) { Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); Preconditions .checkArgument(partitionIntervalInHours > 0, "Invalid value for partitionInterval, should be at least 1."); HashMap<Long, Long> intervalMap = Maps.newHashMap(); if (lowWatermarkValue > highWatermarkValue) { LOG.warn("The low water mark is greater than the high water mark, empty intervals are returned"); return intervalMap; } final Calendar calendar = Calendar.getInstance(); Date nextTime; Date lowWatermarkDate = extractFromTimestamp(Long.toString(lowWatermarkValue)); Date highWatermarkDate = extractFromTimestamp(Long.toString(highWatermarkValue)); final long lowWatermark = lowWatermarkDate.getTime(); final long highWatermark = highWatermarkDate.getTime(); int interval = getInterval(highWatermark - lowWatermark, partitionIntervalInHours, maxIntervals); LOG.info("Recalculated partition interval:" + interval + " hours"); Date startTime = new Date(lowWatermark); Date endTime = new Date(highWatermark); LOG.debug("Start time:" + startTime + "; End time:" + endTime); long lwm; long hwm; if (startTime.getTime() == endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); hwm = lwm; intervalMap.put(lwm, hwm); return intervalMap; } while (startTime.getTime() < endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); calendar.setTime(startTime); calendar.add(Calendar.HOUR, interval); nextTime = calendar.getTime(); hwm = Long.parseLong(inputFormatParser.format(nextTime.getTime() <= endTime.getTime() ? nextTime : endTime)); intervalMap.put(lwm, hwm); LOG.debug("Partition - low:" + lwm + "; high:" + hwm); startTime = nextTime; } return intervalMap; } HourWatermark(String watermarkColumn, String watermarkFormat); }
HourWatermark implements Watermark { @Override synchronized public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionIntervalInHours, int maxIntervals) { Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); Preconditions .checkArgument(partitionIntervalInHours > 0, "Invalid value for partitionInterval, should be at least 1."); HashMap<Long, Long> intervalMap = Maps.newHashMap(); if (lowWatermarkValue > highWatermarkValue) { LOG.warn("The low water mark is greater than the high water mark, empty intervals are returned"); return intervalMap; } final Calendar calendar = Calendar.getInstance(); Date nextTime; Date lowWatermarkDate = extractFromTimestamp(Long.toString(lowWatermarkValue)); Date highWatermarkDate = extractFromTimestamp(Long.toString(highWatermarkValue)); final long lowWatermark = lowWatermarkDate.getTime(); final long highWatermark = highWatermarkDate.getTime(); int interval = getInterval(highWatermark - lowWatermark, partitionIntervalInHours, maxIntervals); LOG.info("Recalculated partition interval:" + interval + " hours"); Date startTime = new Date(lowWatermark); Date endTime = new Date(highWatermark); LOG.debug("Start time:" + startTime + "; End time:" + endTime); long lwm; long hwm; if (startTime.getTime() == endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); hwm = lwm; intervalMap.put(lwm, hwm); return intervalMap; } while (startTime.getTime() < endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); calendar.setTime(startTime); calendar.add(Calendar.HOUR, interval); nextTime = calendar.getTime(); hwm = Long.parseLong(inputFormatParser.format(nextTime.getTime() <= endTime.getTime() ? nextTime : endTime)); intervalMap.put(lwm, hwm); LOG.debug("Partition - low:" + lwm + "; high:" + hwm); startTime = nextTime; } return intervalMap; } HourWatermark(String watermarkColumn, String watermarkFormat); @Override String getWatermarkCondition(QueryBasedExtractor<?, ?> extractor, long watermarkValue, String operator); @Override int getDeltaNumForNextWatermark(); @Override synchronized HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue,
      long partitionIntervalInHours, int maxIntervals); static long adjustWatermark(String baseWatermark, int diff); }
HourWatermark implements Watermark { @Override synchronized public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionIntervalInHours, int maxIntervals) { Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); Preconditions .checkArgument(partitionIntervalInHours > 0, "Invalid value for partitionInterval, should be at least 1."); HashMap<Long, Long> intervalMap = Maps.newHashMap(); if (lowWatermarkValue > highWatermarkValue) { LOG.warn("The low water mark is greater than the high water mark, empty intervals are returned"); return intervalMap; } final Calendar calendar = Calendar.getInstance(); Date nextTime; Date lowWatermarkDate = extractFromTimestamp(Long.toString(lowWatermarkValue)); Date highWatermarkDate = extractFromTimestamp(Long.toString(highWatermarkValue)); final long lowWatermark = lowWatermarkDate.getTime(); final long highWatermark = highWatermarkDate.getTime(); int interval = getInterval(highWatermark - lowWatermark, partitionIntervalInHours, maxIntervals); LOG.info("Recalculated partition interval:" + interval + " hours"); Date startTime = new Date(lowWatermark); Date endTime = new Date(highWatermark); LOG.debug("Start time:" + startTime + "; End time:" + endTime); long lwm; long hwm; if (startTime.getTime() == endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); hwm = lwm; intervalMap.put(lwm, hwm); return intervalMap; } while (startTime.getTime() < endTime.getTime()) { lwm = Long.parseLong(inputFormatParser.format(startTime)); calendar.setTime(startTime); calendar.add(Calendar.HOUR, interval); nextTime = calendar.getTime(); hwm = Long.parseLong(inputFormatParser.format(nextTime.getTime() <= endTime.getTime() ? nextTime : endTime)); intervalMap.put(lwm, hwm); LOG.debug("Partition - low:" + lwm + "; high:" + hwm); startTime = nextTime; } return intervalMap; } HourWatermark(String watermarkColumn, String watermarkFormat); @Override String getWatermarkCondition(QueryBasedExtractor<?, ?> extractor, long watermarkValue, String operator); @Override int getDeltaNumForNextWatermark(); @Override synchronized HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue,
      long partitionIntervalInHours, int maxIntervals); static long adjustWatermark(String baseWatermark, int diff);  }