CSVIngester extends DataIngester { @Override public RowSet getRows(String baseQuery, Map<String, String> preds, int limit, int offset) throws Exception{ filename = conf.getString(MacroBaseConf.CSV_INPUT_FILE); Compression compression = conf.getCsvCompression(); if (compression == Compression.GZIP) { InputStream fileStream = new FileInputStream(filename); InputStream gzipStream = new GZIPInputStream(fileStream); Reader decoder = new InputStreamReader(gzipStream); csvParser = new CSVParser(decoder, CSVFormat.DEFAULT.withHeader()); } else { File csvFile = new File(conf.getString(MacroBaseConf.CSV_INPUT_FILE)); csvParser = CSVParser.parse(csvFile, Charset.defaultCharset(), CSVFormat.DEFAULT.withHeader()); } schema = csvParser.getHeaderMap(); Iterator<CSVRecord> rawIterator = csvParser.iterator(); int rowCount = 0; List<RowSet.Row> rows = Lists.newArrayList(); while (rawIterator.hasNext() && rowCount < limit) { CSVRecord record = rawIterator.next(); List<ColumnValue> columnValues = Lists.newArrayList(); if (includeRow(record, preds)) { for (Map.Entry<String, Integer> se : schema.entrySet()) { columnValues.add(new ColumnValue(se.getKey(),record.get(se.getValue()))); } rows.add(new RowSet.Row(columnValues)); rowCount++; } } return new RowSet(rows); } CSVIngester(MacroBaseConf conf); @Override MBStream<Datum> getStream(); @Override RowSet getRows(String baseQuery,
                          Map<String, String> preds,
                          int limit,
                          int offset);  }
@Test public void testGetRows() throws Exception { MacroBaseConf conf = new MacroBaseConf(); conf.set(MacroBaseConf.CSV_INPUT_FILE, "src/test/resources/data/simple.csv"); conf.set(MacroBaseConf.ATTRIBUTES, Lists.newArrayList("A2","A5")); conf.set(MacroBaseConf.METRICS, Lists.newArrayList("A1","A3","A4")); conf.set(MacroBaseConf.BASE_QUERY, "csv: CSVIngester ingester = new CSVIngester(conf); Map<String, String> preds = new HashMap<>(); preds.put("A2", "1"); RowSet rs = ingester.getRows(conf.getString(MacroBaseConf.BASE_QUERY), preds, 10, 0); assertEquals(1, rs.getRows().size()); System.out.println(rs); }