欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Flink DataSet API 之 Distributed Cache(分布式缓存)

程序员文章站 2022-03-08 09:09:27
...

基本介绍

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

用法

1:注册一个文件:    env.registerCachedFile("hdfs:///path/to/your/file", "hdfsFile") 

2:访问数据:          File myFile = getRuntimeContext().getDistributedCache().getFile("hdfsFile");

使用Demo

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

/**
  * Distributed Cache
  */
object BatchDemoDisCache {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.api.scala._
    env.registerCachedFile("d:/sshdata/count", "count")
    val data = env.fromElements("a","b","c","d")
    val result = data.map(new RichMapFunction[String, String] {
      override def open(parameters: Configuration) = {
        super.open(parameters)
        val myFile = getRuntimeContext.getDistributedCache.getFile("count")
        val lines = FileUtils.readLines(myFile)
        val it = lines.iterator()
        while (it.hasNext) {
          val line = it.next()
          println(s"line: $line")
        }
      }

      override def map(in: String): String = {
        in
      }
    })
    result.print()
  }
}

 

相关标签: Flink