@Test(dataProvider = "segmentReducerDataProvider") public void segmentReducerTest(String reducerId, SegmentReducerConfig reducerConfig, Set<String> expectedFileNames, List<Object[]> expectedRecords, Comparator comparator) throws Exception { File reducerOutputDir = new File(_baseDir, "reducer_output"); FileUtils.deleteQuietly(reducerOutputDir); assertTrue(reducerOutputDir.mkdirs()); SegmentReducer segmentReducer = new SegmentReducer(reducerId, _partDir, reducerConfig, reducerOutputDir); segmentReducer.reduce(); segmentReducer.cleanup(); File[] avroFileNames = reducerOutputDir.listFiles(); assertEquals(avroFileNames.length, expectedFileNames.size()); GenericRow next = new GenericRow(); int numRecords = 0; List<Object[]> actualRecords = new ArrayList<>(); for (File avroFile : reducerOutputDir.listFiles()) { String fileName = avroFile.getName(); assertTrue(expectedFileNames.contains(fileName)); RecordReader avroRecordReader = new AvroRecordReader(); avroRecordReader.init(avroFile, _pinotSchema.getColumnNames(), null); while (avroRecordReader.hasNext()) { avroRecordReader.next(next); actualRecords.add(new Object[]{next.getValue("campaign"), next.getValue("clicks"), next.getValue("timeValue")}); numRecords++; } } assertEquals(numRecords, expectedRecords.size()); if (comparator != null) { actualRecords.sort(comparator); } for (int i = 0; i < numRecords; i++) { assertEquals(actualRecords.get(i)[0], expectedRecords.get(i)[0]); assertEquals(actualRecords.get(i)[1], expectedRecords.get(i)[1]); assertEquals(actualRecords.get(i)[2], expectedRecords.get(i)[2]); } FileUtils.deleteQuietly(reducerOutputDir); }
public SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig, File reducerOutputDir) { _reducerInputDir = reducerInputDir; _reducerOutputDir = reducerOutputDir; _reducerId = reducerId; _pinotSchema = reducerConfig.getPinotSchema(); _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(_pinotSchema); _collector = CollectorFactory.getCollector(reducerConfig.getCollectorConfig(), _pinotSchema); _numRecordsPerPart = reducerConfig.getNumRecordsPerPart(); LOGGER.info("Initialized reducer with id: {}, input dir: {}, output dir: {}, collector: {}, numRecordsPerPart: {}", _reducerId, _reducerInputDir, _reducerOutputDir, _collector.getClass(), _numRecordsPerPart); }
SegmentReducer { public SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig, File reducerOutputDir) { _reducerInputDir = reducerInputDir; _reducerOutputDir = reducerOutputDir; _reducerId = reducerId; _pinotSchema = reducerConfig.getPinotSchema(); _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(_pinotSchema); _collector = CollectorFactory.getCollector(reducerConfig.getCollectorConfig(), _pinotSchema); _numRecordsPerPart = reducerConfig.getNumRecordsPerPart(); LOGGER.info("Initialized reducer with id: {}, input dir: {}, output dir: {}, collector: {}, numRecordsPerPart: {}", _reducerId, _reducerInputDir, _reducerOutputDir, _collector.getClass(), _numRecordsPerPart); } }
SegmentReducer { public SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig, File reducerOutputDir) { _reducerInputDir = reducerInputDir; _reducerOutputDir = reducerOutputDir; _reducerId = reducerId; _pinotSchema = reducerConfig.getPinotSchema(); _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(_pinotSchema); _collector = CollectorFactory.getCollector(reducerConfig.getCollectorConfig(), _pinotSchema); _numRecordsPerPart = reducerConfig.getNumRecordsPerPart(); LOGGER.info("Initialized reducer with id: {}, input dir: {}, output dir: {}, collector: {}, numRecordsPerPart: {}", _reducerId, _reducerInputDir, _reducerOutputDir, _collector.getClass(), _numRecordsPerPart); } SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig,
      File reducerOutputDir); }
SegmentReducer { public SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig, File reducerOutputDir) { _reducerInputDir = reducerInputDir; _reducerOutputDir = reducerOutputDir; _reducerId = reducerId; _pinotSchema = reducerConfig.getPinotSchema(); _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(_pinotSchema); _collector = CollectorFactory.getCollector(reducerConfig.getCollectorConfig(), _pinotSchema); _numRecordsPerPart = reducerConfig.getNumRecordsPerPart(); LOGGER.info("Initialized reducer with id: {}, input dir: {}, output dir: {}, collector: {}, numRecordsPerPart: {}", _reducerId, _reducerInputDir, _reducerOutputDir, _collector.getClass(), _numRecordsPerPart); } SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig,
      File reducerOutputDir); void reduce(); static String createReducerOutputFileName(String reducerId, int part); void cleanup(); }
SegmentReducer { public SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig, File reducerOutputDir) { _reducerInputDir = reducerInputDir; _reducerOutputDir = reducerOutputDir; _reducerId = reducerId; _pinotSchema = reducerConfig.getPinotSchema(); _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(_pinotSchema); _collector = CollectorFactory.getCollector(reducerConfig.getCollectorConfig(), _pinotSchema); _numRecordsPerPart = reducerConfig.getNumRecordsPerPart(); LOGGER.info("Initialized reducer with id: {}, input dir: {}, output dir: {}, collector: {}, numRecordsPerPart: {}", _reducerId, _reducerInputDir, _reducerOutputDir, _collector.getClass(), _numRecordsPerPart); } SegmentReducer(String reducerId, File reducerInputDir, SegmentReducerConfig reducerConfig,
      File reducerOutputDir); void reduce(); static String createReducerOutputFileName(String reducerId, int part); void cleanup();  }