import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.api.java.function.VoidFunction; import scala.Tuple2; import java.util.Arrays; import java.util.Iterator; /** * @Author yqq * @Date 2021/12/06 23:16 * @Version 1.0 */ /** * SparkScala api 与Java api 不同: * 1).java中需要创建 JavaSparkContext * 2).scala中是RDD ,java中是JavaRDD * 3).scala中将RDD转换成K,V格式的数据直接使用 map转出tuple数据即可 * Java中将RDD转换成K,V格式的数据需要使用mapToPair,转出K,V格式的数据 * * 4).JavaPairRDD 在java中代表的是K,V格式的RDD * */ public class WordCountByJava { public static void main(String[] args) { SparkConf conf = new SparkConf(); conf.setMaster("local"); conf.setAppName("word_count_java"); JavaSparkContext context = new JavaSparkContext(conf); JavaRDD<String> lines = context.textFile("data/words"); JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.asList(s.split(" ")).iterator(); } }); JavaPairRDD<String, Integer> pairRDD = words.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { return new Tuple2<>(s, 1); } }); JavaPairRDD<String, Integer> result = pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer integer, Integer integer2) throws Exception { return integer + integer2; } }); result.foreach(new VoidFunction<Tuple2<String, Integer>>() { @Override public void call(Tuple2<String, Integer> tuple2) throws Exception { System.out.println(tuple2); } }); } }
。。。。。很明显Java写的很复杂。。。。。。还是Scala真香