Flink分布式缓存Distributed Cache

1 分布式缓存

• Flink提供了一个分布式缓存，类似于hadoop，可以使用户在并行函数中很方便的读取本地文件，并把它放在taskmanager节点中，防止task重复拉取。
• 此缓存的工作机制如下：程序注册一个文件或者目录(本地或者远程文件系统，例如hdfs或者s3)，通过ExecutionEnvironment注册缓存文件并为它起一个名称。当程序执行，Flink自动将文件或者目录复制到所有taskmanager节点的本地文件系统，仅会执行一次。用户可以通过这个指定的名称查找文件或者目录，然后从taskmanager节点的本地文件系统访问它

• 2 使用技巧

  • 1：注册一个文件

      env.registerCachedFile("hdfs:///path/to/your/file", "hdfsFile")  

  • 2：访问数据

      File myFile = getRuntimeContext().getDistributedCache().getFile("hdfsFile");

　　3 应用案例实战

3.1 在D盘创建一个文件discache.txt，并进行registerCachedFile

3.2 每一个TaskManager都会存在一份，防止MapTask重复拉取文件。

import org.apache.commons.io.FileUtils
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration

object BatchDemoDisCacheScala {

  def main(args: Array[String]): Unit = {

    val env = ExecutionEnvironment.getExecutionEnvironment

    import org.apache.flink.api.scala._

    //1:注册文件
    env.registerCachedFile("d:\\data\\file\\a.txt","b.txt")

    val data = env.fromElements("a","b","c","d")

    val result = data.map(new RichMapFunction[String,String] {

      override def open(parameters: Configuration): Unit = {
        super.open(parameters)
        val myFile = getRuntimeContext.getDistributedCache.getFile("b.txt")
        val lines = FileUtils.readLines(myFile)
        val it = lines.iterator()
        while (it.hasNext){
          val line = it.next();
          println("line:"+line)
        }
      }
      override def map(value: String) = {
        value
      }
    })

    result.print()

  }

}

参考：

https://blog.****.net/shenshouniu/article/details/84499655

https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/batch/