猫先生头条

Flink处理函数解析（ProcessFunction和KeyedProcessFunction）

2023-12-20 22:10 由 Toycon 发表于 #数据库

Flink中的处理函数（ProcessFunction和KeyedProcessFunction）在对于数据进行颗粒化的精确计算时使用较多，处理函数提供了一个定时服务（TimerService），可以向未来注册一个定时服务，我们可以把它理解为一个闹钟，当闹钟响起时，就调用ProcessFunction中的onTimer()方法，会对数据进行一些计算。我们来解析一下这两个函数。

本文基于Flink1.14版本

ProcessFunction

ProcessFunction是Flink中的较为底层的API，当我们对于DataStream调用process函数的时候，需要在里面传入一个对象，即new ProcessFunction[] {}，ProcessFunction是一个抽象类，我们看一下这个抽象类的源码：

源码解析

@PublicEvolving
public abstract class ProcessFunction<I, O> extends AbstractRichFunction {
    private static final long serialVersionUID = 1L;

    public ProcessFunction() {
    }

    public abstract void processElement(I var1, ProcessFunction<I, O>.Context var2, Collector<O> var3) throws Exception;

    public void onTimer(long timestamp, ProcessFunction<I, O>.OnTimerContext ctx, Collector<O> out) throws Exception {
    }

    public abstract class OnTimerContext extends ProcessFunction<I, O>.Context {
        public OnTimerContext() {
            super();
        }

        public abstract TimeDomain timeDomain();
    }

    public abstract class Context {
        public Context() {
        }

        public abstract Long timestamp();

        public abstract TimerService timerService();

        public abstract <X> void output(OutputTag<X> var1, X var2);
    }
}

我们可以看到ProcessFunction这个抽象类有两个泛型参数<I, O>分别是输入（Input）和输出（Output）以及它继承了AbstractRichFunction。
ProcessFunction中processElement这个方法是一个抽象方法，代表实现这个抽象类时必须实现这个方法。这个方法的参数有三个I var1, ProcessFunction<I, O>.Context var2, Collector<O> var3,我们来解析一下：var1的类型是上面抽象类的泛型I，代表输入的数据，var2的类型是ProcessFunction<I, O>.Context，它是 Flink 中 ProcessFunction 的一个内部抽象类，用于为 processElement 方法提供上下文信息。

ProcessFunction<I, O>.Context 的主要功能：

1.获取时间戳：

timestamp(): 这个方法返回当前正在处理的事件的时间戳。如果您使用的是事件时间（EventTime），这个时间戳就是事件本身的时间戳；如果是处理时间（ProcessingTime），这个值可能是 null。

2.定时服务访问：

timerService(): 提供对 TimerService 的访问，您可以用它来注册定时器和查询当前时间。这对于需要基于时间做决策的应用来说非常重要。

3.侧输出：

output(OutputTag tag, X value): 允许您将数据发送到非主输出（即侧输出）。侧输出可以用于将数据发送到多个不同的流，或者用于处理特殊情况的数据，例如错误记录或监控事件。

最后Collector<O> 是一个泛型接口，其中 O 表示输出数据类型，可以使用 Collector 来发射任何类型的数据。Collector 允许从一个输入产生多个输出，这在处理复杂的逻辑或进行多阶段处理时非常有用。

我们来看一个例子：

现在我们有订单数据，需要处理每个订单字符串，如果订单金额超过100，则将其标记为"大订单"，否则标记为"小订单"。

假设我们的订单数据是一个简单的字符串格式，包含订单ID和订单金额，格式为 "orderId,amount"：

    // 示例订单数据流
    val orderDataStream: DataStream[String] = env.fromElements(
      "order1,100.5",
      "order2,50.5",
      "order3,200.0"
    )
1.定义样例类

目的是为了将数据转为对象，便于操作

case class Order(orderId: String, amount: Double)
2.使用 ProcessFunction 处理订单数据
    val processedStream = orderDataStream
      .process(new OrderProcessFunction)  //OrderProcessFunction为自己实现的ProcessFunction
3.实现OrderProcessFunction
class OrderProcessFunction extends ProcessFunction[String, String] {
  override def processElement(
      value: String,
      ctx: ProcessFunction[String, String]#Context,
      out: Collector[String]): Unit = {

    // 解析订单数据
    val parts = value.split(",")	//使用逗号分割数据
    val order = Order(parts(0), parts(1).toDouble)	//将数据转为对象

    // 根据金额分类订单
    val orderType = if (order.amount > 100) "大订单" else "小订单"
    out.collect(s"订单ID: ${order.orderId}, 类型: $orderType")	//将数据发送输出
  }
}
3.打印处理后的结果
    processedStream.print()

结果为：

订单ID: order1, 类型: 大订单
订单ID: order2, 类型: 小订单
订单ID: order3, 类型: 大订单
完整代码：
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

// 定义订单数据的样例类
case class Order(orderId: String, amount: Double)

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

    // 示例订单数据流
    val orderDataStream: DataStream[String] = env.fromElements(
      "order1,100.5",
      "order2,50.5",
      "order3,200.0"
    )

    // 使用 ProcessFunction 处理订单数据
    val processedStream = orderDataStream
      .process(new OrderProcessFunction)

    // 打印处理后的结果
    processedStream.print()

    env.execute("Order Processing")
  }
}

class OrderProcessFunction extends ProcessFunction[String, String] {
  override def processElement(
      value: String,
      ctx: ProcessFunction[String, String]#Context,
      out: Collector[String]): Unit = {

    // 解析订单数据
    val parts = value.split(",")
    val order = Order(parts(0), parts(1).toDouble)

    // 根据金额分类订单
    val orderType = if (order.amount > 100) "大订单" else "小订单"
    out.collect(s"订单ID: ${order.orderId}, 类型: $orderType")
  }
}

KeyedProcessFunction

KeyedProcessFunction允许对一个键控（keyed）流的每个元素进行操作，并提供对状态和定时器的访问。这使得它非常适合处理需要细粒度控制的场景。

键控流：在 Flink 中，一个键控流是通过 keyBy 方法从一个普通流创建的。在键控流中，所有的元素都根据指定的键分组。

KeyedProcessFunction 有两个主要方法：

processElement：

输入：每个流元素，以及上下文信息（Context）。
功能：处理每个元素，可以访问当前元素的键、状态和定时器。
使用场景：适用于每个事件的独立处理，如更新状态、发射输出、注册定时器等。

onTimer：

输入：定时器触发的时间戳，以及定时器上下文信息（OnTimerContext）。
功能：在特定时间点触发的逻辑处理。
使用场景：适用于需要基于时间的操作，如超时检测、定时输出等。

KeyedProcessFunction 还有附加特性:

状态管理：KeyedProcessFunction 支持 Flink 的状态管理，可以为每个键保存和更新状态。这在需要跟踪每个键的历史信息或执行聚合操作时非常有用。

定时器：可以注册处理时间（processing time）或事件时间（event time）定时器。定时器在指定的时间点触发，执行 onTimer 方法中的逻辑。

侧输出（Side Outputs）：除了主输出流外，还可以发射数据到侧输出标签。这允许您处理如异常数据、监控事件等特殊用途的输出。

源码解析

我们来看一下KeyedProcessFunction的源码：

@PublicEvolving
public abstract class KeyedProcessFunction<K, I, O> extends AbstractRichFunction {
    private static final long serialVersionUID = 1L;

    public KeyedProcessFunction() {
    }

    public abstract void processElement(I var1, KeyedProcessFunction<K, I, O>.Context var2, Collector<O> var3) throws Exception;

    public void onTimer(long timestamp, KeyedProcessFunction<K, I, O>.OnTimerContext ctx, Collector<O> out) throws Exception {
    }

    public abstract class OnTimerContext extends KeyedProcessFunction<K, I, O>.Context {
        public OnTimerContext() {
            super();
        }

        public abstract TimeDomain timeDomain();

        public abstract K getCurrentKey();
    }

    public abstract class Context {
        public Context() {
        }

        public abstract Long timestamp();

        public abstract TimerService timerService();

        public abstract <X> void output(OutputTag<X> var1, X var2);

        public abstract K getCurrentKey();
    }
}

KeyedProcessFunction<K, I, O> 是一个抽象类，用于处理键控流中的元素。它允许对每个元素进行复杂的处理，并提供了对定时器和状态的控制。其中 K (即"Key)表示键的类型，I 表示输入元素的类型，O 表示输出元素的类型。

KeyedProcessFunction继承自 AbstractRichFunction，表明 KeyedProcessFunction 可以访问 Flink 的富函数（rich function）特性，如生命周期方法和运行时上下文。

processElement方法和onTimer方法与ProcessFunction中的类似

OnTimerContext类 是 Context 的子类，为定时器提供额外的上下文信息。其中的timeDomain() 能返回当前定时器的时间域（处理时间或事件时间）。getCurrentKey() 可以返回当前处理的键。

Context 类中的 timestamp() 用于返回当前处理元素的时间戳。timerService()提供对定时器服务的访问，允许注册和取消定时器。output(OutputTag<X> var1, X var2)允许将数据发送到侧输出。getCurrentKey()返回当前处理的键。


下面我们来看一个使用KeyedProcessFunction处理订单数据的例子：

例子二：

假设我们有订单数据流，每个订单包含订单ID、用户ID、订单金额和时间戳。现在我们需要计算每个用户的订单总金额，同时对于每个用户，如果在指定时间内没有新的订单，发出一个警告消息，以及对于超过特定金额的订单，将其发送到一个侧输出。

数据源：

    // 示例订单数据流
    val orderDataStream: DataStream[Order] = env.fromElements(
      Order("order1", "user1", 50.0, 1000),
      Order("order2", "user2", 300.0, 2000),
      Order("order3", "user1", 200.0, 3000),
    )
1.定义订单样例类
case class Order(orderId: String, userId: String, amount: Double, timestamp: Long)
2.定义侧输出标签
val largeOrderTag = new OutputTag[String]("large-order")
3.使用 ProcessFunction 处理订单数据
    val processedStream = orderDataStream
      .keyBy(_.userId)
      .process(new ComplexOrderProcessFunction)
4.实现 ComplexOrderProcessFunction
class ComplexOrderProcessFunction extends KeyedProcessFunction[String, Order, String] {
  // 每个用户的订单总额
  val userAmounts = mutable.HashMap[String, Double]()
  // 定时器时间
  val timerInterval: Long = 60000 // 1分钟

  override def processElement(
      order: Order,
      ctx: KeyedProcessFunction[String, Order, String]#Context,
      out: Collector[String]): Unit = {

    // 更新用户的订单总额
    val totalAmount = userAmounts.getOrElse(order.userId, 0.0) + order.amount
    userAmounts(order.userId) = totalAmount

    // 检查订单是否超过特定金额，如果是，则输出到侧输出
    if (order.amount > 250.0) {
      ctx.output(largeOrderTag, s"大额订单: ${order.orderId} - ${order.amount}")
    }

    // 注册定时器，当前时间加上间隔
    ctx.timerService().registerEventTimeTimer(order.timestamp + timerInterval)

    // 输出用户的订单总额
    out.collect(s"用户 ${order.userId} 的订单总额: $totalAmount")
  }

  override def onTimer(
      timestamp: Long,
      ctx: KeyedProcessFunction[String, Order, String]#OnTimerContext,
      out: Collector[String]): Unit = {
    // 定时器触发，发出警告
    out.collect(s"用户 ${ctx.getCurrentKey} 在过去的一分钟内没有新的订单。")
  }
}
5.打印输出
    // 打印主输出
    processedStream.print()

    // 打印侧输出
    processedStream.getSideOutput(largeOrderTag).print("large-orders")

在这个例子中：

  • 当一个订单到达时，我们更新基于用户ID的订单总额。
  • 如果订单金额超过 250，则将其作为大额订单输出到侧输出。
  • 为每个用户设置一个定时器，如果在一分钟内没有新订单，则发出警告。
  • 在主输出中，我们发射每个用户的订单总额。
结果：
large-orders:1> 大额订单: order2 - 300.0
1> 用户 user2 的订单总额: 300.0
1> 用户 user2 在过去的一分钟内没有新的订单。
12> 用户 user1 的订单总额: 50.0
12> 用户 user1 的订单总额: 250.0
12> 用户 user1 在过去的一分钟内没有新的订单。
12> 用户 user1 在过去的一分钟内没有新的订单。
完整代码：
// 定义订单样例类
case class Order(orderId: String, userId: String, amount: Double, timestamp: Long)

object ComplexOrderProcessing {

  // 侧输出标签
  val largeOrderTag = new OutputTag[String]("large-order")

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

    // 示例订单数据流
    val orderDataStream: DataStream[Order] = env.fromElements(
      Order("order1", "user1", 50.0, 1000),
      Order("order2", "user2", 300.0, 2000),
      Order("order3", "user1", 200.0, 3000),
    )

    // 使用 ProcessFunction 处理订单数据
    val processedStream = orderDataStream
      .keyBy(_.userId)
      .process(new ComplexOrderProcessFunction)

    // 打印主输出
    processedStream.print()

    // 打印侧输出
    processedStream.getSideOutput(largeOrderTag).print("large-orders")

    env.execute("Complex Order Processing")
  }
}

class ComplexOrderProcessFunction extends KeyedProcessFunction[String, Order, String] {
  // 每个用户的订单总额
  val userAmounts = mutable.HashMap[String, Double]()
  // 定时器时间
  val timerInterval: Long = 60000 // 1分钟

  override def processElement(
                               order: Order,
                               ctx: KeyedProcessFunction[String, Order, String]#Context,
                               out: Collector[String]): Unit = {

    // 更新用户的订单总额
    val totalAmount = userAmounts.getOrElse(order.userId, 0.0) + order.amount
    userAmounts(order.userId) = totalAmount

    // 检查订单是否超过特定金额，如果是，则输出到侧输出
    if (order.amount > 250.0) {
      ctx.output(largeOrderTag, s"大额订单: ${order.orderId} - ${order.amount}")
    }

    // 注册定时器，当前时间加上间隔
    ctx.timerService().registerEventTimeTimer(order.timestamp + timerInterval)

    // 输出用户的订单总额
    out.collect(s"用户 ${order.userId} 的订单总额: $totalAmount")
  }

  override def onTimer(
                        timestamp: Long,
                        ctx: KeyedProcessFunction[String, Order, String]#OnTimerContext,
                        out: Collector[String]): Unit = {
    // 定时器触发，发出警告
    out.collect(s"用户 ${ctx.getCurrentKey} 在过去的一分钟内没有新的订单。")
  }
}

以上就是本文的全部内容啦(͏ ˉ ꈊ ˉ)✧˖°

热门相关：扑倒老公大人   陆鸣至尊神殿   都市最强小村医   一等狂后：绝色驭兽师   我有一张均富卡