flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhueske <...@git.apache.org>
Subject [GitHub] flink pull request: [FLINK-2692] Untangle CsvInputFormat
Date Tue, 10 Nov 2015 17:10:02 GMT
Github user fhueske commented on a diff in the pull request:

    https://github.com/apache/flink/pull/1266#discussion_r44434150
  
    --- Diff: flink-java/src/main/java/org/apache/flink/api/java/io/CsvInputFormat.java ---
    @@ -18,32 +18,97 @@
     
     package org.apache.flink.api.java.io;
     
    +import com.google.common.base.Preconditions;
    +import com.google.common.primitives.Ints;
    +import org.apache.flink.api.common.io.GenericCsvInputFormat;
    +import org.apache.flink.core.fs.FileInputSplit;
    +import org.apache.flink.types.parser.FieldParser;
     
    -import org.apache.flink.api.common.typeutils.CompositeType;
    -import org.apache.flink.api.java.tuple.Tuple;
    +import java.io.IOException;
     import org.apache.flink.core.fs.Path;
     import org.apache.flink.util.StringUtils;
     
    -public class CsvInputFormat<OUT> extends CommonCsvInputFormat<OUT> {
    +public abstract class CsvInputFormat<OUT> extends GenericCsvInputFormat<OUT>
{
     
     	private static final long serialVersionUID = 1L;
    +
    +	public static final String DEFAULT_LINE_DELIMITER = "\n";
    +
    +	public static final String DEFAULT_FIELD_DELIMITER = ",";
    +
    +	protected transient Object[] parsedValues;
     	
    -	public CsvInputFormat(Path filePath, CompositeType<OUT> typeInformation) {
    -		super(filePath, typeInformation);
    +	protected CsvInputFormat(Path filePath) {
    +		super(filePath);
     	}
    -	
    -	public CsvInputFormat(Path filePath, String lineDelimiter, String fieldDelimiter, CompositeType<OUT>
typeInformation) {
    -		super(filePath, lineDelimiter, fieldDelimiter, typeInformation);
    +
    +	@Override
    +	public void open(FileInputSplit split) throws IOException {
    +		super.open(split);
    +
    +		@SuppressWarnings("unchecked")
    +		FieldParser<Object>[] fieldParsers = (FieldParser<Object>[]) getFieldParsers();
    +
    +		//throw exception if no field parsers are available
    +		if (fieldParsers.length == 0) {
    +			throw new IOException("CsvInputFormat.open(FileInputSplit split) - no field parsers
to parse input");
    +		}
    +
    +		// create the value holders
    +		this.parsedValues = new Object[fieldParsers.length];
    +		for (int i = 0; i < fieldParsers.length; i++) {
    +			this.parsedValues[i] = fieldParsers[i].createValue();
    +		}
    +
    +		// left to right evaluation makes access [0] okay
    +		// this marker is used to fasten up readRecord, so that it doesn't have to check each
call if the line ending is set to default
    +		if (this.getDelimiter().length == 1 && this.getDelimiter()[0] == '\n' ) {
    +			this.lineDelimiterIsLinebreak = true;
    +		}
    +
    +		this.commentCount = 0;
    +		this.invalidLineCount = 0;
     	}
     
     	@Override
    -	protected OUT createTuple(OUT reuse) {
    -		Tuple result = (Tuple) reuse;
    -		for (int i = 0; i < parsedValues.length; i++) {
    -			result.setField(parsedValues[i], i);
    +	public OUT nextRecord(OUT record) throws IOException {
    +		OUT returnRecord = null;
    +		do {
    +			returnRecord = super.nextRecord(record);
    +		} while (returnRecord == null && !reachedEnd());
    +
    +		return returnRecord;
    +	}
    +
    +	public Class<?>[] getFieldTypes() {
    +		return super.getGenericFieldTypes();
    +	}
    +
    +	protected static boolean[] createDefaultMask(int size) {
    --- End diff --
    
    Isn't the default that fields are read one after the other from the start of a line?
    Why do we need this method then?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

Mime
View raw message