Flink的广播变量

1、Flink可以将数据广播到TaskManager上，数据存储到内存中，可以大量减少shuffle的操作；比如在数据join阶段，不可避免的就是大量的shuffle操作，我们可以把其中一个DataSet广播出去，加载到TaskManager的内存中，可以直接在内存中加载数据，这样就能避免大量的shuffle操作导致的集群性能下降。
2、广播变量创建后，可以被使用在集群任何一个节点的function上，而不需要多次传递给集群的节点。
3、可以理解为一个共享变量，我们可以把一个DataSet广播出去，然后在不同的Task节点上都能够获取到，这个变量在每个节点上存储一份在内存中。如果不使用broadcast，则在每个节点的每个task中都要拷贝一份DataSet数据集，比较浪费内存。

注意：
1、广播变量的DataSet不能太大，否则会OOM
2、广播变量在初始化广播出去之后不能修改，这样才能保证每个节点获取到值是一致的。
3、需要手动导入scala.collection.JavaConverters._将Java集合转换为Scala集合。

操作步骤

1.初始化数据
DataSet<Integer> toBroadcast = env.fromElements（1,2,3）
2.广播数据
.WithBroadcastSet(toBroadcast,"broadcastSetName");
3.获取数据
Collection<Integer> broadcastSet = getRuntimeContext().getBroadCastVariable("broadcastSetName");

批处理广播变量示例代码

import java.utilimport org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configurationobject BroadcastDemo{def main(args:Array[String]):Unit = {//1.获取批处理运行环境val env = ExecutionEnviroment.getExecutionEnviroment//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 stuDS:DataSet[(String,String,Int)] = scoreDataSet.map(new RichMapFunction[(Int,String,Int),(String,String,Int)]{var stuMap:Map[Int,String] = nulloverride def open(paraments:Configuration):Unit = {import scala.collection.JavaConversions._stuMap = getRuntimeContext.getBroadcastVariable[(Int,String)]("stuDS").toMap}override def map(value:(Int,String,Int)) = {// 获取学生IDval studentId:Int = value._1//获取学生nameval stuName:String = stuMap.getOrElse(studentId,studentId.toString)//构建元组(stuName,value._2,value._3)}}).WithBroadcastSet(studentDataSet,"stuDS")//4.打印测试stuDS.print()}
}

广播流变量

object BroadcastFlowDemo{def main(args:[String]):Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentval ds1:DataStream[(Int,Char)] = env.fromElements((1,"男"),(2,"女"))val ds2:DataStream[(Int,String,Int,String)] = env.socketTextStream("node02",8888).map(perline=>{val arr = perline.split(",")val id = arr(0).trim.toIntval name = arr(1).trim.toIntval genderFlg = arr(2).trim.toIntval address = arr(3).trim(id,name,genderFlg,address)})val desc = new MapStateDescriptor("genderInfo",BasicTypeInfo.INT_TYPE_INFO,BasicTypeInfo.CHAR_TYPE_INFO)val bcStream = ds1.broadcast(desc)ds2.connect(bcStream).process(new BroadCastProcessFunction[(Int,String,Int,String),(Int,Char),(Int,String,Char,String)](){override def processElement(value:(Int,String,Int,String),ctx:BroadcastProcessFunction[(Int,String,Int,String),(Int,Char),(Int,String,Char,String)],out:Collector[(Int,String,Char,String)]):Unit = {val genderFlg = value._3val gender = ctx.getBroadcastState(desc).get(genderFlg)out.collect((value._1,value._2,gender,value._4))}override def processBroadcastElement(value:(Int,Char),ctx:BroadcastProcessFunction[(Int,String,Int,String),(Int,Char),(Int,String.Char,String)],out:Collector[(Int,String,Char,String)]):Unit = {ctx.getBroadcastState(desc).put(value._1,value._2)}}).print()env.execute()}
}

Flink的累加器

Accumulator即累加器，与MapReduce counter的应用场景差不多，都能很好的观察task在运行期间的数据变化，可以在Flink job任务中的算子函数中操作累加器，但是只能在任务执行结束之后才能获取累加器的最终结果。
Flink目前有以下的内置累加器，每个累加器都实现了Accmulator接口。
·IntCounter
·LongCounter
·DoubleCounter
·Histogram(直方图)
·自定义（实现SimpleAccumulator接口）

操作步骤：

1.创建累加器
private Intcounter numLines = new IntCunter();
2.注册累加器
getRuntimeContext().addAccumulator(“num-lines”, this.numLines);
3.使用累加器
this.numLines.add(1);
4.获取累加器的结果
myJobExecutionResult.getAccumulatorResult(“num-lines”)

object AccumulatorDemo{def main(args:Array[String]):Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.aoi.scala._env.readTextFile("文件路径").filter(_.trim.nonEmpty).map(perline=>{val arr = perline.split(",")Raytek(arr(0),arr(1).trim.toDouble,arr(2).trim,arr(3).toLong,arr(4).trim)}).map(new RichMapFunction[Raytek,String]{val total = new LongCounter(0)val normal = new LongCounter(0)val exception = new LongCounter(0)override def open(parameters:Configation):Unit = {getRuntimeContext.addAccumulator("total",total)getRuntimeContext.addAccumulator("normal",normal)getRuntimeContext.addAccumulator("exception",exception)}override def map(value:Raytek):String= {total.add(1)var msg = "温度过高，需要隔离进行排查！！！"if (value.temperature >= 36.3 && value.temperature <= 37.2){normal.add(1)(s"红外测温仪id：${value.id},旅客名：${value.name},体温：${value.temperature},体温正常，准予放行...")}else{exception.add(1)(s"红外测温仪id:${value.id},旅客名字：${value.name},体温:${value.temperture},应对措施：$【$msg】")}}}).print()val result:JobExceptionResult = env.execute()val total = result.getAccumulatorResult[Long]("total")val normal = result.getAccumulatorResult[Long]("normal")val exception = result.getAccumulatorResult[Long]("exception")println(s"今天统计的情况如下：旅客总数：${total},体温正常的旅客数：${normal},体温异常的旅客数：${exception}")}
}

Flink的分布式缓存

1.Flink提供了一个类似于Hadoop 的分布式缓存，让并行的实例可以方便的访问到本地文件，并把它放在TaskManager节点中，防止task重复拉取；
2.当程序执行Flink自动将文件拉取TaskManager节点本地文件系统，仅会执行一次；
3.可以理解为广播变量，这里广播的是文件；
注意：广播变量是将变量分发到各个worker节点的内存上，而分布式缓存是将文件缓存到各个worker节点上。

操作步骤：

1.注册一个文件
env.registerCacheFile(“hdfs文件路径”)
2.访问数据
File myFile = getRuntimeContext().getDistributeCache().getFile(“abc”);

案例代码：

object DistributeCacheDemo{def main(args:Array[String]):Unit = {val env = StreamExecutionEnviroment.getExecutionEnviromentimport org.apache.flink.api.scala._env.registerCacheFile("hdfs上文件的路径")env.socketTextStream("node01",8888).filter(_.trim.nonEmpty).map(new RichMapFunction[String,(Int,String,Char,String)](){var buffer:ListBuffer[(Int,Char)] = ListBuffer()var map:Map[Int,Char] = _override def open(parameters:Configation):Unit = {var file:File = getRuntimeContext.getDistributedCache.getFile("genderInfo")val src = Source.fromFile(file)	val lines = src.getLines().toListfor(perLine<-lines){val arr = perLine.split(",")var id = arr(0).trim.toIntvar gender = arr(1).trim.toCharArray()(0)buffer.append((id,gender))}map = buffer.toMapsrc.close}override def map(value:String):(Int,String,Char,String) = {val info = value.spalit(",")val id = info(0).trim.toIntval name = info(1).trimval gender = map.getOrElse(info(2).trim.toInt,"男")val address = info(3).trim(id,name,gender,address)}}).print()env.execute()}
}