Flink分布式缓存Distributed Cache应用案例

论坛 期权论坛 脚本     
匿名网站用户   2020-12-20 10:36   54   0

转载自:https://blog.csdn.net/shenshouniu/article/details/84558874

欢迎加入大数据学习群:
**Flink学习视频:**http://edu.51cto.com/sd/88e07

在这里插入图片描述

你可能感兴趣的文章:

1-Flink入门
2-本地环境搭建&构建第一个Flink应用
3-DataSet API
4-DataSteam API
5-集群部署
6-分布式缓存
7-重启策略
8-Flink中的窗口
9-Flink中的Time
Flink时间戳和水印
Broadcast广播变量
FlinkTable&SQL
Flink实战项目实时热销排行
Flink写入RedisSink
Flink消费Kafka写入Mysql
Flink组件和逻辑计划
Flink执行计划生成
JobManager中的基本组件(1)
JobManager中的基本组件(2)
JobManager中的基本组件(3)
TaskManager
算子
网络
水印WaterMark
CheckPoint
任务调度与负载均衡
异常处理
Alibaba Blink新特性

欢迎关注‘大数据技术与架构’ 微信公众号,获取更多关于Spark、flink、hadoop的资料下载。

1 分布式缓存

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

2 使用技巧

  • 1:注册一个文件

      env.registerCachedFile("hdfs:///path/to/your/file", "hdfsFile")  
    
    • 1
  • 2:访问数据

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

3 应用案例实战

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

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

public class BatchDemoDisCache {
public static void main(String[] args) throws Exception{

    //获取运行环境
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

    //1:注册一个文件,可以使用hdfs或者s3上的文件
    env.registerCachedFile("d:\\discache.txt","a.txt");

    DataSource<String> data = env.fromElements("a", "b", "c", "d");

    DataSet<String> result = data.map(new RichMapFunction<String, String>() {
        private ArrayList<String> dataList = new ArrayList<String>();

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            //2:使用文件
            File myFile = getRuntimeContext().getDistributedCache().getFile("a.txt");
            List<String> lines = FileUtils.readLines(myFile);
            for (String line : lines) {
                this.dataList.add(line);
                System.out.println("discache:" + line);
            }
        }
        
        @Override
        public String map(String value) throws Exception {
            //在这里就可以使用dataList
            return value;
        }
    });
    result.print();
}

3.3 结果展示

discache:flink
discache:spark
discache:hadoop
discache:kylin
a
b
c
d

4 总结收尾

分享到 :
0 人收藏
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

积分:1136255
帖子:227251
精华:0
期权论坛 期权论坛
发布
内容

下载期权论坛手机APP