文章目录
![在这里插入图片描述](https://img-blog.csdnimg.cn/20200718215950511.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NTA5OTIw,size_16,color_FFFFFF,t_70)
![在这里插入图片描述](https://img-blog.csdnimg.cn/2020071822000358.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NTA5OTIw,size_16,color_FFFFFF,t_70)
![在这里插入图片描述](https://img-blog.csdnimg.cn/2020071822003021.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NTA5OTIw,size_16,color_FFFFFF,t_70)
步骤
- 获取批处理运行环境
- 分别创建两个数据集
- 使用 RichMapFunction 对 成绩 数据集进行 map 转换
- 在数据集调用 map 方法后,调用 withBroadcastSet 将 学生 数据集创建广播
- 实现 RichMapFunction
a. 将成绩数据(学生 ID,学科,成绩) -> (学生姓名,学科,成绩)
b. 重写 open 方法中,获取广播数据
c. 导入 scala.collection.JavaConverters._ 隐式转换
d. 将广播数据使用 asScala 转换为 Scala 集合,再使用 toList 转换为 scala List 集合
e. 在 map 方法中使用广播进行转换 - 打印测试
package com.czxy.flink.batch
import java.util
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration
/**
* 需求:
* 创建一个 学生 数据集, 包含以下数据
* |学生 ID | 姓名 |
* |------|------|
* List((1, "张三"), (2, "李四"), (3, "王五"))
* 将该数据, 发布到广播。
* 再创建一个 成绩 数据集,
* |学生 ID | 学科 | 成绩 |
* |------|------|-----|
* List( (1, "语文", 50),(2, "数学", 70), (3, "英文", 86))
* 请通过广播获取到学生姓名, 将数据转换为
* List( ("张三", "语文", 50),("李四", "数学", 70), ("王五", "英文", 86))
*/
object BatchBroadcastDemo {
def main(args: Array[String]): Unit = {
/**
* 实现步骤:
* 1) 获取批处理运行环境
* 2) 分别创建两个数据集
* 3) 使用 RichMapFunction 对 成绩 数据集进行 map 转换
* 4) 在数据集调用 map 方法后, 调用 withBroadcastSet 将 学生 数据集创建广播
* 5) 实现 RichMapFunction
* a. 将成绩数据(学生 ID, 学科, 成绩) -> (学生姓名, 学科, 成绩)
* b. 重写 open 方法中, 获取广播数据
* c. 导入 scala.collection.JavaConverters._ 隐式转换d. 将广播数据使用 asScala 转换为 Scala 集合, 再使用 toList 转换为 scala List
* 集合
* e. 在 map 方法中使用广播进行转换
* 6) 打印测试
*/
//1. 获取批处理运行环境
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
import org.apache.flink.api.scala._
//2. 分别创建两个数据集 //创建学生数据集
val studentDataSet: DataSet[(Int, String)] = env.fromCollection(List((1, "张三"), (2, "李四"), (3, "王五")))
//创建成绩数据集
val scoreDataSet: DataSet[(Int, String, Int)] = env.fromCollection(List((1, "语文", 50), (2, "数学", 70), (3, "英文", 86)))
//3. 使用RichMapFunction 对成绩数据集进行map转换 返回值类型(学生名字,学科成名,成绩)
val resultDataSet: DataSet[(String, String, Int)] = scoreDataSet.map(new RichMapFunction[(Int, String, Int), (String, String, Int)] {
//定义获取学生数据集的集合
var stuMap: Map[Int, String] = null
//初始化的时候被执行一次,在对象的生命周期中只被执行一次
override def open(parameters: Configuration): Unit = {
//因为获取到的广播变量中的数据类型是java的集合类型,但是我们的代码是 scala因此需要将java的集合转换成scala的集合
// 我们这里将list转换成了map对象,之所以能够转换是因为list中的元素是对偶 元祖,因此可以转换成kv键值对类型
// 之所以要转换,是因为后面好用,传递一个学生id,可以直接获取到学生的名字
import scala.collection.JavaConversions._
val student: util.List[(Int, String)] = getRuntimeContext.getBroadcastVariable[(Int, String)]("student")
stuMap = student.toMap
}
//要对集合中的每个元素执行map操作,也就是说集合中有多少元素,就被执行多少 次
//每一条数据都会执行一次
override def map(value: (Int, String, Int)): (String, String, Int) = {
//(Int, String, Int)=》(学生id,学科名字,学生成绩)
val name: String = stuMap.getOrElse(value._1, "")
//返回值类型(学生名字,学科成名,成绩)
(name, value._2, value._3)
}
}).withBroadcastSet(studentDataSet, "student")
resultDataSet.print()
}
}