@Test public void testHourlyPartitioner() { Map<String, Object> config = new HashMap<>(); config.put(StorageCommonConfig.DIRECTORY_DELIM_CONFIG, StorageCommonConfig.DIRECTORY_DELIM_DEFAULT); config.put(PartitionerConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, "Record"); config.put(PartitionerConfig.LOCALE_CONFIG, Locale.US.toString()); config.put(PartitionerConfig.TIMEZONE_CONFIG, DATE_TIME_ZONE.toString()); HourlyPartitioner<String> partitioner = new HourlyPartitioner<>(); partitioner.configure(config); String timeZoneString = (String) config.get(PartitionerConfig.TIMEZONE_CONFIG); int year = 2014; int month = 2; int day = 1; int hour = 3; long timestamp = new DateTime(year, month, day, hour, 0, 0, 0, DateTimeZone.forID(timeZoneString)).getMillis(); SinkRecord sinkRecord = createSinkRecord(timestamp); String encodedPartition = partitioner.encodePartition(sinkRecord); Map<String, Object> m = new LinkedHashMap<>(); m.put("year", year); m.put("month", String.format("%02d", month)); m.put("day", String.format("%02d", day)); m.put("hour", String.format("%02d", hour)); assertThat(encodedPartition, is(generateEncodedPartitionFromMap(m))); }
@Override public void configure(Map<String, Object> config) { long partitionDurationMs = TimeUnit.HOURS.toMillis(1); String delim = (String) config.get(StorageCommonConfig.DIRECTORY_DELIM_CONFIG); String pathFormat = "'year'=YYYY" + delim + "'month'=MM" + delim + "'day'=dd" + delim + "'hour'=HH"; config.put(PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMs); config.put(PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat); super.configure(config); }
HourlyPartitioner extends TimeBasedPartitioner<T> { @Override public void configure(Map<String, Object> config) { long partitionDurationMs = TimeUnit.HOURS.toMillis(1); String delim = (String) config.get(StorageCommonConfig.DIRECTORY_DELIM_CONFIG); String pathFormat = "'year'=YYYY" + delim + "'month'=MM" + delim + "'day'=dd" + delim + "'hour'=HH"; config.put(PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMs); config.put(PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat); super.configure(config); } }
HourlyPartitioner extends TimeBasedPartitioner<T> { @Override public void configure(Map<String, Object> config) { long partitionDurationMs = TimeUnit.HOURS.toMillis(1); String delim = (String) config.get(StorageCommonConfig.DIRECTORY_DELIM_CONFIG); String pathFormat = "'year'=YYYY" + delim + "'month'=MM" + delim + "'day'=dd" + delim + "'hour'=HH"; config.put(PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMs); config.put(PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat); super.configure(config); }  }
HourlyPartitioner extends TimeBasedPartitioner<T> { @Override public void configure(Map<String, Object> config) { long partitionDurationMs = TimeUnit.HOURS.toMillis(1); String delim = (String) config.get(StorageCommonConfig.DIRECTORY_DELIM_CONFIG); String pathFormat = "'year'=YYYY" + delim + "'month'=MM" + delim + "'day'=dd" + delim + "'hour'=HH"; config.put(PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMs); config.put(PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat); super.configure(config); }  @Override void configure(Map<String, Object> config); }
HourlyPartitioner extends TimeBasedPartitioner<T> { @Override public void configure(Map<String, Object> config) { long partitionDurationMs = TimeUnit.HOURS.toMillis(1); String delim = (String) config.get(StorageCommonConfig.DIRECTORY_DELIM_CONFIG); String pathFormat = "'year'=YYYY" + delim + "'month'=MM" + delim + "'day'=dd" + delim + "'hour'=HH"; config.put(PartitionerConfig.PARTITION_DURATION_MS_CONFIG, partitionDurationMs); config.put(PartitionerConfig.PATH_FORMAT_CONFIG, pathFormat); super.configure(config); }  @Override void configure(Map<String, Object> config);  }