大数据之Flink（下）

• 前言
• 一、split 和 select 算子
• 二、Connect和 CoMap 算子
• 三、union 算子
• 四、WordCount案例
• • 4.1 离线数据
  • 4.2 在线数据
• 总结

在 大数据之Flink（上）中我们介绍了Flink的特点、架构、两段式提交以及数据流等相关知识，本文介绍Flink独有的算子以及用Flink实现WordCount案例

split算子根据某些特征把一个DataStream拆分成两个或者多个DataStream。

Select算子 从一个SplitStream中获取一个或者多个DataStream。

代码如下：

//根据标签进行切分
val splitStream:SplitStream[Startuplog] = startuplogDstream.split{
      startuplog =>
        var flag:List[String] = null;
        if(startuplog.ch == "appstore"){
          flag = List("apple","usa")
        }else if(startuplog.ch == "huawei"){
          flag = List("android","china")
        }else{
          flag = List("android","other")
        }
        flag
}//根据数据流中的标签列将数据划分为多个流
val appleStream:DataStream[Startuplog] = startuplogDstream.select("apple","china")
val otherStream:DataStream[Startuplog] = startuplogDstream.select("other")
//根据需要，获得切分后的流进行后续的处理

Connect算子连接两个保持他们类型的数据流，两个数据流被Connect之后，只是被放在了同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

CoMap,CoFlatMap算子作用于ConnectedStreams上，功能与map和flatMap一样，对ConnectedStreams中的每一个Stream分别进行map和flatMap处理。

注意：map/flatMap要指定如何将多个数据流合并到一起，即对不同的流指定不同的函数并且函数的返回类型必须一样且和最终的返回类型一致，普通的map/flatMap则是可以直接使用因为只是针对同一个数据流；
代码如下：

val conStream:ConnectedStreams[Startuplog,Startuplog] = appleStream.connect(otherStream)
val allStream:DataStream[String] = conStream.map(
  //每个流都要指定函数，并且函数返回类型必须和需要的类型一致（这里是String）
  (startuplog1:Startuplog) => startuplog1.ch
  (startuplog2:Startuplog) => startuplog2.ch
)

  对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream。注意:如果你将一个DataStream跟它自己做union操作，在新的DataStream中，你将看到每一个元素都出现两次。

代码如下：

val unionStream:DataStream[Startuplog] = appleStream.union(otherStream)

  Union在合并数据流的时候，不要求数据流的类型一致就可以直接合并，而connect则需要先把数据流放入到一个大的流中进行数据类型的转换后才能进行合并。与此同时，**connect一次只能合并两个数据流，而union可以合并多个。

代码如下：

// 创建一个env    创建环境变量
    val env = ExecutionEnvironment.getExecutionEnvironment

    val textDataSet: DataSet[String] = env.readTextFile("D:\data\1.txt")

    val aggset: AggregateDataSet[(String, Int)] = textDataSet.flatMap(_.split(" ")).map((_,1)).groupBy(0).sum(1)

    aggset.print()

代码如下：

// 创建一个env    创建环境变量
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val dataStream: DataStream[String] = env.socketTextStream("hadoop1",7777)

    val aggStream: DataStream[(String, Int)] = dataStream.flatMap(_.split(" ")).map((_,1)).keyBy(0).sum(1)

    aggStream.print()

    env.execute() //在线数据就要加上执行

在 大数据之Flink（上）中我们介绍了Flink的特点、架构、两段式提交以及数据流等相关知识，本文介绍了Flink独有的算子以及用Flink实现WordCount案例。如果有需要补充的或者不足的地方希望大家指出，我们共同进步。