欢迎光临
我们一直在努力
        找回密码
当前位置：171主机测评 > 服务器技术 > 正文

Flink单流转换算子实战解析

2026-06-23 分类：服务器技术 阅读(37) 评论(0)

目录

1. 环境初始化

2. 数据源

3. Map 转换

4. Filter 转换

5. FlatMap 转换

6. KeyBy 转换

7. 简单聚合

8. Reduce 转换

9. 富函数类测试

10. 执行任务

代码拓展

package transform

import org.apache.flink.api.common.functions.{FilterFunction, FlatMapFunction, MapFunction, ReduceFunction, RichMapFunction}
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

case class Event(user:String,url:String,timestamp:Long)
/**
*
* @PROJECT_NAME: flink1.13
* @PACKAGE_NAME: transform
* @author: 赵嘉盟-HONOR
* @data: 2025-05-19 2:45
* @DESCRIPTION
*
*/
object 单流转换算子 {
def main(args: Array[String]): Unit = {
val env=StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)

val data= env.fromElements(
Event("Mary", "./home", 100L),
Event("Sum", "./cart", 500L),
Event("King", "./prod", 1000L),
Event("King", "./root", 200L)
)
//TODO Map
val map=data.map(_.user).print("map")
val mapFunction=data.map(new MapFunction[Event,String] {
override def map(t: Event): String = t.user
}).print("mapFunction")

//TODO Filter
val filter=data.filter(_.user=="Sum").print("filter")
val filterFunction=data.filter(new FilterFunction[Event] {
override def filter(t: Event): Boolean = t.user.contains("m") //包含
}).print("filterFunction")

//TODO FlatMap
val flatMap=data.flatMap(new FlatMapFunction[Event,String] {
override def flatMap(t: Event, collector: Collector[String]): Unit = if(t.user=="Sum") collector.collect(t.url)
}).print("flatMapFunction")

//TODO KeyBy
val keyBy=data.keyBy(_.user).print("keyBy")
val keyByFunction=data.keyBy(new KeySelector[Event,String] {
override def getKey(in: Event): String = in.user
})

//TODO 简单聚合：Sun,Min,Max(抽取聚合前第一条数据),MinBy,MaxBy(抽取当前数据)
//统计当前最大时间戳
keyByFunction.max("timestamp").print("max")
keyByFunction.maxBy(2).print("maxBy")
//keyByFunction.maxBy("_2").print("maxByT") //元组根据位置取元素

//TODO 规约聚合 Reduce
//提取当前最活跃用户
data.map(data=>(data.user,1)).keyBy(_._1)
.reduce(new ReduceFunction[(String, Int)] {
override def reduce(t: (String, Int), t1: (String, Int)): (String, Int) = (t._1,t._2+t1._2)
})
//提取所有元素
.keyBy(data=>true)
.reduce((state,data)=>if(state._2>=data._2) state else data)
.print("reduceFunction")

//TODO 富函数类测试
data.map(new RichMapFunction[Event,Long] {
override def open(parameters: Configuration): Unit = println("索引号为"+getRuntimeContext.getIndexOfThisSubtask+"的任务开始")
override def close(): Unit = println("索引号为"+getRuntimeContext.getIndexOfThisSubtask+"的任务结束")
override def map(in: Event): Long = in.timestamp
})

env.execute("单流转换算子")
}
}

这段代码展示了 Apache Flink 中常用的单流转换算子（Single Stream Transformations）的使用。以下是代码的详细解释：

1. 环境初始化

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)

  • 创建 Flink 流处理执行环境。
  • 设置并行度为 1，表示所有操作在单线程中执行。
2. 数据源

val data = env.fromElements(
Event("Mary", "./home", 100L),
Event("Sum", "./cart", 500L),
Event("King", "./prod", 1000L),
Event("King", "./root", 200L)
)

  • 使用 fromElements 方法创建包含 4 个 Event 对象的流数据。
  • Event 是一个样例类，包含 user（用户）、url（URL）和 timestamp（时间戳）三个字段。
3. Map 转换

val map = data.map(_.user).print("map")
val mapFunction = data.map(new MapFunction[Event, String] {
override def map(t: Event): String = t.user
}).print("mapFunction")

  • map：使用 Lambda 表达式提取 Event 中的 user 字段。
  • mapFunction：使用 MapFunction 实现类提取 Event 中的 user 字段。
  • 结果：将 Event 转换为 String（用户名）并打印。
4. Filter 转换

val filter = data.filter(_.user == "Sum").print("filter")
val filterFunction = data.filter(new FilterFunction[Event] {
override def filter(t: Event): Boolean = t.user.contains("m")
}).print("filterFunction")

  • filter：过滤出 user 为 "Sum" 的事件。
  • filterFunction：过滤出 user 包含字母 "m" 的事件。
  • 结果：打印满足条件的事件。
5. FlatMap 转换

val flatMap = data.flatMap(new FlatMapFunction[Event, String] {
override def flatMap(t: Event, collector: Collector[String]): Unit =
if (t.user == "Sum") collector.collect(t.url)
}).print("flatMapFunction")

  • flatMap：如果 user 为 "Sum"，则提取 url 并发送到下游。
  • 结果：打印满足条件的 url。
6. KeyBy 转换

val keyBy = data.keyBy(_.user).print("keyBy")
val keyByFunction = data.keyBy(new KeySelector[Event, String] {
override def getKey(in: Event): String = in.user
})

  • keyBy：根据 user 字段对流数据进行分组。
  • keyByFunction：使用 KeySelector 实现类根据 user 字段分组。
  • 结果：将流数据按 user 分组。
7. 简单聚合

keyByFunction.max("timestamp").print("max")
keyByFunction.maxBy(2).print("maxBy")

  • max("timestamp")：按 user 分组后，提取每组中 timestamp 最大的记录。
  • maxBy(2)：按 user 分组后，提取每组中第 2 个字段（url）最大的记录。
  • 结果：打印每组中满足条件的记录。
8. Reduce 转换

data.map(data => (data.user, 1)).keyBy(_._1)
.reduce(new ReduceFunction[(String, Int)] {
override def reduce(t: (String, Int), t1: (String, Int)): (String, Int) =
(t._1, t._2 + t1._2)
})
.keyBy(data => true)
.reduce((state, data) => if (state._2 >= data._2) state else data)
.print("reduceFunction")

  • 第一步：将 Event 转换为 (user, 1)，按 user 分组后累加每个用户的点击次数。
  • 第二步：将所有用户数据分组到一个组中，找到点击次数最多的用户。
  • 结果：打印最活跃的用户。
9. 富函数类测试

data.map(new RichMapFunction[Event, Long] {
override def open(parameters: Configuration): Unit =
println("索引号为" + getRuntimeContext.getIndexOfThisSubtask + "的任务开始")
override def close(): Unit =
println("索引号为" + getRuntimeContext.getIndexOfThisSubtask + "的任务结束")
override def map(in: Event): Long = in.timestamp
})

  • RichMapFunction：在 map 转换中，使用富函数类。
  • open：在任务开始时执行，打印任务索引号。
  • close：在任务结束时执行，打印任务索引号。
  • map：提取 Event 中的 timestamp 字段。
10. 执行任务

env.execute("单流转换算子")

  • 启动 Flink 任务，执行上述所有转换操作。

代码拓展

  • 增加更多事件类型：

    • 扩展 Event 类，增加更多字段（如 eventType、ip 等），以支持更复杂的业务逻辑。

    case class Event(user: String, url: String, eventType: String, ip: String, timestamp: Long)

  • 动态数据源：

    • 使用 Kafka、Socket 或文件作为数据源，而不是硬编码的 fromElements。

    val data = env.addSource(new FlinkKafkaConsumer[Event]("topic", new EventSchema, properties))

  • 复杂聚合：

    • 使用 Window 进行时间窗口聚合，例如统计每 5 秒内每个用户的点击次数。

    data.keyBy(_.user)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .aggregate(new CountAggregator)

  • 自定义聚合函数：

    • 实现自定义的聚合函数，例如计算每个用户的平均点击时间间隔。

    class AvgIntervalAggregator extends AggregateFunction[Event, (Long, Int), Double] {
    override def createAccumulator(): (Long, Int) = (0L, 0)
    override def add(in: Event, acc: (Long, Int)): (Long, Int) = (acc._1 + in.timestamp, acc._2 + 1)
    override def getResult(acc: (Long, Int)): Double = acc._1.toDouble / acc._2
    override def merge(acc1: (Long, Int), acc2: (Long, Int)): (Long, Int) = (acc1._1 + acc2._1, acc1._2 + acc2._2)
    }

  • 多流合并：

    • 使用 union 或 connect 将多个流合并，进行联合分析。

    val stream1 = env.fromElements(Event("Mary", "./home", 100L))
    val stream2 = env.fromElements(Event("Sum", "./cart", 500L))
    val mergedStream = stream1.union(stream2)

  • 异常处理：

    • 在转换过程中增加异常处理逻辑，确保任务不会因为数据异常而中断。

    data.map(event => {
    try {
    // 业务逻辑
    } catch {
    case e: Exception => // 异常处理
    }
    })

  • 性能优化：

    • 使用 BroadcastState 或 Async I/O 优化性能，例如从外部数据库查询数据。

    val broadcastStream = data.broadcast
    val resultStream = data.connect(broadcastStream).process(new BroadcastProcessFunction[Event, Config, Result] {
    override def processElement(value: Event, ctx: BroadcastProcessFunction[Event, Config, Result]#ReadOnlyContext, out: Collector[Result]): Unit = {
    val config = ctx.getBroadcastState(configDescriptor)
    // 业务逻辑
    }
    override def processBroadcastElement(value: Config, ctx: BroadcastProcessFunction[Event, Config, Result]#Context, out: Collector[Result]): Unit = {
    ctx.getBroadcastState(configDescriptor).put("key", value)
    }
    })

    • 通过这些扩展，可以使代码更加灵活、健壮，并适应更复杂的业务场景。

    赞(0)
    未经允许不得转载：171主机测评 » Flink单流转换算子实战解析
    分享到： 更多 (0)
    标签：

    相关推荐

    评论 抢沙发

    取消

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址

    热门标签

    #人工智能(57823)#python(19697)#开发语言(16570)#java(15213)#运维(15146)#linux(14778)#大数据(14342)#服务器(12902)#算法(11608)#前端(10686)#网络(9214)#数据库(8749)#ai(8648)#c++(8051)#课程设计(6759)#学习(6641)#机器学习(6609)#区块链(5839)#OpenClaw(5378)#后端(5058)#spring boot(4956)#深度学习(4843)#c语言(4325)#毕业设计(4277)#数据结构(4145)#安全(3946)#docker(3890)#架构(3770)#javascript(3761)#大模型(3597)

    热门文章

    最新文章

    归档

    分类