@Test public void testHeaders() throws Exception { String[] directives = new String[] { "parse-as-csv body , true" }; List<Row> rows = Arrays.asList( new Row("body", "first name, last \t name"), new Row("body", "alice,zed") ); rows = TestingRig.execute(directives, rows); Assert.assertEquals(1, rows.size()); Assert.assertEquals("alice", rows.get(0).getValue("first_name")); Assert.assertEquals("zed", rows.get(0).getValue("last_name")); }
@Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; } }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }  }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); static final String NAME; }