一文详解！flink的CsvReader解析，还看不懂我就没法了_flink csvreader

序

本文主要研究一下flink的CsvReader

实例

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 
        DataSet<RecordDto> csvInput = env.readCsvFile(csvFilePath)
                .pojoType(RecordDto.class, "playerName", "country", "year", "game", "gold", "silver", "bronze", "total");
 
        csvInput.map(new MapFunction<RecordDto, RecordDto>() {
            @Override
            public RecordDto map(RecordDto value) throws Exception {
                LOGGER.info("execute map:{}",value);
                TimeUnit.SECONDS.sleep(5);
                return value;
            }
        }).print();

ExecutionEnvironment.readCsvFile

flink-java-1.6.2-sources.jar！/org/apache/flink/api/java/ExecutionEnvironment.java

	/**
	 * Creates a CSV reader to read a comma separated value (CSV) file. The reader has options to
	 * define parameters and field types and will eventually produce the DataSet that corresponds to
	 * the read and parsed CSV input.
	 *
	 * @param filePath The path of the CSV file.
	 * @return A CsvReader that can be used to configure the CSV input.
	 */
	public CsvReader readCsvFile(String filePath) {
		return new CsvReader(filePath, this);
	}

• 这里根据filePath创建了CsvReader

CsvReader

flink-java-1.6.2-sources.jar！/org/apache/flink/api/java/io/CsvReader.java

	public CsvReader(String filePath, ExecutionEnvironment executionContext) {
		this(new Path(Preconditions.checkNotNull(filePath, "The file path may not be null.")), executionContext);
	}
 
	public CsvReader(Path filePath, ExecutionEnvironment executionContext) {
		Preconditions.checkNotNull(filePath, "The file path may not be null.");
		Preconditions.checkNotNull(executionContext, "The execution context may not be null.");
 
		this.path = filePath;
		this.executionContext = executionContext;
	}
 
	/**
	 * Configures the reader to read the CSV data and parse it to the given type. The all fields of the type
	 * must be public or able to set value. The type information for the fields is obtained from the type class.
	 *
	 * @param pojoType The class of the target POJO.
	 * @param pojoFields The fields of the POJO which are mapped to CSV fields.
	 * @return The DataSet representing the parsed CSV data.
	 */
	public <T> DataSource<T> pojoType(Class<T> pojoType, String... pojoFields) {
		Preconditions.checkNotNull(pojoType, "The POJO type class must not be null.");
		Preconditions.checkNotNull(pojoFields, "POJO fields must be specified (not null) if output type is a POJO.");
 
		final TypeInformation<T> ti = TypeExtractor.createTypeInfo(pojoType);
		if (!(ti instanceof PojoTypeInfo)) {
			throw new IllegalArgumentException(
				"The specified class is not a POJO. The type class must meet the POJO requirements. Found: " + ti);
		}
		final PojoTypeInfo<T> pti = (PojoTypeInfo<T>) ti;
 
		CsvInputFormat<T> inputFormat = new PojoCsvInputFormat<T>(path, this.lineDelimiter, this.fieldDelimiter, pti, pojoFields, this.includedMask);
 
		configureInputFormat(inputFormat);
 
		return new DataSource<T>(executionContext, inputFormat, pti, Utils.getCallLocationName());
	}

• CsvReader提供了pojoType方法，用于将csv的数据映射为java类型，同时转换为flink的DataSource；创建DataSource的时候，这里提供了PojoCsvInputFormat和PojoTypeInfo

任务

flink-runtime_2.11-1.6.2-sources.jar！/org/apache/flink/runtime/taskmanager/Task.java

/**
 * The Task represents one execution of a parallel subtask on a TaskManager.
 * A Task wraps a Flink operator (which may be a user function) and
 * runs it, providing all services necessary for example to consume input data,
 * produce its results (intermediate result partitions) and communicate
 * with the JobManager.
 *
 * <p>The Flink operators (implemented as subclasses of
 * {@link AbstractInvokable} have only data readers, -writers, and certain event callbacks.
 * The task connects those to the network stack and actor messages, and tracks the state
 * of the execution and handles exceptions.
 *
 * <p>Tasks have no knowledge about how they relate to other tasks, or whether they
 * are the first attempt to execute the task, or a repeated attempt. All of that
 * is only known to the JobManager. All the task knows are its own runnable code,
 * the task's configuration, and the IDs of the intermediate results to consume and
 * produce (if any).
 *
 * <p>Each Task is run by one dedicated thread.
 */
public class Task implements Runnable, TaskActions, CheckpointListener {
    //......
 
    /**
     * The core work method that bootstraps the task and executes its code.
     */
    @Override
    public void run() {
            //......
            // now load and instantiate the task's invokable code
            invokable = loadAndInstantiateInvokable(userCodeClassLoader, nameOfInvokableClass, env);
 
            // ----------------------------------------------------------------
            //  actual task core work
            // ----------------------------------------------------------------
 
            // we must make strictly sure that the invokable is accessible to the cancel() call
            // by the time we switched to running.
            this.invokable = invokable;
 
            // switch to the RUNNING state, if that fails, we have been canceled/failed in the meantime
            if (!transitionState(ExecutionState.DEPLOYING, ExecutionState.RUNNING)) {
                throw new CancelTaskException();
            }
 
            // notify everyone that we switched to running
            notifyObservers(ExecutionState.RUNNING, null);
            taskManagerActions.updateTaskExecutionState(new TaskExecutionState(jobId, executionId, ExecutionState.RUNNING));
 
            // make sure the user code classloader is accessible thread-locally
            executingThread.setContextClassLoader(userCodeClassLoader);
 
            // run the invokable
            invokable.invoke();
 
            //......
    }
}

• Task的运行方法会调用invokable.invoke（），这里的invokable为DataSourceTask

DataSourceTask.invoke

flink-runtime_2.11-1.6.2-sources.jar！/org/apache/flink/runtime/operators/DataSourceTask.java

	@Override
	public void invoke() throws Exception {
		// --------------------------------------------------------------------
		// Initialize
		// --------------------------------------------------------------------
		initInputFormat();
 
		LOG.debug(getLogString("Start registering input and output"));
 
		try {
			initOutputs(getUserCodeClassLoader());
		} catch (Exception ex) {
			throw new RuntimeException("The initialization of the DataSource's outputs caused an error: " +
					ex.getMessage(), ex);
		}
 
		LOG.debug(getLogString("Finished registering input and output"));
 
		// --------------------------------------------------------------------
		// Invoke
		// --------------------------------------------------------------------
		LOG.debug(getLogString("Starting data source operator"));
 
		RuntimeContext ctx = createRuntimeContext();
 
		final Counter numRecordsOut;
		{
			Counter tmpNumRecordsOut;
			try {
				OperatorIOMetricGroup ioMetricGroup = ((OperatorMetricGroup) ctx.getMetricGroup()).getIOMetricGroup();
				ioMetricGroup.reuseInputMetricsForTask();
				if (this.config.getNumberOfChainedStubs() == 0) {
					ioMetricGroup.reuseOutputMetricsForTask();
				}
				tmpNumRecordsOut = ioMetricGroup.getNumRecordsOutCounter();
			} catch (Exception e) {
				LOG.warn("An exception occurred during the metrics setup.", e);
				tmpNumRecordsOut = new SimpleCounter();
			}
			numRecordsOut = tmpNumRecordsOut;
		}
		
		Counter completedSplitsCounter = ctx.getMetricGroup().counter("numSplitsProcessed");
 
		if (RichInputFormat.class.isAssignableFrom(this.format.getClass())) {
			((RichInputFormat) this.format).setRuntimeContext(ctx);
			LOG.debug(getLogString("Rich Source detected. Initializing runtime context."));
			((RichInputFormat) this.format).openInputFormat();
			LOG.debug(getLogString("Rich Source detected. Opening the InputFormat."));
		}
 
		ExecutionConfig executionConfig = getExecutionConfig();
 
		boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled();
 
		LOG.debug("DataSourceTask object reuse: " + (objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
		
		final TypeSerializer<OT> serializer = this.serializerFactory.getSerializer();
		
		try {
			// start all chained tasks
			BatchTask.openChainedTasks(this.chainedTasks, this);
			
			// get input splits to read
			final Iterator<InputSplit> splitIterator = getInputSplits();
			
			// for each assigned input split
			while (!this.taskCanceled && splitIterator.hasNext())
			{
				// get start and end
				final InputSplit split = splitIterator.next();
 
				LOG.debug(getLogString("Opening input split " + split.toString()));
				
				final InputFormat<OT, InputSplit> format = this.format;
			
				// open input format
				format.open(split);
	
				LOG.debug(getLogString("Starting to read input from split " + split.toString()));
				
				try {
					final Collector<OT> output = new CountingCollector<>(this.output, numRecordsOut);
 
					if (objectReuseEnabled) {
						OT reuse = serializer.createInstance();
 
						// as long as there is data to read
						while (!this.taskCanceled && !format.reachedEnd()) {
 
							OT returned;
							if ((returned = format.nextRecord(reuse)) != null) {
								output.collect(returned);
							}
						}
					} else {
						// as long as there is data to read
						while (!this.taskCanceled && !format.reachedEnd()) {
							OT returned;
							if ((returned = format.nextRecord(serializer.createInstance())) != null) {
								output.collect(returned);
							}
						}
					}
 
					if (LOG.isDebugEnabled() && !this.taskCanceled) {
						LOG.debug(getLogString("Closing input split " + split.toString()));
					}
				} finally {
					// close. We close here such that a regular close throwing an exception marks a task as failed.
					format.close();
				}
				completedSplitsCounter.inc();
			} // end for all input splits
 
			// close the collector. if it is a chaining task collector, it will close its chained tasks
			this.output.close();
 
			// close all chained tasks letting them report failure
			BatchTask.closeChainedTasks(this.chainedTasks, this);
 
		}
		catch (Exception ex) {
			// close the input, but do not report any exceptions, since we already have another root cause
			try {
				this.format.close();
			} catch (Throwable ignored) {}
 
			BatchTask.cancelChainedTasks(this.chainedTasks);
 
			ex = ExceptionInChainedStubException.exceptionUnwrap(ex);
 
			if (ex instanceof CancelTaskException) {
				// forward canceling exception
				throw ex;
			}
			else if (!this.taskCanceled) {
				// drop exception, if the task was canceled
				BatchTask.logAndThrowException(ex, this);
			}
		} finally {
			BatchTask.clearWriters(eventualOutputs);
			// --------------------------------------------------------------------
			// Closing
			// --------------------------------------------------------------------
			if (this.format != null && RichInputFormat.class.isAssignableFrom(this.format.getClass())) {
				((RichInputFormat) this.format).closeInputFormat();
				LOG.debug(getLogString("Rich Source detected. Closing the InputFormat."));
			}
		}
 
		if (!this.taskCanceled) {
			LOG.debug(getLogString("Finished data source operator"));
		}
		else {
			LOG.debug(getLogString("Data source operator cancelled"));
		}
	}

• DataSourceTask的调用方法这里只要不是taskCanceled及format.reachedEnd（），都会调用format.nextRecord（serializer.createInstance（））来拉取数据，然后执行output.collect（返回）
• 这里的format为CsvInputFormat（PojoCsvInputFormat），不过nextRecord以及reachedEnd方法它是调用的父类DelimitedInputFormat
• PojoCsvInputFormat继承了抽象类CsvInputFormat，而CsvInputFormat继承了抽象类GenericCsvInputFormat，GenericCsvInputFormat则继承了抽象类DelimitedInputFormat

DelimitedInputFormat

flink-core-1.6.2-sources.jar！/org/apache/flink/api/common/io/DelimitedInputFormat.java

	/**
	 * The default read buffer size = 1MB.
	 */
	private static final int DEFAULT_READ_BUFFER_SIZE = 1024 * 1024;
 
	private transient byte[] readBuffer;
 
	private int bufferSize = -1;
 
	private void initBuffers() {
		this.bufferSize = this.bufferSize <= 0 ? DEFAULT_READ_BUFFER_SIZE : this.bufferSize;
 
		if (this.bufferSize <= this.delimiter.length) {
			throw new IllegalArgumentException("Buffer size must be greater than length of delimiter.");
		}
 
		if (this.readBuffer == null || this.readBuffer.length != this.bufferSize) {
			this.readBuffer = new byte[this.bufferSize];
		}
		if (this.wrapBuffer == null || this.wrapBuffer.length < 256) {
			this.wrapBuffer = new byte[256];
		}
 
		this.readPos = 0;
		this.limit = 0;
		this.overLimit = false;
		this.end = false;
	}
 
	/**
	 * Checks whether the current split is at its end.
	 * 
	 * @return True, if the split is at its end, false otherwise.
	 */
	@Override
	public boolean reachedEnd() {
		return this.end;
	}
	
	@Override
	public OT nextRecord(OT record) throws IOException {
		if (readLine()) {
			return readRecord(record, this.currBuffer, this.currOffset, this.currLen);
		} else {
			this.end = true;
			return null;
		}
	}
 
	/**
	 * Fills the read buffer with bytes read from the file starting from an offset.
	 */
	private boolean fillBuffer(int offset) throws IOException {
		int maxReadLength = this.readBuffer.length - offset;
		// special case for reading the whole split.
		if (this.splitLength == FileInputFormat.READ_WHOLE_SPLIT_FLAG) {
			int read = this.stream.read(this.readBuffer, offset, maxReadLength);
			if (read == -1) {
				this.stream.close();
				this.stream = null;
				return false;
			} else {
				this.readPos = offset;
				this.limit = read;
				return true;
			}
		}
		
		// else ..
		int toRead;
		if (this.splitLength > 0) {
			// if we have more data, read that
			toRead = this.splitLength > maxReadLength ? maxReadLength : (int) this.splitLength;
		}
		else {
			// if we have exhausted our split, we need to complete the current record, or read one
			// more across the next split.
			// the reason is that the next split will skip over the beginning until it finds the first
			// delimiter, discarding it as an incomplete chunk of data that belongs to the last record in the
			// previous split.
			toRead = maxReadLength;
			this.overLimit = true;
		}
 
		int read = this.stream.read(this.readBuffer, offset, toRead);
 
		if (read == -1) {
			this.stream.close();
			this.stream = null;
			return false;
		} else {
			this.splitLength -= read;
			this.readPos = offset; // position from where to start reading
			this.limit = read + offset; // number of valid bytes in the read buffer
			return true;
		}
	}

• DelimitedInputFormat首先调用readLine（）读取数据到currBuffer，如果有数据，则调用子类CsvInputFormat实现的readRecord方法，此处传递了currBuffer，currOffset，currLen
• DelimitedInputFormat的readLine（）方法里头会调用fillBuffer方法，fillBuffer方法会根据splitLength（DelimitedInputFormat.getStatistics方法里头FileInputSplit的length）及maxReadLength来确定要读取，然后从偏移开始到读取到从文件读取数据到readBuffer中，然后设置currBuffer，currOffset，currLen
• readBuffer在初始的时候会设置bufferSize，bufferSize初始化的时候为-1，在getStatistics方法里头被设置为4 * 1024，而DEFAULT_READ_BUFFER_SIZE是1024 * 1024

CsvInputFormat.readRecord

flink-java-1.6.2-sources.jar！/org/apache/flink/api/java/io/CsvInputFormat.java

	@Override
	public OUT readRecord(OUT reuse, byte[] bytes, int offset, int numBytes) throws IOException {
		/*
		 * Fix to support windows line endings in CSVInputFiles with standard delimiter setup = \n
		 */
		// Found window's end line, so find carriage return before the newline
		if (this.lineDelimiterIsLinebreak && numBytes > 0 && bytes[offset + numBytes - 1] == '\r') {
			//reduce the number of bytes so that the Carriage return is not taken as data
			numBytes--;
		}
 
		if (commentPrefix != null && commentPrefix.length <= numBytes) {
			//check record for comments
			boolean isComment = true;
			for (int i = 0; i < commentPrefix.length; i++) {
				if (commentPrefix[i] != bytes[offset + i]) {
					isComment = false;
					break;
				}
			}
			if (isComment) {
				this.commentCount++;
				return null;
			}
		}
 
		if (parseRecord(parsedValues, bytes, offset, numBytes)) {
			return fillRecord(reuse, parsedValues);
		} else {
			this.invalidLineCount++;
			return null;
		}
	}

• CsvInputFormat的readRecord方法负责读取原始数据，之后通过parseRecord方法解析原始数据填充到parsedValues（Object[]），之后调用子类的fillRecord方法（这里是PojoCsvInputFormat）将parsededValues填充到重用对象（该对象是DataSourceTask在调用format.nextRecord时传入的serializer.createInstance()）

PojoCsvInputFormat.fillRecord

flink-java-1.6.2-sources.jar！/org/apache/flink/api/java/io/PojoCsvInputFormat.java

/**
 * Input format that reads csv into POJOs.
 * @param <OUT> resulting POJO type
 */
@Internal
public class PojoCsvInputFormat<OUT> extends CsvInputFormat<OUT> {
 
	//......
 
	@Override
	public void open(FileInputSplit split) throws IOException {
		super.open(split);
 
		pojoFields = new Field[pojoFieldNames.length];
 
		Map<String, Field> allFields = new HashMap<String, Field>();
 
		findAllFields(pojoTypeClass, allFields);
 
		for (int i = 0; i < pojoFieldNames.length; i++) {
			pojoFields[i] = allFields.get(pojoFieldNames[i]);
 
			if (pojoFields[i] != null) {
				pojoFields[i].setAccessible(true);
			} else {
				throw new RuntimeException("There is no field called \"" + pojoFieldNames[i] + "\" in " + pojoTypeClass.getName());
			}
		}
	}
 
	@Override
	public OUT fillRecord(OUT reuse, Object[] parsedValues) {
		for (int i = 0; i < parsedValues.length; i++) {
			try {
				pojoFields[i].set(reuse, parsedValues[i]);
			} catch (IllegalAccessException e) {
				throw new RuntimeException("Parsed value could not be set in POJO field \"" + pojoFieldNames[i] + "\"", e);
			}
		}
		return reuse;
	}
 
	//......
}

• PojoCsvInputFormat的open方法用于在执行者的executePlan的时候调用，提前使用反射获取所需的字段
• fillRecord方法此处可以使用反射将解析的值设置到pojo中
• 如果反射设置不成功则引发IllegalAccessException异常

CountingCollector.collect

flink-runtime_2.11-1.6.2-sources.jar！/org/apache/flink/runtime/operators/util/metrics/CountingCollector.java

public class CountingCollector<OUT> implements Collector<OUT> {
	private final Collector<OUT> collector;
	private final Counter numRecordsOut;
 
	public CountingCollector(Collector<OUT> collector, Counter numRecordsOut) {
		this.collector = collector;
		this.numRecordsOut = numRecordsOut;
	}
 
	@Override
	public void collect(OUT record) {
		this.numRecordsOut.inc();
		this.collector.collect(record);
	}
 
	@Override
	public void close() {
		this.collector.close();
	}
}

• 这里的collector为org.apache.flink.runtime.operators.chaining.ChainedMapDriver

ChainedMapDriver

flink-runtime_2.11-1.6.2-sources.jar！/org/apache/flink/runtime/operators/chaining/ChainedMapDriver.java

	@Override
	public void collect(IT record) {
		try {
			this.numRecordsIn.inc();
			this.outputCollector.collect(this.mapper.map(record));
		} catch (Exception ex) {
			throw new ExceptionInChainedStubException(this.taskName, ex);
		}
	}

• 这里会先调用mapper的地图方法，执行地图逻辑，然后调用outputCollector.collect将结果发送出去
• 这里的outputCollector为CountingCollector，它里头包装的collector为org.apache.flink.runtime.operators.shipping.OutputCollector

输出收集器

flink-runtime_2.11-1.6.2-sources.jar！/org/apache/flink/runtime/operators/shipping/OutputCollector.java

	/**
	 * Collects a record and emits it to all writers.
	 */
	@Override
	public void collect(T record)  {
		if (record != null) {
			this.delegate.setInstance(record);
			try {
				for (RecordWriter<SerializationDelegate<T>> writer : writers) {
					writer.emit(this.delegate);
				}
			}
			catch (IOException e) {
				throw new RuntimeException("Emitting the record caused an I/O exception: " + e.getMessage(), e);
			}
			catch (InterruptedException e) {
				throw new RuntimeException("Emitting the record was interrupted: " + e.getMessage(), e);
			}
		}
		else {
			throw new NullPointerException("The system does not support records that are null."
								+ "Null values are only supported as fields inside other objects.");
		}
	}

• 这里调用RecordWriter的emit方法来发射数据

记录作家

flink-runtime_2.11-1.6.2-sources.jar！/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java

	public void emit(T record) throws IOException, InterruptedException {
		for (int targetChannel : channelSelector.selectChannels(record, numChannels)) {
			sendToTarget(record, targetChannel);
		}
	}

• 这里通过channelSelector.selectChannels返回要发送的targetChannel，这里的channelSelector为OutputEmitter

输出发射器

flink-runtime_2.11-1.6.2-sources.jar！/org/apache/flink/runtime/operators/shipping/OutputEmitter.java

	@Override
	public final int[] selectChannels(SerializationDelegate<T> record, int numberOfChannels) {
		switch (strategy) {
		case FORWARD:
			return forward();
		case PARTITION_RANDOM:
		case PARTITION_FORCED_REBALANCE:
			return robin(numberOfChannels);
		case PARTITION_HASH:
			return hashPartitionDefault(record.getInstance(), numberOfChannels);
		case BROADCAST:
			return broadcast(numberOfChannels);
		case PARTITION_CUSTOM:
			return customPartition(record.getInstance(), numberOfChannels);
		case PARTITION_RANGE:
			return rangePartition(record.getInstance(), numberOfChannels);
		default:
			throw new UnsupportedOperationException("Unsupported distribution strategy: " + strategy.name());
		}
	}
 
	private int[] forward() {
		return this.channels;
	}

• 这里的strategy为FORWARD

小结

• CsvReader创建的inputFormat为PojoCsvInputFormat，它主要的方法是fillRecord，利用反射填充数据，而数据的读取则是在DelimitedInputFormat的readLine方法中，它会调用fillBuffer方法，而fillBuffer方法会根据splitLength（DelimitedInputFormat.getStatistics方法里头FileInputSplit的length）以及maxReadLength来确定toRead，之后从偏移开始到toRead从文件读取数据到readBuffer中
• DataSourceTask在调用方法里头会不断循环调用format.nextRecord，然后挨个调用output.collect方法（包装了org.apache.flink.runtime.operators.shipping.OutputCollector的CountingCollector），直到taskCanceled或者format.reachedEnd（）
• output.collect方法，这里的输出为CountingCollector，它代理的collector为ChainedMapDriver； ChainedMapDriver重新读取的数据进行地图操作，最后将地图的结果传递给代理了OutputCollector的CountingCollector，OutputCollector使用RecordWriter来发射数据

doc

• CsvReader