快速入门Flink (6) —— Flink的广播变量、累加器与分布式缓存

import java.util

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration
import org.apache.flink.api.scala._
/*
 * @Author: Alice菌
 * @Date: 2020/8/1 20:30
 * @Description: 
    
 */
object BatchBroadcastDemo {
  def main(args: Array[String]): Unit = {

    // 1、创建批处理运行环境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    // 2、分别创建两个数据集
    // 创建学生数据集
    val stuDataSet: 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 result: DataSet[(String, String, Int)] = scoreDataSet.map(
      new RichMapFunction[(Int, String, Int), (String, String, Int)] {

        // 定义获取学生数据集的集合
        var studentMap: Map[Int, String] = _

        // 初始化的时候被执行一次,在对象的生命周期中只被执行一次
        override def open(parameters: Configuration): Unit = {
          // 因为获取到的广播变量中的数据类型是java的集合类型，但是我们的代码是scala，因此需要将java的集合转换成scala的集合
          // 我们这里将list转换成了map对象,之所以能够转换是因为list中的元素是对偶元组，因此可以转换成 kv 键值对类型
          // 之所以要转换，是因为后面好用，传递一个学生id，可以直接获取到学生的名字
          import scala.collection.JavaConversions._
          // 获取到广播变量的内容
          val studentList: util.List[(Int, String)] = getRuntimeContext.getBroadcastVariable[(Int, String)]("student")
            
          studentMap = studentList.toMap
           
        }
        // 要对集合中的每个元素执行map操作，也就是说集合中有多少元素，就被执行多少次
        override def map(value: (Int, String, Int)): (String, String, Int) = {
          //(Int, String, Int)=》（学生id，学科名字，学生成绩）
          //返回值类型(学生名字，学科名，成绩)
          val stuId: Int = value._1
          val stuName: String = studentMap.getOrElse(stuId, "")

          //(学生名字，学科名，成绩)
          (stuName, value._2, value._3)
        }
      }).withBroadcastSet(stuDataSet,"student")

    result.print()
    //(张三,语文,50)
    //(李四,数学,70)
    //(王五,英文,86)
    
  }
}

import java.io.File

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration
import org.apache.flink.api.scala._

import scala.io.Source
/*
 * @Author: Alice菌
 * @Date: 2020/8/1 22:40
 * @Description: 
    
 */
object BatchDisCachedFile {
  def main(args: Array[String]): Unit = {

    // 获取批处理运行环境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    // 注册一个分布式缓存
    env.registerCachedFile("hdfs://node01:8020/test/input/distribute_cache_student","student")
    /*
    1,张三
    2,李四
    3,王五
     */
    // 创建成绩数据集
    val scoreDataSet: DataSet[(Int, String, Int)] = env.fromCollection(List((1, "语文", 50), (2, "数学", 70), (3, "英文", 86)))

    val resultDataSet: DataSet[(String, String, Int)] = scoreDataSet.map(
      new RichMapFunction[(Int, String, Int), (String, String, Int)] {

        var studentMap: Map[Int, String] = _

        // 初始化的时候被调用一次
        override def open(parameters: Configuration): Unit = {
          // 获取分布式缓存的文件
          val studentFile: File = getRuntimeContext.getDistributedCache.getFile("student")

          val linesIter: Iterator[String] = Source.fromFile(studentFile).getLines()
          studentMap = linesIter.map(lines => {
            val words: Array[String] = lines.split(",")
            (words(0).toInt, words(1))
          }).toMap
        }
        
        override def map(value: (Int, String, Int)): (String, String, Int) = {

          val stuName: String = studentMap.getOrElse(value._1, "")
          (stuName, value._2, value._3)
        }
      })

    // 输出结果
    resultDataSet.print()
    //(张三,语文,50)
    //(李四,数学,70)
    //(王五,英文,86)
    
  }
}

import org.apache.flink.api.common.JobExecutionResult
import org.apache.flink.api.common.accumulators.IntCounter
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode


/*
 * @Author: Alice菌
 * @Date: 2020/8/1 23:26
 * @Description: 
    
 */
object BatchCounterDemo {
  def main(args: Array[String]): Unit = {

    //1、创建执行环境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    //2、创建执行环境
    val sourceDataSet: DataSet[String] = env.fromElements("a","b","c","d")

    //3、对sourceDataSet 进行map操作
    val resultDataSet: DataSet[String] = sourceDataSet.map(new RichMapFunction[String, String] {
      // 创建累加器
      val counter: IntCounter = new IntCounter

      // 初始化的时候执行一次
      override def open(parameters: Configuration): Unit = {
        // 注册累加器
        getRuntimeContext.addAccumulator("MyAccumulator", this.counter)
      }

      // 初始化的时候被执行一次
      override def map(value: String): String = {
        counter.add(1)
        value
      }
    })
    resultDataSet.writeAsText("data/output/Accumulators",WriteMode.OVERWRITE)
    val result: JobExecutionResult = env.execute("BatchCounterDemo")

    val MyAccumlatorValue: Int = result.getAccumulatorResult[Int]("MyAccumulator")

    println("累加器的值:"+MyAccumlatorValue)
    //累加器的值:4
  }
}