TSFHiveRecordReader implements RecordReader<NullWritable, MapWritable>, IReaderSet { @Override public boolean next(NullWritable key, MapWritable value) throws IOException { while (currentIndex < dataSetList.size()) { if (!dataSetList.get(currentIndex).hasNext()) { currentIndex++; } else { RowRecord rowRecord = dataSetList.get(currentIndex).next(); List<Field> fields = rowRecord.getFields(); long timestamp = rowRecord.getTimestamp(); try { MapWritable res = new MapWritable(); getCurrentValue(deviceIdList, currentIndex, timestamp, isReadTime, isReadDeviceId, fields, measurementIds) .forEach((k, v) -> res.put(new Text(k.toString().toLowerCase()), v)); value.putAll(res); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e.getMessage()); } return true; } } return false; } TSFHiveRecordReader(InputSplit split, JobConf job); @Override boolean next(NullWritable key, MapWritable value); @Override NullWritable createKey(); @Override MapWritable createValue(); @Override long getPos(); @Override float getProgress(); @Override void close(); @Override void setReader(TsFileSequenceReader reader); @Override void setMeasurementIds(List<String> measurementIds); @Override void setReadDeviceId(boolean isReadDeviceId); @Override void setReadTime(boolean isReadTime);  }
@Test public void testNext() { NullWritable key = tsfHiveRecordReader.createKey(); MapWritable value = tsfHiveRecordReader.createValue(); try { assertTrue(tsfHiveRecordReader.next(key, value)); assertEquals(1L, ((LongWritable)value.get(new Text("time_stamp"))).get()); assertEquals("device_1", value.get(new Text("device_id")).toString()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_1"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_2"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_3"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_4"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_5"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_6"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_7"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_8"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_9"))).get()); assertEquals(1000000L, ((LongWritable)value.get(new Text("sensor_10"))).get()); for (int i = 0; i < 100; i++) { assertTrue(tsfHiveRecordReader.next(key, value)); } assertEquals(101L, ((LongWritable)value.get(new Text("time_stamp"))).get()); assertEquals("device_1", value.get(new Text("device_id")).toString()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_1"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_2"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_3"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_4"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_5"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_6"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_7"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_8"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_9"))).get()); assertEquals(1000100L, ((LongWritable)value.get(new Text("sensor_10"))).get()); for (int i = 0; i < 999899; i++) { assertTrue(tsfHiveRecordReader.next(key, value)); } assertEquals(1000000L, ((LongWritable)value.get(new Text("time_stamp"))).get()); assertEquals("device_1", value.get(new Text("device_id")).toString()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_1"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_2"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_3"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_4"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_5"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_6"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_7"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_8"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_9"))).get()); assertEquals(1999999L, ((LongWritable)value.get(new Text("sensor_10"))).get()); assertFalse(tsfHiveRecordReader.next(key, value)); } catch (IOException e) { e.printStackTrace(); fail(); } }