七、Flink的分布式缓存

绝地灬酷狼 2024-04-18 23:11 0阅读 0赞

## 分布式缓存 ##

Flink提供了一个分布式缓存，类似于hadoop，可以使用户在并行函数中很方便的读取本地文件，并把它放在taskmanager节点中，防止task重复拉取。

此缓存的工作机制如下：

*  程序注册一个文件或者目录(本地或者远程文件系统，例如hdfs或者s3)，通过ExecutionEnvironment注册缓存文件并为它起一个名称。
 *  当程序执行，Flink自动将文件或者目录复制到所有taskmanager节点的本地文件系统，仅会执行一次。
 *  用户可以通过这个指定的名称查找文件或者目录，然后从taskmanager节点的本地文件系统访问它。

## 示例 ##

在ExecutionEnvironment中注册一个文件：

//获取运行环境
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
    //1：注册一个文件,可以使用hdfs上的文件 也可以是本地文件进行测试
    env.registerCachedFile("/Users/wangzhiwu/WorkSpace/quickstart/text","a.txt");

在**用户函数中访问缓存文件或者目录**(这里是一个map函数)。这个函数**必须继承RichFunction**,因为它需要使用RuntimeContext读取数据:

object DfsCacheTest {
      import org.apache.flink.api.scala._
      def func = new RichMapFunction[String,String] {
        var list :List[(String)] = _
    
        override def open(parameters: Configuration): Unit = {
          super.open(parameters)
          super.open(parameters)
          // 获取缓存的数据
          val file = getRuntimeContext.getDistributedCache.getFile("testfile")
          val lines = Source.fromFile(file.getAbsoluteFile).getLines()
          list = lines.toList
        }
        override def map(value: String): String = {
          var middle: String = ""
          if(list.head.contains(value)) {
            middle = value
          }
          middle
        }
      }
      def main(args: Array[String]): Unit = {
        val env = ExecutionEnvironment.getExecutionEnvironment
        //    固定间隔： 间隔10秒 重启3次
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, Time.of(10, TimeUnit.SECONDS) ))
        //    分布式缓存就相当于spark的广播,把一个变量广播到所有的executor上,也可以看做是Flink的广播流,只不过这里广播的是一个文件.
        // 注册缓存的文件,里面有数据hello,jason
        env.registerCachedFile("cache/test.txt", "testfile")
        val stream = env.fromElements("hello", "jason", "hello", "jim")
        val result = stream.flatMap(_.split(","))
          .map(func)
          .map((_,1L))
          .filter(_._1.nonEmpty)
          .groupBy(0)
          .sum(1)
        result.print()
    //(jason,1)
    //(hello,2)
      }
    }