Flink 常用Api 学习及测试

Flink DataStream

Flink data source

       Flink 做为一款流式计算框架，它可用来做批处理，即处理静态的数据集、历史的数据集；也可以用来做流处理，即实时的处理些实时数据流，实时的产生数据流结果，只要数据源源不断的过来，Flink 就能够一直计算下去，这个 Data Sources 就是数据的来源地。其中可以通过

获取数据源。StreamExecutionEnvironment 中可以使用以下几个已实现的 stream sources，

基于集合

fromCollection(Collection) - 从 Java 的 Java.util.Collection 创建数据流。集合中的所有元素类型必须相同。 fromCollection(Iterator, Class) - 从一个迭代器中创建数据流。Class 指定了该迭代器返回元素的类型。 fromElements(T …) - 从给定的对象序列中创建数据流。所有对象类型必须相同。 fromParallelCollection(SplittableIterator, Class) - 从一个迭代器中创建并行数据流。Class 指定了该迭代器返回元素的类型。 generateSequence(from, to) - 创建一个生成指定区间范围内的数字序列的并行数据流。 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Event> input = env.fromElements(                new Event(1, "barfoo", 1.0),                new Event(2, "start", 2.0),                new Event(3, "foobar", 3.0),                ... );

####基于文件

readTextFile(path) - 读取文本文件，即符合 TextInputFormat 规范的文件，并将其作为字符串返回。

readFile(fileInputFormat, path) - 根据指定的文件输入格式读取文件（一次）。

readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) - 这是上面两个方法内部调用的方法。它根据给定的 fileInputFormat 和读取路径读取文件。根据提供的 watchType，这个 source 可以定期（每隔 interval 毫秒）监测给定路径的新数据（FileProcessingMode.PROCESS_CONTINUOUSLY），或者处理一次路径对应文件的数据并退出（FileProcessingMode.PROCESS_ONCE）。你可以通过 pathFilter 进一步排除掉需要处理的文件。

###各自特点 1、基于集合：有界数据集，更偏向于本地测试用 2、基于文件：适合监听文件修改并读取其内容 3、基于 Socket：监听主机的 host port，从 Socket 中获取数据 4、自定义 addSource：大多数的场景数据都是无界的，会源源不断的过来。比如去消费 Kafka 某个 topic 上的数据，这时候就需要用到这个 addSource，可能因为用的比较多的原因吧，Flink 直接提供了 FlinkKafkaConsumer011 等类可供你直接使用。你可以去看看 FlinkKafkaConsumerBase 这个基础类，它是 Flink Kafka 消费的最根本的类。

自定义source 自定义source * 1.实现并行度为1的自定义source * ---实现sourcefunction * 一般不需要实现容错机制 * 处理好cancel方法（cancel应用时，这个方法被调用） * 注意：SourceFunction 和SourceContext需要泛型 * 2.实现并行化的自定义source * ---实现parallelSourceFunction * 或者继承RichParallelSourceFunction

数据源的并行度不可设置默认为1，累加和结果 可以设置并行度 继承ParallerlSourceFunction

Transformations 算子

转型	描述
Map DataStream->DataStream	采用一个元素生成一个元素：中级可以做一些清洗或转换 DataStream<Integer> datastream=//…. datastream.map(new MapFunction<Integer,Integer>(){ @Override     public Integer map(Integer value) throws Exception {         return 2 * value;     } });
FlatMap DataStream→DataStream	输入一个元素并生成零个，一个或多个元素。将句子分割为单词的flatmap函数： dataStream.flatMap(new FlatMapFunction<String, String>() {     @Override     public void flatMap(String value, Collector<String> out)         throws Exception {         for(String word: value.split(" ")){             out.collect(word);         }     } });
Fliter 过滤	过滤函数，对输入数据进行判断符合条件的留下 dataStream.filter(new FilterFunction<Integer>() {     @Override     public boolean filter(Integer value) throws Exception {         return value != 0;     } });
KeyBy DataStream→KeyedStream
Reduce KeyedStream→DataStream	对数据进行聚合操作，结合当前元素和上一次reduce返回的值进行聚合操作，然后返回一个新值
Union	合并多个流，新的流必须包含所有流中的数据，但是union有一个限制就是所有合并的流类型必须一致 DataStreamSource<Integer> source1=env.addSource(new MySource()); DataStreamSource<Integer> source2=env.addSource(new MySource()); DataStream<Integer> sou=source1.union(source2).union(source1)...;

Connect	和union类型，但是只能连接两个流，两个流数据类型可以不同，会对两个流的数据应用不同的方法 Comap  在connectStream中需要使用这种函数，类似于map和Flatmap
Split	根据规则把一个流切分成多个流 Select和split配合使用功能，选择切分后的流

DataStreamApi 之Partition

RandomPartitiong 随机分区 –dataStream.shffle() Rebalancing 对数据集进行再平衡，充分去，消除数据倾斜—dataStream.rebalance() Rescaling举例如果上游操作有两个并发，而下游有四个并发，数据处理时上游的一个并发结果分配给下有的俩个并发操作，上游的另一个并发分配给下游的其他并发操作，即上下游资源会进行平分 自定义的分区规则 需要实现partition接口

DataStream Sink

weiteAsText（）将元素以字符串形式逐行写入，这些字符串通过调用每个元素的toString()方法实现 pringt（）/printToErr() 打印到并准输出或者错误输出流中 自定义输出addSInk[kafka redis] Redis安装在master节点上 查看是否安装gcc –v ，安装命令  yum install gcc –c++ 安装配置：https://blog.csdn.net/qq_42815754/article/details/82832335 注意：需要额外更改 redis.conf配置bind 必须为0.0.0.0否则外部无法访问，dir 必须确定存在 在slave2节点安装netcat，启动端口9000，再起一个窗口看成功与否 启动redis-server ./redis.config, redis-cli,运行java代码，在slave2 9000端口下输入字符串，结果如下 注意：在windows下运行代码可能会出现无法连接redis，此时 ps –ef|grep redis,发现只允许本机访问，则修改 bind 127.0.0.1 –> bind 0.0.0.0,关闭redis进程，重新启动即可

Flink DataSet

DataSource

###基于文件  readFile(pth) 基于集合  FromCollection(Colection)   和DataStream类似

DataSet Api Transformations

Map 输入一个元素然后返回一个元素，中间可以做一些清洗转换等操作 FlatMap 输入一个元素，可以返回0个一个或多个元素 MapPartition 类似于map ,一次处理一个分区的数据，如果在进行map处理的时候需要获取第三方资源链接，建议使用MapPartition Fliter 过滤函数，对传入的数进行判断，符合条件的会被留下 Reduce 对数据进行聚合操作，结婚当前元素和上一次reduce返回的值进行聚合处理，然后返回一个新的值 Aggregate sum max min Distinct  返回一个数据集中去重之后的元素 Join 内连接  Outerjoin 外连接 Cross 获取两个数据集的笛卡尔积 Union 返回两个数据集的总和数据类型要一致 First-n 获取集合中的前n个元素 Sort Partition 在本地对数据集的所有分区进行排序，通过sortPartition的连接调用完成对多个字段的排序。