class CSVIngester extends DataIngester {

    @Override
    public MBStream<Datum> getStream() throws Exception {
        if(!loaded) {
            long st = System.currentTimeMillis();

            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(); //equal to resultSet.getmetadata or smt

            for (Map.Entry<String, Integer> se : schema.entrySet()) {
                conf.getEncoder().recordAttributeName(se.getValue() + 1, se.getKey()); //numbering off each column for encoding
            }

            // Load all records into memory to filter out rows with missing data
            Iterator<CSVRecord> rawIterator = csvParser.iterator();

            int numRows = 0;
            while (rawIterator.hasNext()) {
                try {
                    CSVRecord record = rawIterator.next();
                    Datum curRow = parseRecord(record);
                    dataStream.add(curRow);
                    numRows++;
                } catch (NumberFormatException e) {
                    badRows++;
                }
            }
            log.info("{}/{} rows successfully parsed ({} malformed rows)", numRows, numRows + badRows, badRows);
        }

        return dataStream;
    }

    public  CSVIngester(MacroBaseConf conf);

}