java Flink（三十一）Flink分布式缓存

╰半橙微兮° 2022-08-30 04:29 257阅读 0赞

### **什么是Flink的分布式缓存?** ###

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

从上下文中就可以使用我们注册的文件。

### **代码：** ###

**  文件内容：**

**  **![20210719110941549.png][]

import org.apache.commons.io.FileUtils;
    import org.apache.flink.api.common.functions.RichMapFunction;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.io.File;
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * flink 分布式缓存
     */
    public class DistributedCache {
    
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            //将文件注册到上下文，第一个参数是文件地址，第二个是名称
            env.registerCachedFile("D:\\idle\\FlinkPrepared\\DistributedCache\\src\\main\\resources\\test.txt", "test");
            //模拟数据
            final DataStream<String> data = env.fromElements("a", "b", "c");
    
            DataStream<String> resultStream = data.map(new RichMapFunction<String, String>() {
                private ArrayList<String> dataList = new ArrayList<String>();
    
                @Override
                public void open(Configuration parameters) throws Exception {
                    super.open(parameters);
                    //使用文件
                    File file = getRuntimeContext().getDistributedCache().getFile("test");
                    List<String> lines = FileUtils. readLines(file);
                    for (String line : lines) {
                        this.dataList.add(line);
                        System.out.println("分布式缓存为:" + line);
                    }
                }
    
                public String map(String s) throws Exception {
                    System.out.println(dataList + ":" + s);
                    return s;
                }
            });
    
            resultStream.print();
            env.execute();
    
        }
    
    }

**    结果：**  
        ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

[20210719110941549.png]: /images/20220829/a55b60a96d3d4044b14c9c39373d7e7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20220829/793ca9a1b92645ab9f1e3a56b4ad5092.png