test_class.declarator: log = LoggerFactory.getLogger(CSVIngesterTest.class)
test_class.modifier: private static final
test_class.original_string: private static final Logger log = LoggerFactory.getLogger(CSVIngesterTest.class);
test_class.type: Logger
test_class.var_name: log
test_class.file: legacy/src/test/java/macrobase/ingest/CSVIngesterTest.java
test_class.identifier: CSVIngesterTest
test_case.body: @Test
      /**
       * Test that when we encounter missing metrics we skip the entire row and read
       * all of the valid rows.
       */
      public void testMissingData() throws Exception {
          MacroBaseConf conf = new MacroBaseConf();
          conf.set(MacroBaseConf.CSV_INPUT_FILE, "src/test/resources/data/missingdata.csv");
          conf.set(MacroBaseConf.ATTRIBUTES, Lists.newArrayList("a1"));
          conf.set(MacroBaseConf.METRICS, Lists.newArrayList("m1","m2"));
  
          CSVIngester ingester = new CSVIngester(conf);
          List<Datum> data = ingester.getStream().drain();
  
          int count = 0;
          for(Datum d : data) {
              if (count == 3) {
                  assertEquals(5, d.metrics().getEntry(0), 0.0);
              }
              count++;
          }
  
          assertEquals(4, count);
      }
test_case.class_method_signature: CSVIngesterTest.testMissingData()
test_case.full_signature: @Test /** * Test that when we encounter missing metrics we skip the entire row and read * all of the valid rows. */ public void testMissingData()
test_case.identifier: testMissingData
test_case.invocations: set set newArrayList set newArrayList drain getStream assertEquals getEntry metrics assertEquals
test_case.modifiers: @Test /** * Test that when we encounter missing metrics we skip the entire row and read * all of the valid rows. */ public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testMissingData()
focal_class.declarator: log = LoggerFactory.getLogger(CSVIngester.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(CSVIngester.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: csvParser
focal_class.modifier: private
focal_class.original_string: private CSVParser csvParser;
focal_class.type: CSVParser
focal_class.var_name: csvParser
focal_class.declarator: schema
focal_class.modifier: private
focal_class.original_string: private Map<String, Integer> schema;
focal_class.type: Map<String, Integer>
focal_class.var_name: schema
focal_class.declarator: filename
focal_class.modifier: private
focal_class.original_string: private String filename;
focal_class.type: String
focal_class.var_name: filename
focal_class.declarator: dataStream = new MBStream<>()
focal_class.modifier: private
focal_class.original_string: private MBStream<Datum> dataStream = new MBStream<>();
focal_class.type: MBStream<Datum>
focal_class.var_name: dataStream
focal_class.declarator: loaded = false
focal_class.modifier: private
focal_class.original_string: private boolean loaded = false;
focal_class.type: boolean
focal_class.var_name: loaded
focal_class.declarator: badRows = 0
focal_class.modifier: private
focal_class.original_string: private int badRows = 0;
focal_class.type: int
focal_class.var_name: badRows
focal_class.file: legacy/src/main/java/macrobase/ingest/CSVIngester.java
focal_class.identifier: CSVIngester
focal_class.class_method_signature: CSVIngester.CSVIngester(MacroBaseConf conf)
focal_class.full_signature: public  CSVIngester(MacroBaseConf conf)
focal_class.identifier: CSVIngester
focal_class.modifiers: public
focal_class.parameters: (MacroBaseConf conf)
focal_class.signature:  CSVIngester(MacroBaseConf conf)
focal_class.class_method_signature: CSVIngester.parseRecord(CSVRecord record)
focal_class.full_signature: private Datum parseRecord(CSVRecord record)
focal_class.identifier: parseRecord
focal_class.modifiers: private
focal_class.parameters: (CSVRecord record)
focal_class.return: Datum
focal_class.signature: Datum parseRecord(CSVRecord record)
focal_class.class_method_signature: CSVIngester.getStream()
focal_class.full_signature: @Override public MBStream<Datum> getStream()
focal_class.identifier: getStream
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: MBStream<Datum>
focal_class.signature: MBStream<Datum> getStream()
focal_class.class_method_signature: CSVIngester.getRows(String baseQuery,
                            Map<String, String> preds,
                            int limit,
                            int offset)
focal_class.full_signature: @Override public RowSet getRows(String baseQuery,
                            Map<String, String> preds,
                            int limit,
                            int offset)
focal_class.identifier: getRows
focal_class.modifiers: @Override public
focal_class.parameters: (String baseQuery,
                            Map<String, String> preds,
                            int limit,
                            int offset)
focal_class.return: RowSet
focal_class.signature: RowSet getRows(String baseQuery,
                            Map<String, String> preds,
                            int limit,
                            int offset)
focal_class.class_method_signature: CSVIngester.includeRow(CSVRecord record, Map<String, String> preds)
focal_class.full_signature: private boolean includeRow(CSVRecord record, Map<String, String> preds)
focal_class.identifier: includeRow
focal_class.modifiers: private
focal_class.parameters: (CSVRecord record, Map<String, String> preds)
focal_class.return: boolean
focal_class.signature: boolean includeRow(CSVRecord record, Map<String, String> preds)
focal_class.superclass: extends DataIngester
focal_method.body: @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;
      }
focal_method.class_method_signature: CSVIngester.getStream()
focal_method.full_signature: @Override public MBStream<Datum> getStream()
focal_method.identifier: getStream
focal_method.invocations: currentTimeMillis getString getCsvCompression withHeader getString parse defaultCharset withHeader getHeaderMap entrySet recordAttributeName getEncoder getValue getKey iterator hasNext next parseRecord add info
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: MBStream<Datum>
focal_method.signature: MBStream<Datum> getStream()