@Test(dataProvider = "segmentMapperConfigProvider") public void segmentMapperTest(String mapperId, SegmentMapperConfig segmentMapperConfig, Map<String, List<Object[]>> partitionToRecords) throws Exception { File mapperOutputDir = new File(_baseDir, "mapper_output"); FileUtils.deleteQuietly(mapperOutputDir); assertTrue(mapperOutputDir.mkdirs()); SegmentMapper segmentMapper = new SegmentMapper(mapperId, _inputSegment, segmentMapperConfig, mapperOutputDir); segmentMapper.map(); segmentMapper.cleanup(); File[] partitionDirs = mapperOutputDir.listFiles(); assertEquals(partitionDirs.length, partitionToRecords.size()); for (File partitionDir : partitionDirs) { String partition = partitionDir.getName(); assertTrue(partitionToRecords.containsKey(partition)); File[] avroFiles = partitionDir.listFiles(); assertEquals(avroFiles.length, 1); assertEquals(avroFiles[0].getName(), SegmentMapper.createMapperOutputFileName(mapperId)); RecordReader avroRecordReader = new AvroRecordReader(); avroRecordReader.init(avroFiles[0], _pinotSchema.getColumnNames(), null); int numRecords = 0; List<Object[]> expectedRecords = partitionToRecords.get(partition); GenericRow next = new GenericRow(); while (avroRecordReader.hasNext()) { avroRecordReader.next(next); assertEquals(next.getValue("campaign"), expectedRecords.get(numRecords)[0]); assertEquals(next.getValue("clicks"), expectedRecords.get(numRecords)[1]); assertEquals(next.getValue("timeValue"), expectedRecords.get(numRecords)[2]); numRecords++; } assertEquals(numRecords, expectedRecords.size()); } FileUtils.deleteQuietly(mapperOutputDir); }
public SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir) { _inputSegment = inputSegment; _mapperOutputDir = mapperOutputDir; _mapperId = mapperId; _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(mapperConfig.getPinotSchema()); _recordTransformer = RecordTransformerFactory.getRecordTransformer(mapperConfig.getRecordTransformerConfig()); _recordFilter = RecordFilterFactory.getRecordFilter(mapperConfig.getRecordFilterConfig()); for (PartitionerConfig partitionerConfig : mapperConfig.getPartitionerConfigs()) { _partitioners.add(PartitionerFactory.getPartitioner(partitionerConfig)); } _numPartitioners = _partitioners.size(); LOGGER.info( "Initialized mapper with id: {}, input segment: {}, output dir: {}, recordTransformer: {}, recordFilter: {}, partitioners: {}", _mapperId, _inputSegment, _mapperOutputDir, _recordTransformer.getClass(), _recordFilter.getClass(), _partitioners.stream().map(p -> p.getClass().toString()).collect(Collectors.joining(","))); }
SegmentMapper { public SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir) { _inputSegment = inputSegment; _mapperOutputDir = mapperOutputDir; _mapperId = mapperId; _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(mapperConfig.getPinotSchema()); _recordTransformer = RecordTransformerFactory.getRecordTransformer(mapperConfig.getRecordTransformerConfig()); _recordFilter = RecordFilterFactory.getRecordFilter(mapperConfig.getRecordFilterConfig()); for (PartitionerConfig partitionerConfig : mapperConfig.getPartitionerConfigs()) { _partitioners.add(PartitionerFactory.getPartitioner(partitionerConfig)); } _numPartitioners = _partitioners.size(); LOGGER.info( "Initialized mapper with id: {}, input segment: {}, output dir: {}, recordTransformer: {}, recordFilter: {}, partitioners: {}", _mapperId, _inputSegment, _mapperOutputDir, _recordTransformer.getClass(), _recordFilter.getClass(), _partitioners.stream().map(p -> p.getClass().toString()).collect(Collectors.joining(","))); } }
SegmentMapper { public SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir) { _inputSegment = inputSegment; _mapperOutputDir = mapperOutputDir; _mapperId = mapperId; _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(mapperConfig.getPinotSchema()); _recordTransformer = RecordTransformerFactory.getRecordTransformer(mapperConfig.getRecordTransformerConfig()); _recordFilter = RecordFilterFactory.getRecordFilter(mapperConfig.getRecordFilterConfig()); for (PartitionerConfig partitionerConfig : mapperConfig.getPartitionerConfigs()) { _partitioners.add(PartitionerFactory.getPartitioner(partitionerConfig)); } _numPartitioners = _partitioners.size(); LOGGER.info( "Initialized mapper with id: {}, input segment: {}, output dir: {}, recordTransformer: {}, recordFilter: {}, partitioners: {}", _mapperId, _inputSegment, _mapperOutputDir, _recordTransformer.getClass(), _recordFilter.getClass(), _partitioners.stream().map(p -> p.getClass().toString()).collect(Collectors.joining(","))); } SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir); }
SegmentMapper { public SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir) { _inputSegment = inputSegment; _mapperOutputDir = mapperOutputDir; _mapperId = mapperId; _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(mapperConfig.getPinotSchema()); _recordTransformer = RecordTransformerFactory.getRecordTransformer(mapperConfig.getRecordTransformerConfig()); _recordFilter = RecordFilterFactory.getRecordFilter(mapperConfig.getRecordFilterConfig()); for (PartitionerConfig partitionerConfig : mapperConfig.getPartitionerConfigs()) { _partitioners.add(PartitionerFactory.getPartitioner(partitionerConfig)); } _numPartitioners = _partitioners.size(); LOGGER.info( "Initialized mapper with id: {}, input segment: {}, output dir: {}, recordTransformer: {}, recordFilter: {}, partitioners: {}", _mapperId, _inputSegment, _mapperOutputDir, _recordTransformer.getClass(), _recordFilter.getClass(), _partitioners.stream().map(p -> p.getClass().toString()).collect(Collectors.joining(","))); } SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir); void map(); void cleanup(); static String createMapperOutputFileName(String mapperId); }
SegmentMapper { public SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir) { _inputSegment = inputSegment; _mapperOutputDir = mapperOutputDir; _mapperId = mapperId; _avroSchema = SegmentProcessorUtils.convertPinotSchemaToAvroSchema(mapperConfig.getPinotSchema()); _recordTransformer = RecordTransformerFactory.getRecordTransformer(mapperConfig.getRecordTransformerConfig()); _recordFilter = RecordFilterFactory.getRecordFilter(mapperConfig.getRecordFilterConfig()); for (PartitionerConfig partitionerConfig : mapperConfig.getPartitionerConfigs()) { _partitioners.add(PartitionerFactory.getPartitioner(partitionerConfig)); } _numPartitioners = _partitioners.size(); LOGGER.info( "Initialized mapper with id: {}, input segment: {}, output dir: {}, recordTransformer: {}, recordFilter: {}, partitioners: {}", _mapperId, _inputSegment, _mapperOutputDir, _recordTransformer.getClass(), _recordFilter.getClass(), _partitioners.stream().map(p -> p.getClass().toString()).collect(Collectors.joining(","))); } SegmentMapper(String mapperId, File inputSegment, SegmentMapperConfig mapperConfig, File mapperOutputDir); void map(); void cleanup(); static String createMapperOutputFileName(String mapperId);  }