SparkJavaAPI:join的使用

将一组数据转化为RDD后，分别创造出两个PairRDD，然后再对两个PairRDD进行归约（即合并相同Key对应的Value），过程如下图所示：

代码实现如下：

public class SparkRDDDemo {
public static void main(String[] args){
SparkConf conf = new SparkConf().setAppName("SparkRDD").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
List<Integer> data = Arrays.asList(1,2,3,4,5);
JavaRDD<Integer> rdd = sc.parallelize(data);
//FirstRDD
JavaPairRDD<Integer, Integer> firstRDD = rdd.mapToPair(new PairFunction<Integer, Integer, Integer>() {
@Override
public Tuple2<Integer, Integer> call(Integer num) throws Exception {
return new Tuple2<>(num, num * num);
}
});
//SecondRDD
JavaPairRDD<Integer, String> secondRDD = rdd.mapToPair(new PairFunction<Integer, Integer, String>() {
@Override
public Tuple2<Integer, String> call(Integer num) throws Exception {
return new Tuple2<>(num, String.valueOf((char)(64 + num * num)));
}
});
JavaPairRDD<Integer, Tuple2<Integer, String>> joinRDD = firstRDD.join(secondRDD);
JavaRDD<String> res = joinRDD.map(new Function<Tuple2<Integer, Tuple2<Integer, String>>, String>() {
@Override
public String call(Tuple2<Integer, Tuple2<Integer, String>> integerTuple2Tuple2) throws Exception {
int key = integerTuple2Tuple2._1();
int value1 = integerTuple2Tuple2._2()._1();
String value2 = integerTuple2Tuple2._2()._2();
return "<" + key + ",<" + value1 + "," + value2 + ">>";
}
});
List<String> resList = res.collect();
for(String str : resList)
System.out.println(str);
sc.stop();
}
}

SparkJavaAPI:join的使用相关推荐

ajax请求get方法的封装,使用jQuery中Ajax的封装函数——$.get()
MATLAB中图片格式与视频格式的转换参考:https://blog.csdn.net/iracer/article/details/48876473 Matlab-将绘图保存为视频VideoWri ...
浅显易懂 Makefile 入门（06）— 文件名操作函数（dir、notdir、suffix、basename、addsuffix、addperfix、join、wildcard）
编写 Makefile 的时候,很多情况下需要对文件名进行操作.例如获取文件的路径,去除文件的路径,取出文件前缀或后缀等等. 注意:下面的每个函数的参数字符串都会被当作或是一个系列的文件名来看待. 1 ...
python路径拼接os.path.join()函数的用法
os.path.join()函数:连接两个或更多的路径名组件 1.如果各组件名首字母不包含'/',则函数会自动加上 2.如果有一个组件是一个绝对路径,则在它之前的所有组件均会被舍弃 3.如果最后一个组 ...
Python join()方法
描述 Python join() 方法用于将序列中的元素以指定的字符连接生成一个新的字符串. 语法 join()方法语法: str.join(sequence).join(sequence) 参数 s ...
SQL性能下降原因及Join查询
原因查询语句写得烂索引失效关联查询过多(设计缺陷) 服务器调优及各个参数设置(缓冲.线程数等) Join查询的SQL执行顺序手写 SELECT <select_list> FROM ...
2021年大数据Flink（四十五）：扩展阅读双流Join
目录扩展阅读双流Join 介绍 Window Join Interval Join 代码演示1 代码演示2 重点注意扩展阅读双流Join 介绍 https:// ...
python join 的使用
join 方法解释 join : 可以以指定的方法来连接字符串 join 的使用 list_a = ["Pythons", "is", "a" ...
Java fork join ForkJoinPool 用法例子
本例是把一个大的数组求和的计算的大任务分解到在小范围内求和的小任务,然后把这些小任务之和加起来就是所求之结果. 技术:JDK8.0, Java fork-join模式下的RecursiveTask技术 ...
python join split
本文记录python,join和split函数的用法. 参考 http://blog.csdn.net/doiido/article/details/43538833 http://blog.csdn ...

SparkJavaAPI:join的使用

SparkJavaAPI:join的使用相关推荐

最新文章

热门文章