import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
@Override
public Stream<Row> parse(FileSource source) {
	InputStreamReader csvStreamReader = new InputStreamReader(source.toInputStream(), csvParserConfiguration.getCharset());
	Iterator<String[]> csvRowIterator = new CSVReaderBuilder(csvStreamReader)
		.withCSVParser(
			new CSVParserBuilder()
				.withEscapeChar(csvParserConfiguration.getEscapeChar())
				.withFieldAsNull(csvParserConfiguration.getNullFieldIndicator())
				.withIgnoreLeadingWhiteSpace(csvParserConfiguration.isIgnoreLeadingWhiteSpace())
				.withIgnoreQuotations(csvParserConfiguration.isIgnoreQuotations())
				.withQuoteChar(csvParserConfiguration.getQuoteChar())
				.withSeparator(csvParserConfiguration.getSeparator())
				.withStrictQuotes(csvParserConfiguration.isStrictQuotes())
				.build()
		)
		.withKeepCarriageReturn(csvParserConfiguration.isKeepCr())
		.build()
		.iterator();

	return IteratorStreams
		.stream(new CsvRowIterator(csvRowIterator))
		.onClose(() -> IteratorStreams.close(csvStreamReader));
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
public CSVParserBuilder createCSVParserBuilder() {

    return new CSVParserBuilder()
        .withEscapeChar(this.escapeChar)
        .withIgnoreLeadingWhiteSpace(this.ignoreLeadingWhitespace)
        .withIgnoreQuotations(this.ignoreQuotations)
        .withQuoteChar(this.quoteChar)
        .withSeparator(this.separatorChar)
        .withStrictQuotes(this.strictQuotes)
        .withFieldAsNull(nullFieldIndicator);
  }
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
public void initialize(Map<String, Object> config) {
  if(config.containsKey(COLUMNS_KEY)) {
    columnMap = getColumnMap(config);
  }
  else {
    throw new IllegalStateException("CSVExtractor requires " + COLUMNS_KEY + " configuration");
  }
  char separator = ',';
  if(config.containsKey(SEPARATOR_KEY)) {
    separator = config.get(SEPARATOR_KEY).toString().charAt(0);

  }
  parser = new CSVParserBuilder().withSeparator(separator)
            .build();
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
private ICSVParser createParser() {
    final ICSVParser parser;
    if (_configuration.getEscapeChar() == _configuration.getQuoteChar()) {
        parser = new RFC4180ParserBuilder().withSeparator(_configuration.getSeparatorChar())
                .withQuoteChar(_configuration.getQuoteChar()).build();
    } else {
        parser = new CSVParserBuilder().withSeparator(_configuration.getSeparatorChar())
                .withQuoteChar(_configuration.getQuoteChar()).withEscapeChar(_configuration.getEscapeChar())
                .build();
    }
    return parser;
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
private CSVReader openInputData() throws FileNotFoundException, UnsupportedEncodingException {
    return new CSVReaderBuilder(new InputStreamReader(new FileInputStream(dataInPath + scenarioName + "." + CSV_TYPE), DEFAULT_ENDODING))
            .withCSVParser(new CSVParserBuilder().withSeparator(CSV_CHAR_SEPARATOR).build()).build();
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
private CSVReader openOutputData() throws FileNotFoundException, UnsupportedEncodingException {
    return new CSVReaderBuilder(new InputStreamReader(new FileInputStream(dataOutPath + scenarioName + "." + CSV_TYPE), DEFAULT_ENDODING))
            .withCSVParser(new CSVParserBuilder().withSeparator(CSV_CHAR_SEPARATOR).build()).build();
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
public static Table read(CsvReadOptions options) throws IOException {
    ColumnType[] types = options.columnTypes();
    byte[] bytes = options.reader() != null
        ? CharStreams.toString(options.reader()).getBytes() : null;
    if (types == null) {
      InputStream detectTypesStream = options.reader() != null
          ? new ByteArrayInputStream(bytes)
          : new FileInputStream(options.file());
      types = detectColumnTypes(detectTypesStream, options.header(), options.separator(), options.sample()); 
    }

    // All other read methods end up here, make sure we don't have leading Unicode BOM
    InputStream stream = options.reader() != null
        ? new ByteArrayInputStream(bytes)
        : new FileInputStream(options.file());
    UnicodeBOMInputStream ubis = new UnicodeBOMInputStream(stream);
    ubis.skipBOM();

    Table table;
    CSVParser csvParser = new CSVParserBuilder()
            .withSeparator(options.separator())
            .build();

    try (CSVReader reader = new CSVReaderBuilder(new InputStreamReader(ubis)).withCSVParser(csvParser).build()) {
        String[] nextLine;
        String[] columnNames;
        List<String> headerRow;
        if (options.header()) {
            nextLine = reader.readNext();
            headerRow = Lists.newArrayList(nextLine);
            columnNames = selectColumnNames(headerRow, types);
        } else {
            columnNames = makeColumnNames(types);
            headerRow = Lists.newArrayList(columnNames);
        }

        table = Table.create(options.tableName());
        cleanNames(headerRow);
        for (int x = 0; x < types.length; x++) {
            if (types[x] != SKIP) {
                String columnName = headerRow.get(x);
                if (Strings.isNullOrEmpty(columnName)) {
                    columnName = "Column " + table.columnCount();
                }
                Column newColumn = TypeUtils.newColumn(columnName, types[x]);
                table.addColumn(newColumn);
            }
        }
        int[] columnIndexes = new int[columnNames.length];
        for (int i = 0; i < columnIndexes.length; i++) {
            // get the index in the original table, which includes skipped fields
            columnIndexes[i] = headerRow.indexOf(columnNames[i]);
        }
        // Add the rows
        long rowNumber = options.header() ? 1L : 0L;
        while ((nextLine = reader.readNext()) != null) {
            // for each column that we're including (not skipping)
            int cellIndex = 0;
            for (int columnIndex : columnIndexes) {
                Column column = table.column(cellIndex);
                try {
                    column.appendCell(nextLine[columnIndex]);
                } catch (Exception e) {
                    throw new AddCellToColumnException(e, columnIndex, rowNumber, columnNames, nextLine);
                }
                cellIndex++;
            }
            rowNumber++;
        }
    }
    return table;
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
/**
 * Returns a Table constructed from a CSV File with the given file name
 * <p>
 * The @code{fileName} is used as the initial table name for the new table
 *
 * @param types           An array of the types of columns in the file, in the order they appear
 * @param header          Is the first row in the file a header?
 * @param columnSeparator the delimiter
 * @param file        The fully specified file name. It is used to provide a default name for the table
 * @return A Relation containing the data in the csv file.
 * @throws IOException if file cannot be read
 */
public static Table headerOnly(ColumnType types[], boolean header, char columnSeparator, File file)
        throws IOException {

    FileInputStream fis = new FileInputStream(file);       
    // make sure we don't have leading Unicode BOM
    UnicodeBOMInputStream ubis = new UnicodeBOMInputStream(fis);
    ubis.skipBOM();
    
    Reader reader = new InputStreamReader(ubis);
    BufferedReader streamReader = new BufferedReader(reader);

    Table table;
    CSVParser csvParser = new CSVParserBuilder()
            .withSeparator(columnSeparator)
            .build();
    try (CSVReader csvReader = new CSVReaderBuilder(streamReader).withCSVParser(csvParser).build()) {

        String[] nextLine;
        String[] columnNames;
        List<String> headerRow;
        if (header) {
            nextLine = csvReader.readNext();
            headerRow = Lists.newArrayList(nextLine);
            columnNames = selectColumnNames(headerRow, types);
        } else {
            columnNames = makeColumnNames(types);
            headerRow = Lists.newArrayList(columnNames);
        }

        table = Table.create(file.getName());
        for (int x = 0; x < types.length; x++) {
            if (types[x] != SKIP) {
                Column newColumn = TypeUtils.newColumn(headerRow.get(x).trim(), types[x]);
                table.addColumn(newColumn);
            }
        }
        int[] columnIndexes = new int[columnNames.length];
        for (int i = 0; i < columnIndexes.length; i++) {
            // get the index in the original table, which includes skipped fields
            columnIndexes[i] = headerRow.indexOf(columnNames[i]);
        }
    }
    return table;
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
/**
 * Estimates and returns the type for each column in the delimited text file {@code file}
 * <p>
 * The type is determined by checking a sample of the data in the file. Because only a sample of the data is
 * checked,
 * the types may be incorrect. If that is the case a Parse Exception will be thrown.
 * <p>
 * The method {@code printColumnTypes()} can be used to print a list of the detected columns that can be
 * corrected and
 * used to explicitly specify the correct column types.
 */
protected static ColumnType[] detectColumnTypes(InputStream stream, boolean header, char delimiter, boolean skipSampling)
        throws IOException {

    int linesToSkip = header ? 1 : 0;

    // to hold the results
    List<ColumnType> columnTypes = new ArrayList<>();

    // to hold the data read from the file
    List<List<String>> columnData = new ArrayList<>();

    int rowCount = 0; // make sure we don't go over maxRows

    // make sure we don't have leading Unicode BOM
    UnicodeBOMInputStream ubis = new UnicodeBOMInputStream(stream);
    ubis.skipBOM();

    CSVParser csvParser = new CSVParserBuilder()
            .withSeparator(delimiter)
            .build();
    try (CSVReader reader = new CSVReaderBuilder(new InputStreamReader(ubis))
            .withCSVParser(csvParser)
            .withSkipLines(linesToSkip)
            .build()) {
        String[] nextLine;
        int nextRow = 0;
        while ((nextLine = reader.readNext()) != null) {
            // initialize the arrays to hold the strings. we don't know how many we need until we read the first row
            if (rowCount == 0) {
                for (int j = 0; j < nextLine.length; j++) {
                    columnData.add(new ArrayList<>());
                }
            }
            int columnNumber = 0;
            if (rowCount == nextRow) {
                for (String field : nextLine) {
                    columnData.get(columnNumber).add(field);
                    columnNumber++;
                }
            }
            if (rowCount == nextRow) {
                if (skipSampling) {
                    nextRow = nextRowWithoutSampling(nextRow);
                } else {
                    nextRow = nextRow(nextRow);
                }
            }
            rowCount++;
        }
    }

    // now detect
    for (List<String> valuesList : columnData) {
        ColumnType detectedType = detectType(valuesList);
        columnTypes.add(detectedType);
    }
    return columnTypes.toArray(new ColumnType[columnTypes.size()]);
}
 

import com.opencsv.CSVParserBuilder; //导入依赖的package包/类
private void initialize(Map<String, Object> config) {
    char separator = config.containsKey(DELIMITER_CONF)?config.get(DELIMITER_CONF).toString().charAt(0):',';
    parser = new CSVParserBuilder().withSeparator(separator)
            .build();
    columnMap = (Map<String, Integer>) config.get(COLUMN_MAP_CONF);
}