@Test public void testMultipleRecords() throws IOException, MalformedRecordException { final List<RecordField> fields = getDefaultFields(); fields.replaceAll(f -> f.getFieldName().equals("balance") ? new RecordField("balance", doubleDataType) : f); final RecordSchema schema = new SimpleRecordSchema(fields); try (final InputStream fis = new FileInputStream(new File("src/test/resources/csv/multi-bank-account.csv")); final JacksonCSVRecordReader reader = createReader(fis, schema, format)) { final Object[] firstRecord = reader.nextRecord().getValues(); final Object[] firstExpectedValues = new Object[] {"1", "John Doe", 4750.89D, "123 My Street", "My City", "MS", "11111", "USA"}; Assert.assertArrayEquals(firstExpectedValues, firstRecord); final Object[] secondRecord = reader.nextRecord().getValues(); final Object[] secondExpectedValues = new Object[] {"2", "Jane Doe", 4820.09D, "321 Your Street", "Your City", "NY", "33333", "USA"}; Assert.assertArrayEquals(secondExpectedValues, secondRecord); assertNull(reader.nextRecord()); } }
@Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); if (recordStream.hasNext()) { String[] csvRecord = recordStream.next(); if (rawFieldNames == null) { if (!hasHeader || ignoreHeader) { rawFieldNames = schema.getFieldNames(); } else { rawFieldNames = Arrays.asList(csvRecord); if (recordStream.hasNext()) { csvRecord = recordStream.next(); } else { return null; } } } boolean foundRecord = true; if (csvRecord == null || (csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = false; while (recordStream.hasNext()) { csvRecord = recordStream.next(); if (csvRecord != null && !(csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = true; break; } } } if (!foundRecord) { return null; } final Map<String, Object> values = new HashMap<>(rawFieldNames.size() * 2); final int numFieldNames = rawFieldNames.size(); for (int i = 0; i < csvRecord.length; i++) { final String rawFieldName = numFieldNames <= i ? "unknown_field_index_" + i : rawFieldNames.get(i); String rawValue = (i >= csvRecord.length) ? null : csvRecord[i]; final Optional<DataType> dataTypeOption = schema.getDataType(rawFieldName); if (!dataTypeOption.isPresent() && dropUnknownFields) { continue; } final Object value; if (coerceTypes && dataTypeOption.isPresent()) { value = convert(rawValue, dataTypeOption.get(), rawFieldName); } else if (dataTypeOption.isPresent()) { value = convertSimpleIfPossible(rawValue, dataTypeOption.get(), rawFieldName); } else { value = rawValue; } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; }
JacksonCSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); if (recordStream.hasNext()) { String[] csvRecord = recordStream.next(); if (rawFieldNames == null) { if (!hasHeader || ignoreHeader) { rawFieldNames = schema.getFieldNames(); } else { rawFieldNames = Arrays.asList(csvRecord); if (recordStream.hasNext()) { csvRecord = recordStream.next(); } else { return null; } } } boolean foundRecord = true; if (csvRecord == null || (csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = false; while (recordStream.hasNext()) { csvRecord = recordStream.next(); if (csvRecord != null && !(csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = true; break; } } } if (!foundRecord) { return null; } final Map<String, Object> values = new HashMap<>(rawFieldNames.size() * 2); final int numFieldNames = rawFieldNames.size(); for (int i = 0; i < csvRecord.length; i++) { final String rawFieldName = numFieldNames <= i ? "unknown_field_index_" + i : rawFieldNames.get(i); String rawValue = (i >= csvRecord.length) ? null : csvRecord[i]; final Optional<DataType> dataTypeOption = schema.getDataType(rawFieldName); if (!dataTypeOption.isPresent() && dropUnknownFields) { continue; } final Object value; if (coerceTypes && dataTypeOption.isPresent()) { value = convert(rawValue, dataTypeOption.get(), rawFieldName); } else if (dataTypeOption.isPresent()) { value = convertSimpleIfPossible(rawValue, dataTypeOption.get(), rawFieldName); } else { value = rawValue; } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } }
JacksonCSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); if (recordStream.hasNext()) { String[] csvRecord = recordStream.next(); if (rawFieldNames == null) { if (!hasHeader || ignoreHeader) { rawFieldNames = schema.getFieldNames(); } else { rawFieldNames = Arrays.asList(csvRecord); if (recordStream.hasNext()) { csvRecord = recordStream.next(); } else { return null; } } } boolean foundRecord = true; if (csvRecord == null || (csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = false; while (recordStream.hasNext()) { csvRecord = recordStream.next(); if (csvRecord != null && !(csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = true; break; } } } if (!foundRecord) { return null; } final Map<String, Object> values = new HashMap<>(rawFieldNames.size() * 2); final int numFieldNames = rawFieldNames.size(); for (int i = 0; i < csvRecord.length; i++) { final String rawFieldName = numFieldNames <= i ? "unknown_field_index_" + i : rawFieldNames.get(i); String rawValue = (i >= csvRecord.length) ? null : csvRecord[i]; final Optional<DataType> dataTypeOption = schema.getDataType(rawFieldName); if (!dataTypeOption.isPresent() && dropUnknownFields) { continue; } final Object value; if (coerceTypes && dataTypeOption.isPresent()) { value = convert(rawValue, dataTypeOption.get(), rawFieldName); } else if (dataTypeOption.isPresent()) { value = convertSimpleIfPossible(rawValue, dataTypeOption.get(), rawFieldName); } else { value = rawValue; } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } JacksonCSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat, final boolean hasHeader, final boolean ignoreHeader,
                                  final String dateFormat, final String timeFormat, final String timestampFormat, final String encoding); }
JacksonCSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); if (recordStream.hasNext()) { String[] csvRecord = recordStream.next(); if (rawFieldNames == null) { if (!hasHeader || ignoreHeader) { rawFieldNames = schema.getFieldNames(); } else { rawFieldNames = Arrays.asList(csvRecord); if (recordStream.hasNext()) { csvRecord = recordStream.next(); } else { return null; } } } boolean foundRecord = true; if (csvRecord == null || (csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = false; while (recordStream.hasNext()) { csvRecord = recordStream.next(); if (csvRecord != null && !(csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = true; break; } } } if (!foundRecord) { return null; } final Map<String, Object> values = new HashMap<>(rawFieldNames.size() * 2); final int numFieldNames = rawFieldNames.size(); for (int i = 0; i < csvRecord.length; i++) { final String rawFieldName = numFieldNames <= i ? "unknown_field_index_" + i : rawFieldNames.get(i); String rawValue = (i >= csvRecord.length) ? null : csvRecord[i]; final Optional<DataType> dataTypeOption = schema.getDataType(rawFieldName); if (!dataTypeOption.isPresent() && dropUnknownFields) { continue; } final Object value; if (coerceTypes && dataTypeOption.isPresent()) { value = convert(rawValue, dataTypeOption.get(), rawFieldName); } else if (dataTypeOption.isPresent()) { value = convertSimpleIfPossible(rawValue, dataTypeOption.get(), rawFieldName); } else { value = rawValue; } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } JacksonCSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat, final boolean hasHeader, final boolean ignoreHeader,
                                  final String dateFormat, final String timeFormat, final String timestampFormat, final String encoding); @Override Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields); @Override void close(); }
JacksonCSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); if (recordStream.hasNext()) { String[] csvRecord = recordStream.next(); if (rawFieldNames == null) { if (!hasHeader || ignoreHeader) { rawFieldNames = schema.getFieldNames(); } else { rawFieldNames = Arrays.asList(csvRecord); if (recordStream.hasNext()) { csvRecord = recordStream.next(); } else { return null; } } } boolean foundRecord = true; if (csvRecord == null || (csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = false; while (recordStream.hasNext()) { csvRecord = recordStream.next(); if (csvRecord != null && !(csvRecord.length == 1 && StringUtils.isEmpty(csvRecord[0]))) { foundRecord = true; break; } } } if (!foundRecord) { return null; } final Map<String, Object> values = new HashMap<>(rawFieldNames.size() * 2); final int numFieldNames = rawFieldNames.size(); for (int i = 0; i < csvRecord.length; i++) { final String rawFieldName = numFieldNames <= i ? "unknown_field_index_" + i : rawFieldNames.get(i); String rawValue = (i >= csvRecord.length) ? null : csvRecord[i]; final Optional<DataType> dataTypeOption = schema.getDataType(rawFieldName); if (!dataTypeOption.isPresent() && dropUnknownFields) { continue; } final Object value; if (coerceTypes && dataTypeOption.isPresent()) { value = convert(rawValue, dataTypeOption.get(), rawFieldName); } else if (dataTypeOption.isPresent()) { value = convertSimpleIfPossible(rawValue, dataTypeOption.get(), rawFieldName); } else { value = rawValue; } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } JacksonCSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat, final boolean hasHeader, final boolean ignoreHeader,
                                  final String dateFormat, final String timeFormat, final String timestampFormat, final String encoding); @Override Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields); @Override void close();  }