import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
public ContinuousFileMonitoringFunction(
	FileInputFormat<OUT> format,
	FileProcessingMode watchType,
	int readerParallelism,
	long interval) {

	Preconditions.checkArgument(
		watchType == FileProcessingMode.PROCESS_ONCE || interval >= MIN_MONITORING_INTERVAL,
		"The specified monitoring interval (" + interval + " ms) is smaller than the minimum " +
			"allowed one (" + MIN_MONITORING_INTERVAL + " ms)."
	);

	this.format = Preconditions.checkNotNull(format, "Unspecified File Input Format.");
	this.path = Preconditions.checkNotNull(format.getFilePath().toString(), "Unspecified Path.");

	this.interval = interval;
	this.watchType = watchType;
	this.readerParallelism = Math.max(readerParallelism, 1);
	this.globalModificationTime = Long.MIN_VALUE;
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
public <X> DataSource<X> readFile(FileInputFormat<X> inputFormat, String filePath) {
	if (inputFormat == null) {
		throw new IllegalArgumentException("InputFormat must not be null.");
	}
	if (filePath == null) {
		throw new IllegalArgumentException("The file path must not be null.");
	}

	inputFormat.setFilePath(new Path(filePath));
	try {
		return createInput(inputFormat, TypeExtractor.getInputFormatTypes(inputFormat));
	}
	catch (Exception e) {
		throw new InvalidProgramException("The type returned by the input format could not be automatically determined. " +
				"Please specify the TypeInformation of the produced type explicitly by using the " +
				"'createInput(InputFormat, TypeInformation)' method instead.");
	}
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
private SplitReader(FileInputFormat<OT> format,
			TypeSerializer<OT> serializer,
			SourceFunction.SourceContext<OT> readerContext,
			Object checkpointLock,
			List<TimestampedFileInputSplit> restoredState) {

	this.format = checkNotNull(format, "Unspecified FileInputFormat.");
	this.serializer = checkNotNull(serializer, "Unspecified Serializer.");
	this.readerContext = checkNotNull(readerContext, "Unspecified Reader Context.");
	this.checkpointLock = checkNotNull(checkpointLock, "Unspecified checkpoint lock.");

	this.shouldClose = false;
	this.isRunning = true;

	this.pendingSplits = new PriorityQueue<>();

	// this is the case where a task recovers from a previous failed attempt
	if (restoredState != null) {
		this.pendingSplits.addAll(restoredState);
	}
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
@Override
public List<PlanNode> getAlternativePlans(CostEstimator estimator) {
	if (this.cachedPlans != null) {
		return this.cachedPlans;
	}
	
	SourcePlanNode candidate = new SourcePlanNode(this, "DataSource ("+this.getPactContract().getName()+")");
	candidate.updatePropertiesWithUniqueSets(getUniqueFields());
	
	final Costs costs = new Costs();
	if (FileInputFormat.class.isAssignableFrom(getPactContract().getFormatWrapper().getUserCodeClass()) &&
			this.estimatedOutputSize >= 0)
	{
		estimator.addFileInputCost(this.estimatedOutputSize, costs);
	}
	candidate.setCosts(costs);

	// since there is only a single plan for the data-source, return a list with that element only
	List<PlanNode> plans = new ArrayList<PlanNode>(1);
	plans.add(candidate);

	this.cachedPlans = plans;
	return plans;
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
public <X> DataSource<X> readFile(FileInputFormat<X> inputFormat, String filePath) {
	if (inputFormat == null) {
		throw new IllegalArgumentException("InputFormat must not be null.");
	}
	if (filePath == null) {
		throw new IllegalArgumentException("The file path must not be null.");
	}
	
	inputFormat.setFilePath(new Path(filePath));
	try {
		return createInput(inputFormat, TypeExtractor.getInputFormatTypes(inputFormat));
	}
	catch (Exception e) {
		throw new InvalidProgramException("The type returned by the input format could not be automatically determined. " +
				"Please specify the TypeInformation of the produced type explicitly.");
	}
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
private <OUT> DataStreamSource<OUT> createFileInput(FileInputFormat<OUT> inputFormat,
													TypeInformation<OUT> typeInfo,
													String sourceName,
													FileProcessingMode monitoringMode,
													long interval) {

	Preconditions.checkNotNull(inputFormat, "Unspecified file input format.");
	Preconditions.checkNotNull(typeInfo, "Unspecified output type information.");
	Preconditions.checkNotNull(sourceName, "Unspecified name for the source.");
	Preconditions.checkNotNull(monitoringMode, "Unspecified monitoring mode.");

	Preconditions.checkArgument(monitoringMode.equals(FileProcessingMode.PROCESS_ONCE) ||
			interval >= ContinuousFileMonitoringFunction.MIN_MONITORING_INTERVAL,
		"The path monitoring interval cannot be less than " +
				ContinuousFileMonitoringFunction.MIN_MONITORING_INTERVAL + " ms.");

	ContinuousFileMonitoringFunction<OUT> monitoringFunction =
		new ContinuousFileMonitoringFunction<>(inputFormat, monitoringMode, getParallelism(), interval);

	ContinuousFileReaderOperator<OUT> reader =
		new ContinuousFileReaderOperator<>(inputFormat);

	SingleOutputStreamOperator<OUT> source = addSource(monitoringFunction, sourceName)
			.transform("Split Reader: " + sourceName, typeInfo, reader);

	return new DataStreamSource<>(source);
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/**
 * Creates a new instance for the given file using the given file input format.
 *
 * @param f The {@link org.apache.flink.api.common.io.FileInputFormat} implementation used to read the data.
 * @param operatorInfo The type information for the output type.
 * @param filePath The file location. The file path must be a fully qualified URI, including the address schema.
 * @param name The given name for the Pact, used in plans, logs and progress messages.
 */
public FileDataSourceBase(FileInputFormat<OUT> f, OperatorInformation<OUT> operatorInfo, String filePath, String name) {
	super(f, operatorInfo, name);

	Preconditions.checkNotNull(filePath, "The file path may not be null.");

	this.filePath = filePath;
	f.setFilePath(filePath);
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
private static void initializeIOFormatClasses() {
	try {
		Method im = FileInputFormat.class.getDeclaredMethod("initDefaultsFromConfiguration");
		im.setAccessible(true);
		im.invoke(null);
		
		Method om = FileOutputFormat.class.getDeclaredMethod("initDefaultsFromConfiguration");
		om.setAccessible(true);
		om.invoke(null);
	}
	catch (Exception e) {
		LOG.error("Cannot (re) initialize the globally loaded defaults. Some classes might mot follow the specified default behavior.");
	}
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/**
 * Reads the contents of the user-specified {@code filePath} based on the given {@link FileInputFormat}. Depending
 * on the provided {@link FileProcessingMode}.
 *
 * <p>See {@link #readFile(FileInputFormat, String, FileProcessingMode, long)}
 *
 * @param inputFormat
 * 		The input format used to create the data stream
 * @param filePath
 * 		The path of the file, as a URI (e.g., "file:///some/local/file" or "hdfs://host:port/file/path")
 * @param watchType
 * 		The mode in which the source should operate, i.e. monitor path and react to new data, or process once and exit
 * @param interval
 * 		In the case of periodic path monitoring, this specifies the interval (in millis) between consecutive path scans
 * @param filter
 * 		The files to be excluded from the processing
 * @param <OUT>
 * 		The type of the returned data stream
 * @return The data stream that represents the data read from the given file
 *
 * @deprecated Use {@link FileInputFormat#setFilesFilter(FilePathFilter)} to set a filter and
 * 		{@link StreamExecutionEnvironment#readFile(FileInputFormat, String, FileProcessingMode, long)}
 *
 */
@PublicEvolving
@Deprecated
public <OUT> DataStreamSource<OUT> readFile(FileInputFormat<OUT> inputFormat,
											String filePath,
											FileProcessingMode watchType,
											long interval,
											FilePathFilter filter) {
	inputFormat.setFilesFilter(filter);

	TypeInformation<OUT> typeInformation;
	try {
		typeInformation = TypeExtractor.getInputFormatTypes(inputFormat);
	} catch (Exception e) {
		throw new InvalidProgramException("The type returned by the input format could not be " +
				"automatically determined. Please specify the TypeInformation of the produced type " +
				"explicitly by using the 'createInput(InputFormat, TypeInformation)' method instead.");
	}
	return readFile(inputFormat, filePath, watchType, interval, typeInformation);
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/**
 * Generic method to create an input data stream with {@link org.apache.flink.api.common.io.InputFormat}.
 *
 * <p>The data stream is typed to the given TypeInformation. This method is intended for input formats
 * where the return type cannot be determined by reflection analysis, and that do not implement the
 * {@link org.apache.flink.api.java.typeutils.ResultTypeQueryable} interface.
 *
 * <p><b>NOTES ON CHECKPOINTING: </b> In the case of a {@link FileInputFormat}, the source
 * (which executes the {@link ContinuousFileMonitoringFunction}) monitors the path, creates the
 * {@link org.apache.flink.core.fs.FileInputSplit FileInputSplits} to be processed, forwards
 * them to the downstream {@link ContinuousFileReaderOperator} to read the actual data, and exits,
 * without waiting for the readers to finish reading. This implies that no more checkpoint
 * barriers are going to be forwarded after the source exits, thus having no checkpoints.
 *
 * @param inputFormat
 * 		The input format used to create the data stream
 * @param typeInfo
 * 		The information about the type of the output type
 * @param <OUT>
 * 		The type of the returned data stream
 * @return The data stream that represents the data created by the input format
 */
@PublicEvolving
public <OUT> DataStreamSource<OUT> createInput(InputFormat<OUT, ?> inputFormat, TypeInformation<OUT> typeInfo) {
	DataStreamSource<OUT> source;

	if (inputFormat instanceof FileInputFormat) {
		@SuppressWarnings("unchecked")
		FileInputFormat<OUT> format = (FileInputFormat<OUT>) inputFormat;

		source = createFileInput(format, typeInfo, "Custom File source",
				FileProcessingMode.PROCESS_ONCE, -1);
	} else {
		source = createInput(inputFormat, typeInfo, "Custom Source");
	}
	return source;
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
public ContinuousFileReaderOperator(FileInputFormat<OUT> format) {
	this.format = checkNotNull(format);
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/** the Flink input format for this input */
public FileInputFormat<FData> inputFormat ( String path ) { return null; }
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/** the Flink input format for this input */
public FileInputFormat<FData> inputFormat ( String path  ) {
    FDataInputFormat sf = new FDataInputFormat();
    sf.setFilePath(path.toString());
    return sf;
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/** the Flink input format for this input */
public FileInputFormat<FData> inputFormat ( String path ) {
    return new ParsedInputFormat(path);
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/** the Flink input format for this input */
abstract public FileInputFormat<FData> inputFormat ( String path );
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
public static <T extends FileInputFormat<?>> InputFormatVertex createInput(T stub, String path, String name, JobGraph graph,
		int degreeOfParallelism)
{
	stub.setFilePath(path);
	return createInput(new UserCodeObjectWrapper<T>(stub), name, graph, degreeOfParallelism);
}
 

import org.apache.flink.api.common.io.FileInputFormat; //导入依赖的package包/类
/**
 * Creates a new instance for the given file using the given file input format.
 * 
 * @param f The {@link FileInputFormat} implementation used to read the data.
 * @param filePath The file location. The file path must be a fully qualified URI, including the address schema.
 * @param name The given name for the Pact, used in plans, logs and progress messages.
 */
public FileDataSource(FileInputFormat<Record> f, String filePath, String name) {
	super(f, OperatorInfoHelper.source(), filePath, name);
}