一、得到流式环境

StreamExecutionEnvironment.getExecutionEnvironment()
相应的子api

- createLocalEnvironmentWithWebUI( new Configuration())
- createLocalEnvironment(...) 本地环境
- createRemoteEnvironmen(<host>,<port>,String... jarFiles)

并行序优先级：
自有平行度配置数据高于父亲并行度的设置setParallelism(2);

二、获取源 - Data Source API

关联数据到应用程序 : StreamExecutionEnvironment.addSource(sourceFunction)

sourceFunction

可以自定义编写单例源、并行源或富并行源。

SourceFunction、ParallelSourceFunction 、RichParallelSourceFunction。

自定义sourceFunction

//只支持一个并行度的
env.addSource(new SourceFunction<T>(){
       @Override
       public void run(SourceContext sourceContext) throws Exception {
        
        }

       @Override
       public void cancel() {
       }
});

内置源---pre-implemments

基于文件

env.readTextFile(file)

.readTextFile(rootPath+"example/wordcount.txt");

更多文件读取函数

 readFile(fileInputFormat, path) 
 readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo...)：

基于socket

env.socketTextStream(<host>,<port>,",")

基于集合

集合

fromCollection(Collection) - DataStream来自Java.util.Collection. 
fromCollection(Iterator, Class) - DataStream来自java.util.iterator. 
fromElements(T ...) -  元素-对像和须是相同的数据类型
generateSequence(from, to) -- DataStream来自给定的序列化相同对象.

自定义集合

addSource(...)

可并行的数据流

fromParallelCollection(SplittableIterator, Class) -  DataStream来自环境下的
fromParallelCollection(SplittableIterator<OUT> iterator,TypeInformation<OUT>) 
- SplittableIterator：数据源包括在一个切开的迭代器中取中
- TypeInformation：

三、Bundled Connectors实现同第三方工具的集成

Kafka(source/sink) - FlinkKafkaConsumer/FlinkKafkaProducer

ip:port、group.id、topic

Cassandra(sink)
Amazon Kinesis Stream(source/sink)
ElasticSearch(sink)
文件系统或HDFS(流与批两种方式，sink）
RabbmitMQ(source/sink)
Apapche NiFi(source/sink)
JDBC(sink)
Google PubSub(source/sink)
Twitter Streaming API(source）
...

四、数据转换 - DataStream Operators

数据转换： Transformation API

1、Map：DataStream -> DataStream

元素一一对应

DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {
    @Override
    public Integer map(String value) {
        return Integer.parseInt(value);
    }
});