logo
搜索
我的图书馆

发文章

发文工具

其他工具

留言交流
搜索
分享

Flink从入门到入土（详细教程）...

 任易x9cojw24ff 2021-05-27

和其他所有的计算框架一样，flink也有一些基础的开发步骤以及基础，核心的API，从开发步骤的角度来讲，主要分为四大部分

1.Environment

Flink Job在提交执行计算时，需要首先建立和Flink框架之间的联系，也就指的是当前的flink运行环境，只有获取了环境信息，才能将task调度到不同的taskManager执行。而这个环境对象的获取方式相对比较简单

  1. // 批处理环境
  2. val env = ExecutionEnvironment.getExecutionEnvironment
  3. // 流式数据处理环境
  4. val env = StreamExecutionEnvironment.getExecutionEnvironment

2.Source

Flink框架可以从不同的来源获取数据，将数据提交给框架进行处理, 我们将获取数据的来源称之为数据源.

2.1.从集合读取数据

一般情况下，可以将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用。这里的数据结构采用集合类型是比较普遍的

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList 
  5.  * date: 2020/8/28 19:02 
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：从集合读取数据
  11.  */
  12. object SourceList {
  14.   def main(args: Array[String]): Unit = {
  15.       //1.创建执行的环境
  16.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     //2.从集合中读取数据
  19.     val sensorDS: DataStream[WaterSensor] = env.fromCollection(
  20.       // List(1,2,3,4,5)
  21.       List(
  22.         WaterSensor("ws_001", 1577844001, 45.0),
  23.         WaterSensor("ws_002", 1577844015, 43.0),
  24.         WaterSensor("ws_003", 1577844020, 42.0)
  25.       )
  26.     )
  27.     //3.打印
  28.     sensorDS.print()
  29.     //4.执行
  30.     env.execute("sensor")
  32.   }
  34.   /**
  35.    * 定义样例类：水位传感器：用于接收空高数据
  36.    *
  37.    * @param id 传感器编号
  38.    * @param ts 时间戳
  39.    * @param vc 空高
  40.    */
  41.   case class WaterSensor(id: String, ts: Long, vc: Double)
  42. }

2.2从文件中读取数据

通常情况下，我们会从存储介质中获取数据，比较常见的就是将日志文件作为数据源

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList 
  5.  * date: 2020/8/28 19:02 
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：从文件读取数据
  11.  */
  12. object SourceFile {
  14.   def main(args: Array[String]): Unit = {
  15.     //1.创建执行的环境
  16.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  17.     //2.从指定路径获取数据
  18.     val fileDS: DataStream[String] = env.readTextFile("input/data.log")
  20.     //3.打印
  21.     fileDS.print()
  23.     //4.执行
  24.     env.execute("sensor")
  26.   }
  27. }
  28. /**
  29.  * 在读取文件时，文件路径可以是目录也可以是单一文件。如果采用相对文件路径，会从当前系统参数user.dir中获取路径
  30.  * System.getProperty("user.dir")
  31.  */
  32. /**
  33.  * 如果在IDEA中执行代码，那么系统参数user.dir自动指向项目根目录，
  34.  * 如果是standalone集群环境, 默认为集群节点根目录，当然除了相对路径以外，
  35.  * 也可以将路径设置为分布式文件系统路径，如HDFS
  36.  val fileDS: DataStream[String] =
  37.  env.readTextFile( "hdfs://hadoop02:9000/test/1.txt")
  38.  */

如果是standalone集群环境, 默认为集群节点根目录，当然除了相对路径以外，也可以将路径设置为分布式文件系统路径，如HDFS

  1. val fileDS: DataStream[String] =
  2. env.readTextFile( "hdfs://hadoop02:9000/test/1.txt")

默认读取时，flink的依赖关系中是不包含Hadoop依赖关系的，所以执行上面代码时，会出现错误。

解决方法就是增加相关依赖jar包就可以了

2.3 kafka读取数据

Kafka作为消息传输队列，是一个分布式的，高吞吐量，易于扩展地基于主题发布/订阅的消息系统。在现今企业级开发中，Kafka 和 Flink成为构建一个实时的数据处理系统的首选

2.3.1 引入kafka连接器的依赖

  1. <!-- https:///artifact/org.apache.flink/flink-connector-kafka-0.11 -->
  2. <dependency>
  3.     <groupId>org.apache.flink</groupId>
  4.     <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
  5.     <version>1.10.0</version>
  6. </dependency>

2.3.2 代码实现参考

  1. import java.util.Properties
  3. import org.apache.flink.streaming.api.scala._
  4. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
  5. import org.apache.flink.streaming.util.serialization.SimpleStringSchema
  7. /**
  8.  * description: SourceList 
  9.  * date: 2020/8/28 19:02 
  10.  * version: 1.0
  11.  *
  12.  * @author 阳斌
  13.  *         邮箱：1692207904@qq.com
  14.  *         类的说明：从kafka读取数据
  15.  */
  16. object SourceKafka {
  18.   def main(args: Array[String]): Unit = {
  19.     val env: StreamExecutionEnvironment =
  20.       StreamExecutionEnvironment.getExecutionEnvironment
  22.     val properties = new Properties()
  23.     properties.setProperty("bootstrap.servers", "hadoop02:9092")
  24.     properties.setProperty("group.id", "consumer-group")
  25.     properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  26.     properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  27.     properties.setProperty("auto.offset.reset", "latest")
  29.     val kafkaDS: DataStream[String] = env.addSource(
  30.       new FlinkKafkaConsumer011[String](
  31.         "sensor",
  32.         new SimpleStringSchema(),
  33.         properties)
  34.     )
  35.     kafkaDS.print()
  36.     env.execute("sensor")
  37.   }
  38. }

2.4 自定义数据源

大多数情况下，前面的数据源已经能够满足需要，但是难免会存在特殊情况的场合，所以flink也提供了能自定义数据源的方式

2.4.1  创建自定义数据源

  1. import com.atyang.day01.Source.SourceList.WaterSensor
  2. import org.apache.flink.streaming.api.functions.source.SourceFunction
  4. import scala.util.Random
  6. /**
  7.  * description: ss 
  8.  * date: 2020/8/28 20:36 
  9.  * version: 1.0
  10.  *
  11.  * @author 阳斌
  12.  *         邮箱：1692207904@qq.com
  13.  *         类的说明：自定义数据源
  14.  */
  15. class MySensorSource extends SourceFunction[WaterSensor] {
  16.   var flg = true
  17.   override def run(ctx: SourceFunction.SourceContext[WaterSensor]): Unit = {
  18.     while ( flg ) {
  19.       // 采集数据
  20.       ctx.collect(
  21.         WaterSensor(
  22.           "sensor_" +new Random().nextInt(3),
  23.           1577844001,
  24.           new Random().nextInt(5)+40
  25.         )
  26.       )
  27.       Thread.sleep(100)
  28.     }
  29.   }
  31.   override def cancel(): Unit = {
  32.     flg = false;
  33.   }
  34. }

3.Transform

在Spark中，算子分为转换算子和行动算子，转换算子的作用可以通过算子方法的调用将一个RDD转换另外一个RDD，Flink中也存在同样的操作，可以将一个数据流转换为其他的数据流。

转换过程中，数据流的类型也会发生变化，那么到底Flink支持什么样的数据类型呢，其实我们常用的数据类型，Flink都是支持的。比如：Long, String, Integer, Int, 元组，样例类，List, Map等。

3.1 map

  • 映射：将数据流中的数据进行转换, 形成新的数据流，消费一个元素并产出一个元素

  • 参数：Scala匿名函数或MapFunction

  • 返回：DataStream

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList 
  5.  * date: 2020/8/28 19:02 
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：从集合读取数据
  11.  */
  12. object Transfrom_map {
  14.   def main(args: Array[String]): Unit = {
  15.       //1.创建执行的环境
  16.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     //2.从集合中读取数据
  19.     val sensorDS: DataStream[WaterSensor] = env.fromCollection(
  20.       // List(1,2,3,4,5)
  21.       List(
  22.         WaterSensor("ws_001", 1577844001, 45.0),
  23.         WaterSensor("ws_002", 1577844015, 43.0),
  24.         WaterSensor("ws_003", 1577844020, 42.0)
  25.       )
  26.     )
  28.     val sensorDSMap = sensorDS.map(x => (x.id+"_1",x.ts+"_1",x.vc + 1))
  30.     //3.打印
  31.     sensorDSMap.print()
  32.     //4.执行
  33.     env.execute("sensor")
  35.   }
  37.   /**
  38.    * 定义样例类：水位传感器：用于接收空高数据
  39.    *
  40.    * @param id 传感器编号
  41.    * @param ts 时间戳
  42.    * @param vc 空高
  43.    */
  44.   case class WaterSensor(id: String, ts: Long, vc: Double)
  47. }

3.1.1 MapFunction

Flink为每一个算子的参数都至少提供了Scala匿名函数和函数类两种的方式，其中如果使用函数类作为参数的话，需要让自定义函数继承指定的父类或实现特定的接口。例如：MapFunction

sensor-data.log 文件数据

  1. sensor_1,1549044122,10
  2. sensor_1,1549044123,20
  3. sensor_1,1549044124,30
  4. sensor_2,1549044125,40
  5. sensor_1,1549044126,50
  6. sensor_2,1549044127,60
  7. sensor_1,1549044128,70
  8. sensor_3,1549044129,80
  9. sensor_3,1549044130,90
  10. sensor_3,1549044130,100
  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList 
  5.  * date: 2020/8/28 19:02 
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：从文件读取数据
  11.  */
  12. object SourceFileMap {
  14.   def main(args: Array[String]): Unit = {
  15.     //1.创建执行的环境
  16.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  17.     //2.从指定路径获取数据
  18.     val fileDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  20.     val MapDS = fileDS.map(
  21.       lines => {
  22.         //更加逗号切割 获取每个元素
  23.         val datas: Array[String] = lines.split(",")
  24.         WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  25.       }
  26.     )
  28.     //3.打印
  29.     MapDS.print()
  31.     //4.执行
  32.     env.execute("map")
  34.   }
  36.   /**
  37.    * 定义样例类：水位传感器：用于接收空高数据
  38.    *
  39.    * @param id 传感器编号
  40.    * @param ts 时间戳
  41.    * @param vc 空高
  42.    */
  43.   case class WaterSensor(id: String, ts: Long, vc: Double)
  45. }

  1. import org.apache.flink.api.common.functions.MapFunction
  2. import org.apache.flink.streaming.api.scala._
  4. /**
  5.  * description: SourceList 
  6.  * date: 2020/8/28 19:02 
  7.  * version: 1.0
  8.  *
  9.  * @author 阳斌
  10.  *         邮箱：1692207904@qq.com
  11.  *         类的说明：从文件读取数据
  12.  */
  13. object Transform_MapFunction {
  15.   def main(args: Array[String]): Unit = {
  16.     //1.创建执行的环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     //2.从指定路径获取数据
  19.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  21.      sensorDS.map()
  23.     //3.打印
  24.   //  MapDS.print()
  26.     //4.执行
  27.     env.execute("map")
  29.   }
  31.   /**
  32.    * 自定义继承 MapFunction
  33.    * MapFunction[T,O]
  34.    * 自定义输入和输出
  35.    *
  36.    */
  37.   class MyMapFunction extends MapFunction[String,WaterSensor]{
  38.     override def map(t: String): WaterSensor = {
  40.       val datas: Array[String] = t.split(",")
  42.       WaterSensor(datas(0),datas(1).toLong,datas(2).toInt)
  43.     }
  44.   }
  46.   /**
  47.    * 定义样例类：水位传感器：用于接收空高数据
  48.    *
  49.    * @param id 传感器编号
  50.    * @param ts 时间戳
  51.    * @param vc 空高
  52.    */
  53.   case class WaterSensor(id: String, ts: Long, vc: Double)
  55. }

3.1.2 RichMapFunction

所有Flink函数类都有其Rich版本。它与常规函数的不同在于，可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。也有意味着提供了更多的，更丰富的功能。例如：RichMapFunction

sensor-data.log 文件数据 同上一致

  1. import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
  2. import org.apache.flink.configuration.Configuration
  3. import org.apache.flink.streaming.api.scala._
  5. /**
  6.  * description: SourceList 
  7.  * date: 2020/8/28 19:02 
  8.  * version: 1.0
  9.  *
  10.  * @author 阳斌
  11.  *         邮箱：1692207904@qq.com
  12.  *         类的说明：从文件读取数据
  13.  */
  14. object Transform_RichMapFunction {
  16.   def main(args: Array[String]): Unit = {
  17.     //1.创建执行的环境
  18.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  19.     //2.从指定路径获取数据
  20.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  22.     val myMapDS: DataStream[WaterSensor] = sensorDS.map(new MyRichMapFunction)
  24.     //3.打印
  25.     myMapDS.print()
  27.     //4.执行
  28.     env.execute("map")
  30.   }
  32.   /**
  33.    * 自定义继承 MapFunction
  34.    * MapFunction[T,O]
  35.    * 自定义输入和输出
  36.    *
  37.    */
  38.   class MyRichMapFunction extends RichMapFunction[String,WaterSensor]{
  40.     override def map(value: String): WaterSensor = {
  41.       val datas: Array[String] = value.split(",")
  42.       //      WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  43.       WaterSensor(getRuntimeContext.getTaskName, datas(1).toLong, datas(2).toInt)
  44.     }
  46.     // 富函数提供了生命周期方法
  47.     override def open(parameters: Configuration): Unit = {}
  49.     override def close(): Unit = {}
  52.   }
  54.   /**
  55.    * 定义样例类：水位传感器：用于接收空高数据
  56.    *
  57.    * @param id 传感器编号
  58.    * @param ts 时间戳
  59.    * @param vc 空高
  60.    */
  61.   case class WaterSensor(id: String, ts: Long, vc: Double)
  63. }

Rich Function有一个生命周期的概念。典型的生命周期方法有：

  • open()方法是rich function的初始化方法，当一个算子例如map或者filter被调 用之前open()会被调用

  • close()方法是生命周期中的最后一个调用的方法，做一些清理工作

  • getRuntimeContext()方法提供了函数的RuntimeContext的一些信息，例如函数执行         的并行度，任务的名字，以及state状态

3.1.3 flatMap

  • 扁平映射：将数据流中的整体拆分成一个一个的个体使用，消费一个元素并产生零到多个元素

  • 参数：Scala匿名函数或FlatMapFunction

  • 返回：DataStream

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：FlatMap
  11.  */
  12. object Transform_FlatMap {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val listDS: DataStream[List[Int]] = env.fromCollection(
  22.       List(
  23.         List(1, 2, 3, 4),
  24.         List(5, 6, 7,1,1,1)
  25.       )
  26.     )
  28.     val resultDS: DataStream[Int] = listDS.flatMap(list => list)
  30.     resultDS.print()
  33.     // 4. 执行
  34.     env.execute()
  35.   }
  38. }

3.2. filter

  • 过滤：根据指定的规则将满足条件（true）的数据保留，不满足条件(false)的数据丢弃

  • 参数：Scala匿名函数或FilterFunction

  • 返回：DataStream

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：Filter
  11.  */
  12. object Transform_Filter {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val listDS: DataStream[List[Int]] = env.fromCollection(
  22.       List(
  23.         List(1, 2, 3, 4,1, 2, 3, 4),
  24.         List(5, 6, 7,1,1,1,1, 2, 3, 4,1, 2, 3, 4),
  25.         List(1, 2, 3, 4),
  26.         List(5, 6, 7,1,1,1),
  27.         List(1, 2, 3, 4),
  28.         List(5, 6, 7,1,1,1)
  29.       )
  30.     )
  31.     // true就留下，false就抛弃
  32.     listDS.filter(num => {
  33.       num.size>5
  34.       })
  35.       .print("filter")
  36.     // 4. 执行
  37.     env.execute()
  38.   }
  39. }

3.3 keyBy

在Spark中有一个GroupBy的算子，用于根据指定的规则将数据进行分组，在flink中也有类似的功能，那就是keyBy，根据指定的key对数据进行分流

  • 分流：根据指定的Key将元素发送到不同的分区，相同的Key会被分到一个分区（这里分区指的就是下游算子多个并行节点的其中一个）。keyBy()是通过哈希来分区的

  • 参数：Scala匿名函数或POJO属性或元组索引，不能使用数组

  • 返回：KeyedStream

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：FlatMap
  11.  */
  12. object Transform_KeyBy {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  23.     //3.转换为样例类
  24.     val mapDS = sensorDS.map(
  25.       lines => {
  26.         val datas = lines.split(",")
  27.         WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  28.       }
  29.     )
  31.     // 4. 使用keyby进行分组
  32.     // TODO 关于返回的key的类型：
  33.     // 1. 如果是位置索引 或 字段名称 ，程序无法推断出key的类型，所以给一个java的Tuple类型
  34.     // 2. 如果是匿名函数 或 函数类 的方式，可以推断出key的类型，比较推荐使用
  35.     // *** 分组的概念：分组只是逻辑上进行分组,打上了记号(标签)，跟并行度没有绝对的关系
  36.     //      同一个分组的数据在一起（不离不弃）
  37.     //      同一个分区里可以有多个不同的组
  39.     //        val sensorKS: KeyedStream[WaterSensor, Tuple] = mapDS.keyBy(0)
  40.     //    val sensorKS: KeyedStream[WaterSensor, Tuple] = mapDS.keyBy("id")
  41.     val sensorKS: KeyedStream[WaterSensor, String] = mapDS.keyBy(_.id)
  42.     //    val sensorKS: KeyedStream[WaterSensor, String] = mapDS.keyBy(
  43.     //      new KeySelector[WaterSensor, String] {
  44.     //        override def getKey(value: WaterSensor): String = {
  45.     //          value.id
  46.     //        }
  47.     //      }
  48.     //    )
  50.     sensorKS.print().setParallelism(5)
  52.     // 4. 执行
  53.     env.execute()
  54.   }
  56.   /**
  57.    * 定义样例类：水位传感器：用于接收空高数据
  58.    *
  59.    * @param id 传感器编号
  60.    * @param ts 时间戳
  61.    * @param vc 空高
  62.    */
  63.   case class WaterSensor(id: String, ts: Long, vc: Double)
  64. }

3.4 shuffle

  • 打乱重组（洗牌）：将数据按照均匀分布打散到下游

  • 参数：无

  • 返回：DataStream

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：FlatMap
  11.  */
  12. object Transform_Shuffle {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  23.     val shuffleDS = sensorDS.shuffle
  25.     sensorDS.print("data")
  27.     shuffleDS.print("shuffle")
  28.     // 4. 执行
  29.     env.execute()
  30.   }
  31. }

3.5. split

在某些情况下，我们需要将数据流根据某些特征拆分成两个或者多个数据流，给不同数据流增加标记以便于从流中取出。

需求：将水位传感器数据按照空高高低（以40cm,30cm为界），拆分成三个流

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：FlatMap
  11.  */
  12. object Transform_Split {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  23.     // 3.转换成样例类
  24.     val mapDS: DataStream[WaterSensor] = sensorDS.map(
  25.       lines => {
  26.         val datas: Array[String] = lines.split(",")
  27.         WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  28.       }
  29.     )
  30.     val splitSS: SplitStream[WaterSensor] = mapDS.split(
  31.       sensor => {
  32.         if (sensor.vc < 40) {
  33.           Seq("normal")
  34.         } else if (sensor.vc < 80) {
  35.           Seq("Warn")
  36.         } else {
  37.           Seq("alarm")
  38.         }
  39.       }
  40.     )
  42.     // 4. 执行
  43.     env.execute()
  44.   }
  46.   /**
  47.    * 定义样例类：水位传感器：用于接收空高数据
  48.    *
  49.    * @param id 传感器编号
  50.    * @param ts 时间戳
  51.    * @param vc 空高
  52.    */
  53.   case class WaterSensor(id: String, ts: Long, vc: Double)
  54. }

3.6 select

将数据流进行切分后，如何从流中将不同的标记取出呢，这时就需要使用select算子了。

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：FlatMap
  11.  */
  12. object Transform_Split {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  23.     // 3.转换成样例类
  24.     val mapDS: DataStream[WaterSensor] = sensorDS.map(
  25.       lines => {
  26.         val datas: Array[String] = lines.split(",")
  27.         WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  28.       }
  29.     )
  30.     val splitDS: SplitStream[WaterSensor] = mapDS.split(
  31.       sensor => {
  32.         if (sensor.vc < 40) {
  33.           Seq("info")
  34.         } else if (sensor.vc < 80) {
  35.           Seq("warn")
  36.         } else {
  37.           Seq("error")
  38.         }
  39.       }
  40.     )
  41.     val errorDS: DataStream[WaterSensor] = splitDS.select("error")
  42.     val warnDS: DataStream[WaterSensor] = splitDS.select("warn")
  43.     val infoDS: DataStream[WaterSensor] = splitDS.select("info")
  45.     infoDS.print("info")
  46.     warnDS.print("warn")
  47.     errorDS.print("error")
  49.     // 4. 执行
  50.     env.execute()
  51.   }
  53.   /**
  54.    * 定义样例类：水位传感器：用于接收空高数据
  55.    *
  56.    * @param id 传感器编号
  57.    * @param ts 时间戳
  58.    * @param vc 空高
  59.    */
  60.   case class WaterSensor(id: String, ts: Long, vc: Double)
  61. }

3.7 connect

在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息，这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。

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

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：FlatMap
  11.  */
  12. object Transform_Connect {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  23.     // 3.转换成样例类
  24.     val mapDS: DataStream[WaterSensor] = sensorDS.map(
  25.       lines => {
  26.         val datas: Array[String] = lines.split(",")
  27.         WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  28.       }
  29.     )
  31.     // 4. 从集合中再读取一条流
  32.     val numDS: DataStream[Int] = env.fromCollection(List(1, 2, 3, 4, 5, 6))
  34.     val resultCS: ConnectedStreams[WaterSensor, Int] = mapDS.connect(numDS)
  36.     // coMap表示连接流调用的map，各自都需要一个 function
  37.     resultCS.map(
  38.       sensor=>sensor.id,
  39.       num=>num+1
  40.     ).print()
  42.     // 4. 执行
  43.     env.execute()
  44.   }
  46.   /**
  47.    * 定义样例类：水位传感器：用于接收空高数据
  48.    *
  49.    * @param id 传感器编号
  50.    * @param ts 时间戳
  51.    * @param vc 空高
  52.    */
  53.   case class WaterSensor(id: String, ts: Long, vc: Double)
  54. }

3.8 union

对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream

connect与 union 区别：

  1. union之前两个流的类型必须是一样，connect可以不一样

  2. connect只能操作两个流，union可以操作多个。

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：FlatMap
  11.  */
  12. object Transform_Union {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2. 从集合中读取流
  21.     val num1DS: DataStream[Int] = env.fromCollection(List(1, 2, 3, 4))
  22.     val num2DS: DataStream[Int] = env.fromCollection(List(7, 8, 9, 10))
  23.     val num3DS: DataStream[Int] = env.fromCollection(List(17, 18, 19, 110))
  25.     // TODO union 真正将多条流合并成一条流
  26.     // 合并的流，类型必须一致
  27.     // 可以合并多条流，只要类型一致
  28.     num1DS.union(num2DS).union(num3DS)
  29.       .print()
  32.     // 4. 执行
  33.     env.execute()
  34.   }
  36.   /**
  37.    * 定义样例类：水位传感器：用于接收空高数据
  38.    *
  39.    * @param id 传感器编号
  40.    * @param ts 时间戳
  41.    * @param vc 空高
  42.    */
  43.   case class WaterSensor(id: String, ts: Long, vc: Double)
  44. }

3.9 Operator

Flink作为计算框架，主要应用于数据计算处理上， 所以在keyBy对数据进行分流后，可以对数据进行相应的统计分析

3.9.1 滚动聚合算子（Rolling Aggregation）

这些算子可以针对KeyedStream的每一个支流做聚合。执行完成后，会将聚合的结果合成一个流返回，所以结果都是DataStream

sum()

min()

max()

3.9.2 reduce

一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。

  1. import org.apache.flink.streaming.api.scala._
  3. /**
  4.  * description: SourceList
  5.  * date: 2020/8/28 19:02
  6.  * version: 1.0
  7.  *
  8.  * @author 阳斌
  9.  *         邮箱：1692207904@qq.com
  10.  *         类的说明：Reduce
  11.  */
  12. object Transform_Reduce {
  14.   def main(args: Array[String]): Unit = {
  16.     // 1.创建执行环境
  17.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.     env.setParallelism(1)
  20.     // 2.读取数据
  21.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  23.     // 3.转换成样例类
  24.     val mapDS: DataStream[WaterSensor] = sensorDS.map(
  25.       lines => {
  26.         val datas: Array[String] = lines.split(",")
  27.         WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  28.       }
  29.     )
  30.     val sensorKS: KeyedStream[WaterSensor, String] = mapDS.keyBy(_.id)
  31.     // 输入的类型一样，输出类型和输出类型也要一样
  32.     // 组内的第一条数据，不进入reduce计算
  33.     val reduceDS: DataStream[WaterSensor] = sensorKS.reduce(
  34.       (ws1, ws2) => {
  35.         println(ws1 + "<===>" + ws2)
  36.         WaterSensor(ws1.id, System.currentTimeMillis(), ws1.vc + ws2.vc)
  37.       }
  38.     )
  39.     reduceDS.print("reduce")
  40.     // 4. 执行
  41.     env.execute()
  42.   }
  44.   /**
  45.    * 定义样例类：水位传感器：用于接收空高数据
  46.    *
  47.    * @param id 传感器编号
  48.    * @param ts 时间戳
  49.    * @param vc 空高
  50.    */
  51.   case class WaterSensor(id: String, ts: Long, vc: Double)
  52. }

3.9.3process

Flink在数据流通过keyBy进行分流处理后，如果想要处理过程中获取环境相关信息，可以采用process算子自定义实现 1)继承KeyedProcessFunction抽象类，并定义泛型：[KEY, IN, OUT]

class MyKeyedProcessFunction extends KeyedProcessFunction[String, WaterSensor, String]{}

  1. 重写方法

  1. // 自定义KeyedProcessFunction,是一个特殊的富函数
  2.   // 1.实现KeyedProcessFunction，指定泛型：K - key的类型， I - 上游数据的类型， O - 输出的数据类型
  3.   // 2.重写 processElement方法，定义 每条数据来的时候 的 处理逻辑
  5. /**
  6.       * 处理逻辑：来一条处理一条
  7.       *
  8.       * @param value 一条数据
  9.       * @param ctx   上下文对象
  10.       * @param out   采集器：收集数据，并输出
  11.       */
  12.     override def processElement(value: WaterSensor, ctx: KeyedProcessFunction[String, WaterSensor, String]#Context, out: Collector[String]): Unit = {
  13.       out.collect("我来到process啦，分组的key是="+ctx.getCurrentKey+",数据=" + value)
  14.       // 如果key是tuple，即keyby的时候，使用的是 位置索引 或 字段名称，那么key获取到是一个tuple
  15. //      ctx.getCurrentKey.asInstanceOf[Tuple1].f0 //Tuple1需要手动引入Java的Tuple
  16.     }

完整代码：

  1. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  2. import org.apache.flink.streaming.api.scala._
  3. import org.apache.flink.util.Collector
  5. /**
  6.  * description: SourceList
  7.  * date: 2020/8/28 19:02
  8.  * version: 1.0
  9.  *
  10.  * @author 阳斌
  11.  *         邮箱：1692207904@qq.com
  12.  *         类的说明：Reduce
  13.  */
  14. object Transform_Process {
  16.   def main(args: Array[String]): Unit = {
  18.     // 1.创建执行环境
  19.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  20.     env.setParallelism(1)
  22.     // 2.读取数据
  23.     val sensorDS: DataStream[String] = env.readTextFile("input/sensor-data.log")
  25.     // 3.转换成样例类
  26.     val mapDS: DataStream[WaterSensor] = sensorDS.map(
  27.       lines => {
  28.         val datas: Array[String] = lines.split(",")
  29.         WaterSensor(datas(0), datas(1).toLong, datas(2).toInt)
  30.       }
  31.     )
  32.     //按照ID  进行分组
  33.     val sensorKS: KeyedStream[WaterSensor, String] = mapDS.keyBy(_.id)
  35.     sensorKS.process(new MyKeyedProcessFunction)
  37.     // 4. 执行
  38.     env.execute()
  39.   }
  41.   // 自定义KeyedProcessFunction,是一个特殊的富函数
  42.   // 1.实现KeyedProcessFunction，指定泛型：K - key的类型， I - 上游数据的类型， O - 输出的数据类型
  43.   // 2.重写 processElement方法，定义 每条数据来的时候 的 处理逻辑
  44.   class MyKeyedProcessFunction extends KeyedProcessFunction[String, WaterSensor, String] {
  45.     /**
  46.      * 处理逻辑：来一条处理一条
  47.      *
  48.      * @param value 一条数据
  49.      * @param ctx   上下文对象
  50.      * @param out   采集器：收集数据，并输出
  51.      */
  52.     override def processElement(value: WaterSensor, ctx: KeyedProcessFunction[String, WaterSensor, String]#Context, out: Collector[String]): Unit = {
  53.       out.collect("我来到process啦，分组的key是="+ctx.getCurrentKey+",数据=" + value)
  54.       // 如果key是tuple，即keyby的时候，使用的是 位置索引 或 字段名称，那么key获取到是一个tuple
  55.       //      ctx.getCurrentKey.asInstanceOf[Tuple1].f0 //Tuple1需要手动引入Java的Tuple
  56.     }
  57.   }
  59.   /**
  60.    * 定义样例类：水位传感器：用于接收空高数据
  61.    *
  62.    * @param id 传感器编号
  63.    * @param ts 时间戳
  64.    * @param vc 空高
  65.    */
  66.   case class WaterSensor(id: String, ts: Long, vc: Double)
  67. }

4.Sink

Sink有下沉的意思，在Flink中所谓的Sink其实可以表示为将数据存储起来的意思，也可以将范围扩大，表示将处理完的数据发送到指定的存储系统的输出操作

之前我们一直在使用的print方法其实就是一种Sink。

  1.   @PublicEvolving
  2.     public DataStreamSink<T> print(String sinkIdentifier) {
  3.         PrintSinkFunction<T> printFunction = new PrintSinkFunction(sinkIdentifier, false);
  4.         return this.addSink(printFunction).name("Print to Std. Out");
  5.     }

官方提供了一部分的框架的sink。除此以外，需要用户自定义实现sink

  1. 精彩推荐
  2. SpringBoot内容聚合IntelliJ IDEA内容聚合Mybatis内容聚合
  4. 欢迎长按下图关注公众号后端技术精选

    本站是提供个人知识管理的网络存储空间，所有内容均由用户发布，不代表本站观点。请注意甄别内容中的联系方式、诱导购买等信息，谨防诈骗。如发现有害或侵权内容，请点击一键举报。
    转藏 分享 献花（0）

    来自： 任易x9cojw24ff > 《Flink》

    举报/认领

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多