Flink学习笔记

对用户行为或者相关事件进行实时监测和分析，基于风控规则进行预警

类似于自定义Java Bean的方式

Option、Either、List、Map

map方法的声明：def map[R: TypeInformation](fun: T => R): DataStream[R]

map的方法声明上定义了隐式形参（上下文界定），需要存在一个隐式值，将R => Typeinformation[R]

//导入隐式转换函数createTypeInformation即可
import org.apache.flink.streaming.api.scala._

createTypeInformation隐式转换函数定义在org.apache.flink.streaming.api.scala包对象中

例如flatMap类型，泛型可以在Collector[O]上设置

代码示例

在调用DataStream API时通过形参手动传入

调用returns方法返回

Types和TypeHint之间的区别

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

/**
 * map方法, 每来一条数据就会调用一次, 将一种数据类型转换成另外一种数据类型
 * 一般用作数据类型的转换, 常见如从Kafka读取String的数据, 然后使用map转换成样例类
 */
object MapFunctionDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.fromCollection(Seq(1, 2, 3))
      //可以传入匿名函数
      .map(data => data * 2)
      //或者自定义类实现MapFunction接口, 重写map方法
      .map(new MyMapFunction2)
      .print("map test>>>").setParallelism(1)

    env.execute()
  }

}

/**
 *
 * @tparam I 输入的数据类型
 * @tparam O 输出的数据类型
 */
class MyMapFunction2 extends MapFunction[Int, Int] {
  override def map(value: Int) = value * 2
}

import org.apache.flink.streaming.api.scala._

/**
 * 对于每一个流内元素进行判断，若为 true 则元素正常输出，若为 false 则元素被过滤掉。一般用于数据的过滤
 */
object FilterFunctionDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.fromElements(1, 2, 3)
      //使用匿名函数
      .filter(data => data.==(1))
      .print("filter >").setParallelism(1)

    env.execute()
  }

}

import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * flatMap, 传入一个数据, 输出0到多个数据, 可以实现filter + map的操作
 * 
 */
object FlatMapFunctionDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    
    env.fromElements(1, 2, 3)
      //对于匿名函数, 如果有泛型的, 需要在形参上显示指定
      .flatMap((data, out: Collector[String]) => {
        if (data.==(1)) {
          out.collect(data.toString)
        }
      })
      .print("flatMap>")
    
    env.execute()
    
  }

}

在输入流上，对指定的字段做叠加求和的操作

在输入流上，对指定的字段求最小值

在输入流上，对指定的字段求最大值

与 min()类似，在输入流上针对指定字段求最小值。

不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值

而 minBy()则会返回包含字段最小值的整条数据

和minBy类似

import org.apache.flink.streaming.api.scala._

/**
 * keyBy之后使用简单聚合算子
 * 例如sum、max、min等
 */
object TransTupleAggregationDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.fromElements(("a", 1), ("a", 3), ("b", 3), ("b", 4))

    stream.keyBy(_._1).sum(1).print("sum position>")
    stream.keyBy(_._1).sum("_2").print("sum field>")
    stream.keyBy(_._1).max(1).print("max position>")
    stream.keyBy(_._1).max("_2").print("max field>")
    stream.keyBy(_._1).min(1).print("min position>")
    stream.keyBy(_._1).min("_2").print("min field>")
    stream.keyBy(_._1).maxBy(1).print("maxBy position>")
    stream.keyBy(_._1).maxBy("_2").print("maxBy field>")
    stream.keyBy(_._1).minBy(1).print("minBy position>")
    stream.keyBy(_._1).minBy("_2").print("minBy field>")

    env.execute()
  }

}

import com.yanggu.flink.lowlevelapi.pojo.Event
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.util.concurrent.TimeUnit
import scala.util.Random

object ReduceFunctionDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.addSource(new ClickSource)

    stream
      .map(data => (data.name, 1L))
      //DataStream => KeyedStream
      .keyBy(_._1)
      //KeyedStream => DataStream
      //实现求和的功能（模仿sum）
      .reduce((v1, v2) => (v1._1, v1._2 + v2._2))
      //这里返回true, 意味着所有的数据的key都相同, 被发到同一个slot上执行
      .keyBy(_ => true)
      //实现最大值的功能（模仿maxBy）
      .reduce((v1, v2) => (v1._1, v1._2.max(v2._2)))
      .print()

    env.execute()
  }

}

class ClickSource extends SourceFunction[Event] {

  private var runFlag = true

  override def run(ctx: SourceFunction.SourceContext[Event]) = {
    val users = Array("Mary", "Alice", "Bob", "Cary")
    val urls = Array("./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2")
    val usersLength = users.length
    val urlLength = urls.length
    val random = new Random()
    while (runFlag) {
      ctx.collect(Event(users(random.nextInt(usersLength)), urls(random.nextInt(urlLength)), System.currentTimeMillis()))
      TimeUnit.MILLISECONDS.sleep(100L)
    }
  }

  override def cancel() = {
    runFlag = false
  }
  
}

Scala

Java

“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其Rich 版本。

富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction 等

富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能

复函数提供两个生命周期方法open()方法和close()方法

在对应的map、filter等方法业务逻辑方法前被调用

在并行子任务开始后只会被调用一次，这时候上下文信息已经初始化好。

文件 IO 的创建，数据库连接的创建，配置文件的读取等等这样一次性的工作，都适合在 open()方法中完成

在任务停止前调用，一般用来做清理工作

例如关闭数据库的链接等

import com.yanggu.flink.lowlevelapi.pojo.SensorReading
import org.apache.flink.api.common.accumulators.IntCounter
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._


/**
 * 测试自定义的RichFunction函数
 * open()方法、close()方法生命周期方法在每个并行度上只会被调用一次
 * getRuntimeContext()方法获取运行时上下文例如函 数执行的并行度, 任务的名字
 */
object UDFFunctionDemo {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = environment.fromCollection(List(
      SensorReading("sensor_1", System.currentTimeMillis(), 35.8),
      SensorReading("sensor_2", System.currentTimeMillis(), 45.8),
      SensorReading("sensor_3", System.currentTimeMillis(), 65.8)
    ))

    stream
      //这里设置了map的并行度为2
      .map(new MyMapFunction).setParallelism(2)
      .print("测试自定义RichFunction").setParallelism(1)

    //可以获取执行的结果
    val result = environment.execute("测试自定义RichFunction")

    //这里可以获取累加器的计算结果
    val value = result.getAccumulatorResult[Int]("MyFunction的计数器")
    println(s"MyFunction的计数器计算结果: $value")
  }

}

/**
 * Rich Function是Function函数的Rich版本,
 * 拥有一些声明周期方法, 可以获取运行时的上下文, 因此可以实现一些更复杂的功能
 *
 * @tparam IN  SensorReading 输入的数据类型
 * @tparam Out SensorReading 输出的数据类型
 */
class MyMapFunction extends RichMapFunction[SensorReading, SensorReading] {

  var subtask = 0

  var intCounter: IntCounter = _

  /**
   * open()方法是rich function的初始化方法,
   * 当一个算子例如map或者filter被调用之前open()会被执行, 且只会执行一次
   * 注意这句话
   * 在每个并行度上只会被调用一次
   *
   * @param parameters
   */
  override def open(parameters: Configuration): Unit = {
    //这里可以进行一些初始化的工作, 例如建立一个和hdfs的链接
    //在这里可以获取上下文信息
    this.subtask = getRuntimeContext.getIndexOfThisSubtask
    this.intCounter = getRuntimeContext.getIntCounter("MyFunction的计数器")
    //只会被打印2次, 设置了map的并行度为2
    println(Thread.currentThread() + "open方法被执行了")
  }

  /**
   * 默认生命周期方法, 最后一个方法, 做一些清理工作
   * 注意这句话
   * 在每个并行度上只会被调用一次, 而且是最后调用
   */
  override def close(): Unit = {
    //只会被打印2次, 设置了map的并行度为2
    println(Thread.currentThread() + "close方法被执行了")
    println(Thread.currentThread() + intCounter.toString)
  }

  /**
   * 需要重写的核心方法
   *
   * @param value
   * @return
   */
  override def map(value: SensorReading): SensorReading = {
    println(Thread.currentThread() + "map方法被执行了")
    intCounter.add(1)
    value
  }

}

import org.apache.flink.streaming.api.scala._

/**
 * 随机分区服从均匀分布（uniform distribution），所以可以把流中的数据随机打乱，均匀地传递到下游任务分区。
 * 因为是完全随机的，所以对于同样的输入数据, 每次执行得到的结果也不会相同
 * 上游算子并行度为1, 下游为4，上游算子的数据会随机发送到下游，每次执行的结果都不同
 */
//shuffle:4> 5
//shuffle:2> 4
//shuffle:3> 3
//shuffle:1> 1
//shuffle:1> 2
object ShuffleDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 读取数据源，并行度为 1
    val stream = env.fromElements(1, 2, 3, 4, 5)
    // 经洗牌后打印输出，并行度为 4
    stream
      .shuffle
      .print("shuffle").setParallelism(4)

    env.execute()
  }

}

import org.apache.flink.streaming.api.scala._

/**
 * rebalance使用的是 Round-Robin 负载均衡算法，可以将输入流数据平均分配到下游的并行任务中去
 */
//rebalance:2> 1
//rebalance:3> 2
//rebalance:4> 3
//rebalance:1> 4
//rebalance:2> 5
object RebalanceDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 读取数据源，并行度为 1
    val stream = env.fromElements(1, 2, 3, 4, 5)

    stream
      //默认的分区算法
      .rebalance
      .print("rebalance").setParallelism(4)

    env.execute()
  }

}

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._


/**
 * rescale是会将数据轮询发送到下游并行任务的一部分中
 * rebalance是上游的所有子任务都与下游的子任务建立联系, 然后轮询发送数据, 是一种笛卡尔积的关系(M * N)
 * 如果下游算子的并行度是上游的算子的整数倍, 那么只是上游算子只是与下游的部分算子建立联系, 然后轮询发送数据
 * 具体为上游并行度为2(source0、source1), 下游并行度为4(rescale1、rescale2、rescale3、rescale4)
 * source0中的数据为2、4、6、8
 * source1中的数据为1、3、5、7
 * 上游source0子任务和下游rescale1子任务、下游rescale2子任务建立联系, 同时轮询发送数据(2、4、6、8)
 * 上游source1子任务和下游rescale3子任务、下游rescale4子任务建立联系, 同时轮询发送数据(1、3、5、7)
 *
 */
//source:0> 2、4、6、8
//source:1> 1、3、5、7
//rescale:1> 2    rescale:2> 4    rescale:1> 6    rescale:2> 8
//rescale:3> 1    rescale:4> 3    rescale:3> 5    rescale:4> 7
object RescaleDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 这里使用了并行数据源的富函数版本
    // 这样可以调用 getRuntimeContext 方法来获取运行时上下文的一些信息
    env.addSource(new RichParallelSourceFunction[Int] {
      override def run(ctx: SourceFunction.SourceContext[Int]) = {
        for (i <- 0 until 8) {
          // 将奇数发送到索引为 1 的并行子任务
          // 将偶数发送到索引为 0 的并行子任务
          if ((i + 1) % 2 == getRuntimeContext.getIndexOfThisSubtask) {
            ctx.collect(i + 1)
          }
        }
      }

      override def cancel() = {}

    }).setParallelism(2)
      .rescale
      .print("rescale").setParallelism(4)

    env.execute()
  }

}

import org.apache.flink.streaming.api.scala._

//可以看到，数据被复制然后广播到了下游的所有并行任务中去了。
//broadcast:1> 1
//broadcast:2> 1
//broadcast:3> 1
//broadcast:4> 1
object BroadcastDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 读取数据源，并行度为 1
    val stream = env.fromElements(1)
    // 经广播后打印输出，并行度为 4
    stream
      .broadcast
      .print("broadcast").setParallelism(4)

    env.execute()
  }

}

import org.apache.flink.streaming.api.scala._

/**
 * 可以看到, 即使下游的print算子的并行度为2, 但是调用了global方法之后, 所有的数据都会发往下游的第一个算子
 * 可以理解为强行将下游算子的并行度调整为1
 */
//global:1> 1
//global:1> 2
//global:1> 3
//global:1> 4
object GlobalDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.fromElements(1, 2, 3, 4)

    stream.global
      .print("global").setParallelism(2)

    env.execute()
  }

}

import org.apache.flink.api.common.functions.Partitioner
import org.apache.flink.streaming.api.scala._

/**
 * 当Flink提供的所有分区策略都不能满足用户的需求时，我们可以通过使用partitionCustom()方法来自定义分区策略
 */
//CustomPartition:1> 2
//CustomPartition:2> 1
object CustomPartitionDemo {

  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 将自然数按照奇偶分区// 将自然数按照奇偶分区

    env
      .fromElements(1, 2, 3, 4, 5, 6, 7, 8)
      .partitionCustom(new Partitioner[Int]() {
        def partition(key: Int, numPartitions: Int) = key % 2
      }, data => data)
      .print("CustomPartition").setParallelism(2)

    env.execute()
  }

}

maxTimestamp - outOfOrdernessMillis - 1