import org.apache.flink.api.java.typeutils.InputTypeConfigurable; //导入依赖的package包/类
/**
 * Adds the given sink to this DataStream. Only streams with sinks added
 * will be executed once the {@link StreamExecutionEnvironment#execute()}
 * method is called.
 *
 * @param sinkFunction
 *            The object containing the sink's invoke function.
 * @return The closed DataStream.
 */
public DataStreamSink<T> addSink(SinkFunction<T> sinkFunction) {

	// read the output type of the input Transform to coax out errors about MissingTypeInfo
	transformation.getOutputType();

	// configure the type if needed
	if (sinkFunction instanceof InputTypeConfigurable) {
		((InputTypeConfigurable) sinkFunction).setInputType(getType(), getExecutionConfig());
	}

	StreamSink<T> sinkOperator = new StreamSink<>(clean(sinkFunction));

	DataStreamSink<T> sink = new DataStreamSink<>(this, sinkOperator);

	getExecutionEnvironment().addOperator(sink.getTransformation());
	return sink;
}
 

import org.apache.flink.api.java.typeutils.InputTypeConfigurable; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void setInputType(TypeInformation<?> type, ExecutionConfig executionConfig) {
    if (this.writerTemplate instanceof InputTypeConfigurable) {
        ((InputTypeConfigurable) writerTemplate).setInputType(type, executionConfig);
    }
}
 

import org.apache.flink.api.java.typeutils.InputTypeConfigurable; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void setInputType(TypeInformation<?> type, ExecutionConfig executionConfig) {
	if (this.writerTemplate instanceof InputTypeConfigurable) {
		((InputTypeConfigurable) writerTemplate).setInputType(type, executionConfig);
	}
}
 

import org.apache.flink.api.java.typeutils.InputTypeConfigurable; //导入依赖的package包/类
/**
 * Emits a DataSet using an {@link OutputFormat}. This method adds a data sink to the program.
 * Programs may have multiple data sinks. A DataSet may also have multiple consumers (data sinks
 * or transformations) at the same time.
 *
 * @param outputFormat The OutputFormat to process the DataSet.
 * @return The DataSink that processes the DataSet.
 *
 * @see OutputFormat
 * @see DataSink
 */
public DataSink<T> output(OutputFormat<T> outputFormat) {
	Preconditions.checkNotNull(outputFormat);

	// configure the type if needed
	if (outputFormat instanceof InputTypeConfigurable) {
		((InputTypeConfigurable) outputFormat).setInputType(getType(), context.getConfig());
	}

	DataSink<T> sink = new DataSink<>(this, outputFormat, getType());
	this.context.registerDataSink(sink);
	return sink;
}
 

import org.apache.flink.api.java.typeutils.InputTypeConfigurable; //导入依赖的package包/类
@Override
public void setInputType(TypeInformation<?> type, ExecutionConfig executionConfig) {
	if (format instanceof InputTypeConfigurable) {
		InputTypeConfigurable itc = (InputTypeConfigurable) format;
		itc.setInputType(type, executionConfig);
	}
}
 

import org.apache.flink.api.java.typeutils.InputTypeConfigurable; //导入依赖的package包/类
public <IN, OUT> void addOperator(
		Integer vertexID,
		String slotSharingGroup,
		StreamOperator<OUT> operatorObject,
		TypeInformation<IN> inTypeInfo,
		TypeInformation<OUT> outTypeInfo,
		String operatorName) {

	if (operatorObject instanceof StoppableStreamSource) {
		addNode(vertexID, slotSharingGroup, StoppableSourceStreamTask.class, operatorObject, operatorName);
	} else if (operatorObject instanceof StreamSource) {
		addNode(vertexID, slotSharingGroup, SourceStreamTask.class, operatorObject, operatorName);
	} else {
		addNode(vertexID, slotSharingGroup, OneInputStreamTask.class, operatorObject, operatorName);
	}

	TypeSerializer<IN> inSerializer = inTypeInfo != null && !(inTypeInfo instanceof MissingTypeInfo) ? inTypeInfo.createSerializer(executionConfig) : null;

	TypeSerializer<OUT> outSerializer = outTypeInfo != null && !(outTypeInfo instanceof MissingTypeInfo) ? outTypeInfo.createSerializer(executionConfig) : null;

	setSerializers(vertexID, inSerializer, null, outSerializer);

	if (operatorObject instanceof OutputTypeConfigurable && outTypeInfo != null) {
		@SuppressWarnings("unchecked")
		OutputTypeConfigurable<OUT> outputTypeConfigurable = (OutputTypeConfigurable<OUT>) operatorObject;
		// sets the output type which must be know at StreamGraph creation time
		outputTypeConfigurable.setOutputType(outTypeInfo, executionConfig);
	}

	if (operatorObject instanceof InputTypeConfigurable) {
		InputTypeConfigurable inputTypeConfigurable = (InputTypeConfigurable) operatorObject;
		inputTypeConfigurable.setInputType(inTypeInfo, executionConfig);
	}

	if (LOG.isDebugEnabled()) {
		LOG.debug("Vertex: {}", vertexID);
	}
}
 

import org.apache.flink.api.java.typeutils.InputTypeConfigurable; //导入依赖的package包/类
/**
 * Emits a DataSet using an {@link OutputFormat}. This method adds a data sink to the program.
 * Programs may have multiple data sinks. A DataSet may also have multiple consumers (data sinks
 * or transformations) at the same time.
 * 
 * @param outputFormat The OutputFormat to process the DataSet.
 * @return The DataSink that processes the DataSet.
 * 
 * @see OutputFormat
 * @see DataSink
 */
public DataSink<T> output(OutputFormat<T> outputFormat) {
	Validate.notNull(outputFormat);
	
	// configure the type if needed
	if (outputFormat instanceof InputTypeConfigurable) {
		((InputTypeConfigurable) outputFormat).setInputType(this.type);
	}
	
	DataSink<T> sink = new DataSink<T>(this, outputFormat, this.type);
	this.context.registerDataSink(sink);
	return sink;
}