學習筆記:從0開始學習大資料-14. java spark程式設計實踐
阿新 • • 發佈：2018-12-05
上節搭建好了eclipse spark程式設計環境
在測試執行scala 或java 編寫spark程式 ，在eclipse平臺都可以執行，但打包匯出jar，提交 spark-submit執行，都不能執行，最後確定是版本問題，就是你在eclipse除錯的spark版本需和spark-submit 提交spark的執行版本一致，還有就是scala版本一致，才能正常執行。
以下是java spark程式執行
1.新建maven專案 SparkApps
注意 pom.xml 中spark-core 的版本
我原來除錯使用的是
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>2.4.0</version>
</dependency>
打包成jar到提交 spark-submit 執行，總是提示錯誤，因為spark下載的是spark-1.6.0-cdh5.16.0版本的,與eclipse中spark2.4.0版本有些語句用法不一致。
2. 專案中新建類JavaWordCount
package com.linbin.SparkApps; import scala.Tuple2; 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 java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.regex.Pattern; public class JavaWordCount { private static final Pattern SPACE = Pattern.compile(" "); public static void main(String[] args) throws Exception { if (args.length < 1) { System.err.println("Usage: JavaWordCount <file>"); System.exit(1); } SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount"); // setMaster 在打包匯出時無需設定 sparkConf.setMaster("local[2]"); JavaSparkContext ctx = new JavaSparkContext(sparkConf); JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override /* 以下spark2.X * * public Iterator<String> call(String s) { * return (Arrays.asList(SPACE.split(s)).iterator(); * } */ // 以下spark1.X public Iterable<String> call(String s) throws Exception { return Arrays.asList(SPACE.split(s)); } }); JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) { return new Tuple2<String, Integer>(s, 1); } }); JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer i1, Integer i2) { return i1 + i2; } }); List<Tuple2<String, Integer>> output = counts.collect(); for (Tuple2<?,?> tuple : output) { System.out.println(tuple._1() + ": " + tuple._2()); } ctx.stop(); ctx.close(); } }
3. 在eclipse中執行 as “java Application”
正常輸出結果
4. Eclipse中打包匯出為 sparkapps.jar
5. 提交給spark中執行
[[email protected] bin]# ./spark-submit --master spark://centos7:7077 --class com.linbin.SparkApps.JavaWordCount /home/linbin/workspace/sparkapps.jar hdfs://centos7:8020/hello.txt
6. 執行結果，正常輸出
[ 7. 在瀏覽器可以看到作業記錄
