StreamExecutionEnvironment

ExecutionConfig

Parallelism Parallelism

1. Have default values , be equal to cpu Number
2. getter and setter Method
3. Maximum parallelism : 0 < maxParallelism <= 2^15 - 1

Restart strategy RestartStrategies

1. setRestartStrategy()
2. getRestartStrategy()

Chain optimization isChainingEnabled

1. isChainingEnabled
2. disableOperatorChaining

Retry count numberOfExecutionRetries

1. setNumberOfExecutionRetries
2. getNumberOfExecutionRetries

TimeCharacteristic

The default is processing time

1. setStreamTimeCharacteristic( Time type )

   1. Processing time AutoWatermarkInterval The watermark interval is 0ms
   2. The time of the event AutoWatermarkInterval The watermark interval is 200ms

CheckpointConfig

1. checkpointingMode :EXACTLY_ONCE、AT_LEAST_ONCE

   1. getCheckpointingMode()

2. checkpointTimeout

3. CheckpointInterval

   1. enableCheckpointing

4. forceCheckpointing

   1. isForceCheckpointing
   2. setForceCheckpointing

5. PauseBetweenCheckpoints

   1. minPauseBetweenCheckpoints
   2. maxPauseBetweenCheckpoints

6. maxConcurrentCheckpoints

7. failOnCheckpointingErrors

   1. isFailOnCheckpointingErrors

8. ExternalizedCheckpointCleanup

   Allow external persistent checkpoints

   Enumeration values are ：

   1. DELETE_ON_CANCELLATION：job Delete persistent checkpoints and states when canceling . After canceling the job , Cannot recover from an external checkpoint .
   2. RETAIN_ON_CANCELLATION：job Keep persistent checkpoints and states when canceling , After canceling the job , Checkpoint metadata and status must be deleted manually

   Method ：

   enableExternalizedCheckpoints(cleanupMode) Set the mode of checkpoint persistent storage

   isExternalizedCheckpointsEnabled() Whether to allow persistence to external storage .

   getExternalizedCheckpointCleanup() Returns the pattern of persistent external storage

StateBackend

1. Describes how to store and set operator States and checkpoints

2. Defines during execution Checkpoint data and state will be persistent data structures （ It can be hashtable、RockDB、 Other data storage ）

The main method

establish environment

1. StreamExecutionEnvironment.getExecutionEnvironment
2. StreamExecutionEnvironment.createLocalEnvironment
3. StreamExecutionEnvironment.createRemoteEnvironment

data source

Collection based data sources

generateSequence

env.generateSequence(long from,long to) Generated from from To to A sequence of numbers , And take the sequence data as the input data source . If the parallelism is set to 1, Then the generated element sequence is ordered .

fromElements

Create a data flow from a given sequence of objects , All objects must be of the same type . The degree of parallelism is 1.

For uncertain generic classes , You may need to display the specified TypeInformation.

fromCollection

Create a data stream from a given non empty set or iterator , The degree of parallelism is 1.

fromParallelCollection

Create a data stream that contains elements in a separable iterator , Allows the framework to create a parallel data stream to return elements in the iterator .

Iterators are not modified until they are actually executed , Therefore, the data type returned by the iterator must be in Type The form of class shows （ because JAVA The compiler removes generic information ）

File based data sources

readTextFile

1. Read a given file line by line , And create a data flow , Contains a string containing the contents of each line . The file will be read using the system's default character set or the specified character set .
2. readTextFile Called readFile(TextInputFormat( route , Character set , Separator ), File processing mode . Deal with it once , The time interval , Read the file as text ) Method
3. readFile() Method is called createFileInput(TextInputFormat( route , Character set , Separator ),typeInformation, File processing mode , The time interval ) Method .
4. createFileInput call addSource() Method .

readFile

According to the specified file format TextInputFormat Read the file .

createFileInput

createFileInput(TextInputFormat( route , Character set , Separator ),typeInformation, File processing mode , The time interval )

//  Create a periodic file scanning monitor with incoming time interval 
ContinuousFileMonitoringFunction<OUT> monitoringFunction = new ContinuousFileMonitoringFunction<>(inputFormat, monitoringMode, getParallelism(), interval);
//  Create a ready operator 
ContinuousFileReaderOperator<OUT> reader = new ContinuousFileReaderOperator<>(inputFormat);
//  Create an input source , Concurrency reading environment Concurrency configuration in .
SingleOutputStreamOperator<OUT> source = addSource(monitoringFunction, sourceName).transform("Split Reader: " + sourceName, typeInfo, reader);

FileProcessingMode There are two kinds of ,

1. PROCESS_ONCE Process the current content on the path , And then quit .
2. PROCESS_CONTINUOUSLY Scan paths regularly for new content .

createInput

according to TextInputFormat A common way to create an input stream

be based on Socket Data source

socketTextStream

From a socket Create a data stream containing the received wireless string , The received string is defaulted by the system decode Of .socket Itself does not report termination , The result is only socket The graceful termination will start retry .

socketTextStream(hostname, port, String.valueOf(delimiter), maxRetry);

addSource Add data sources

Add a data source to the topology of the data flow .

The default concurrency is 1, If you want to execute concurrently , Can achieve ParallelSourceFunction Or inheritance RichParallelSourceFunction
function – the user defined function
sourceName – Name of the data source
typeInfo – the user defined type information for the stream

@SuppressWarnings("unchecked")
public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, String sourceName, TypeInformation<OUT> typeInfo) {
    
	if (typeInfo == null) {
    
		if (function instanceof ResultTypeQueryable) {
    
			typeInfo = ((ResultTypeQueryable<OUT>) function).getProducedType();
		} else {
    
			try {
    
				typeInfo = TypeExtractor.createTypeInfo(
						SourceFunction.class,
						function.getClass(), 0, null, null);
			} catch (final InvalidTypesException e) {
    
				typeInfo = (TypeInformation<OUT>) new MissingTypeInfo(sourceName, e);
			}
		}
	}

	boolean isParallel = function instanceof ParallelSourceFunction;

	clean(function);
	StreamSource<OUT, ?> sourceOperator;
	if (function instanceof StoppableFunction) {
    
		sourceOperator = new StoppableStreamSource<>(cast2StoppableSourceFunction(function));
	} else {
    
		sourceOperator = new StreamSource<>(function);
	}

	return new DataStreamSource<>(this, typeInfo, sourceOperator, isParallel, sourceName);
}

Task run

execute

Trigger program execution . The environment will execute the program sink All upstream and downstream of the operator .sink There are two kinds of operators , Print the result set or send the result set to the message queue .

You can display the specified jobName, The default is “Flink Streaming Job”.

getStreamGraph

Get the of the streaming task streamGraph.

1. take environment Medium State backend , Whether chain optimization , Environmental context Construct a flow chart as a parameter
2. take sinks Before operator transformList To iterate , Set in iteration streamGraph Of bufferTimeout、transofrmationUID、transformationUserHash、source

getExecutionPlan

1. Running the program is to create a plan , Return to the executed data flow graph.

2. This method is before the plan is executed , Need to be called .

3. return JSON character string .

cast2StoppableSourceFunction

clean

Returns the of the given function “closure cleaned” edition . Only when the ExecutionConfig Only when closure cleanup is not disabled in

addOperator

Add an operator to the transform list in .List<StreamTransformation<?>> transformations

Not necessarily called by the user , To create an operator API Method must call this method .

registerCacheFile

Register a file in the distributed cache , This file will be customized by all users when running function As a local localPath visit .

The file can be a local file , It can also be a distributed file . If necessary , The runtime will temporarily copy files to the local cache .