Flink原理实战每日一篇02

分手后的思念是犯贱 2022-01-11 07:19 214阅读 0赞

今天的标题是 DataStream API学习回顾，查漏补全..........

1，主要是基本API都已经很熟悉，没必要的地方我就跳过去了，简单的聊一聊。

1）DataSource ，我们只要知道几种：

读文件   env.readTextFile("user/aa.log")

Socket 数据源，我们本地测试用用   env.socketTextStream("localhpst",9999)

手动创建：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

外部数据源 常用的读取kafka的数据 :直接上代码，不过先导入依赖哦

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

2，DataStrem的转化操作，说白了有哪些算子，现在写都是简单的：

1)map算子  mapFunction   RichMapFunction

多一句嘴 每个算子都有对应的RichFunction

看代码，就多了个open初始化方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

2） 这里其余算子就不讲了 ，官网有相应的解释说明，下面要说的是**keyBy算子**，因为后续的Flink学习里面会经常的遇到它，逃不掉的，所以要好好学习。

keyBy 算子会将DataStream转成KeyedStream\[T\] ，也就是数据集中执行partition操作（思考一下，会不会出现数据倾斜？），将相同的key值的数据放置在相同的分区中（思考一下，怎么统计所有的key？spark有此类算子，Flink没有），如下图，相同的key的数据会在一起。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

3）Reduce 算子，对相同的key的值累加

val data8 =env.fromElements((1,2),(1,3))
    data8.keyBy(0).reduce{(t1,t2)=>(t1._1,t1._2+t2._2)}

4) 聚合类算子 说白了就是 sum min minBy max maxBy 等等

data8.keyBy(0).sum(1）

5）流跟流的操作，我是用的少，不过必须得了解，要不然怎么显得基础扎实呢 是把。

Union  合并流

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

val data8 =env.fromElements((1,2),(1,3))
      val data9 =env.fromElements((1,2),(1,3))
    val data10 = data8.union(data9)

6）还有其他的操作， Connet  CoMao CoflatMap\[DataStream -> DataStream \]

7) 下面的算子很有意思了

split 算子  按条件拆分，不是真正的拆分哦 ，只是做了**拆分标记**，这点要懂

select 算子 选择前面标记好的信息，生成新的流，有没有像 side output 流？

具体代码：

val rs: SplitStream[(Int, Int)] = data8.split(x=>if (x._2 %2==0) Seq("even") else Seq ("odd"))
    val rs2: DataStream[(Int, Int)] = rs.select("odd")
    val rs3: DataStream[(Int, Int)] = rs.select("odd","even")

iterate 算子  适合迭代计算场景，通过每一次的迭代计算，将结果反馈到下一次计算结果，具体还是看业务场景，代码如下：

val iterateData = env.fromElements(1,2,3,45).map{t => t}
    val aa = iterateData.iterate((input : ConnectedStreams[Int,String])=>{
      //todo 定义两个map方法  ，
      //todo 第一个流反馈给下一次迭代，
      //todo 指标加1 不等于2 则直接输出到下游DataStream，
      //todo 第二个map(s=>s)是将数据输出到下游数据集
      val heard= input.map(i=>(i+1).toString,s=>s)
      (heard.filter(_ =="2"),heard.filter(_ !="2"))
    },1000)  // 1000是指定最长迭代时间 单位为ms，超过该时间没有数据接入 则停止

总结，数据加1 等于"2" 就继续迭代，反之则输出到下一个流。

今天就到这里把。。。。。。感觉迭代流  还有 split  select 算子之前都没怎么学习留意，今天还是收获很多的

下一次主要讲分区 广播变量 还是sink输出等等。。。。下次见。。。。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20220111/f3a550678298429db51702c08fdfe53e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20220111/d5283359756241468a7c1579eed2a990.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20220111/579823c21c8d4ec88307c9bdecdce36d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20220111/591ef6718ec647dd957a6b7b2c518982.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20220111/8031f4383852469cb41836539dbfd939.png