Flink的分布式缓存
分布式缓存
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读取数据:
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.err.println("分布式缓存为:" + line); } } @Override public String map(String value) throws Exception { //在这里就可以使用dataList System.err.println("使用datalist:" + dataList + "------------" +value); //业务逻辑 return dataList +":" + value; } }); result.printToErr(); }
完整代码如下,仔细看注释:
public class DisCacheTest { public static void main(String[] args) throws Exception{ //获取运行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); //1:注册一个文件,可以使用hdfs上的文件 也可以是本地文件进行测试 //text 中有4个单词:hello flink hello FLINK env.registerCachedFile("/Users/wangzhiwu/WorkSpace/quickstart/text","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.err.println("分布式缓存为:" + line); } } @Override public String map(String value) throws Exception { //在这里就可以使用dataList System.err.println("使用datalist:" + dataList + "------------" +value); //业务逻辑 return dataList +":" + value; } }); result.printToErr(); } }//
输出结果如下:
[hello, flink, hello, FLINK]:a [hello, flink, hello, FLINK]:b [hello, flink, hello, FLINK]:c [hello, flink, hello, FLINK]:d
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Flink重启策略
概述 Flink支持不同的重启策略,以在故障发生时控制作业如何重启 集群在启动时会伴随一个默认的重启策略,在没有定义具体重启策略时会使用该默认策略。 如果在工作提交时指定了一个重启策略,该策略会覆盖集群的默认策略默认的重启策略可以通过 Flink 的配置文件 flink-conf.yaml 指定。配置参数 restart-strategy 定义了哪个策略被使用。 常用的重启: 1.策略固定间隔 (Fixed delay) 2.失败率 (Failure rate) 3.无重启 (No restart) 如果没有启用 checkpointing,则使用无重启 (no restart) 策略。如果启用了 checkpointing,但没有配置重启策略,则使用固定间隔 (fixed-delay) 策略 重启策略可以在flink-conf.yaml中配置,表示全局的配置。也可以在应用代码中动态指定,会覆盖全局配置 固定间隔 第一种:全局配置 flink-conf.yaml restart-strategy: fixed-delay restart-strategy.fixed-delay.att...
- 下一篇
Flink中的窗口
窗口 窗口类型 flink支持两种划分窗口的方式(time和count) 如果根据时间划分窗口,那么它就是一个time-window 如果根据数据划分窗口,那么它就是一个count-window flink支持窗口的两个重要属性(size和interval) 如果size=interval,那么就会形成tumbling-window(无重叠数据) 如果size>interval,那么就会形成sliding-window(有重叠数据) 如果size 通过组合可以得出四种基本窗口: time-tumbling-window 无重叠数据的时间窗口,设置方式举例:timeWindow(Time.seconds(5)) time-sliding-window 有重叠数据的时间窗口,设置方式举例:timeWindow(Time.seconds(5), Time.seconds(3)) count-tumbling-window无重叠数据的数量窗口,设置方式举例:countWindow(5) count-sliding-window 有重叠数据的数量窗口,设置方式举例:countWindow(...
相关文章
文章评论
共有0条评论来说两句吧...