编写并运行第一个spark java程序

spark版本：3.5.2
第一部分：
1、进入spark，尝试spark shell编程
spark-shell 进入模式 ctrl+D退出
2、shell测试代码
//此处是hadoop的路径
var lines=sc.textFile(“/user/input/1.txt”)
lines.count() //输出count行数
lines.first() //输出RDD的第一行元素 1.txt的第一行数据
val pythonLines = lines.filter(line => line.contains(“Python”)) //lambda语法糖
原出处：

原博客上的注释：
上一篇搭建Hadoop单机环境运行MapReduce后本来是计划搭建完全分布式环境的但是公司内部开会讨论后,考虑效率以及业务场景,决定不使用MapReduce做批计算原因是太慢…而且后期需要集成其他流处理框架,建模才能满足不同需求现决定采用spark技术栈做批处理和流处理,底层还是Hadoop的HDFS分布式文件系统Elasticsearch实时索引,logstash清洗数据,kafka队列

原文链接：https://blog.csdn.net/youbitch1/article/details/88355111

第二部分：
参考下文，写第二个例子
https://blog.csdn.net/youbitch1/article/details/88421965
1、创建maven工程。名称是spark01
2、使用hadoop的hdfs存储及处理文件。
3、pom文件内容：

<projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>spark01</artifactId><version>1.0-SNAPSHOT</version><name>Archetype - spark01</name><url>http://maven.apache.org</url><properties><spark.version>3.5.2</spark.version></properties><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.13</artifactId><version>${spark.version}</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>3.2.4</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><transformers><transformerimplementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>web.SparkApplication</mainClass></transformer></transformers></configuration></execution></executions></plugin></plugins></build></project>

4、创建SparkApplication
首先处理下工程的兼容性：
Project Structure中，Project/Modules中4处的sdk 都选择为1.8。
Setting的Build，Execution，Deployment中的Compiler下的Java Compiler下，bytecode版本两处都改为8。

packageweb;importorg.apache.spark.SparkConf;importorg.apache.spark.api.java.JavaPairRDD;importorg.apache.spark.api.java.JavaRDD;importorg.apache.spark.api.java.JavaSparkContext;importorg.apache.spark.api.java.function.FlatMapFunction;importorg.apache.spark.api.java.function.Function;importorg.apache.spark.api.java.function.Function2;importorg.apache.spark.api.java.function.PairFunction;importscala.Tuple2;importjava.util.Arrays;importjava.util.Iterator;importjava.util.regex.Pattern;publicclassSparkApplication{privatestaticfinalPatternSPACE=Pattern.compile(" ");publicstaticvoidmain(String[] args){SparkConf sparkConf=newSparkConf().setAppName("sparkBoot").setMaster("local");JavaSparkContext sparkContext=newJavaSparkContext(sparkConf);JavaRDD<String> lines=sparkContext.textFile("/user/input/").cache();

        lines.map(newFunction<String,Object>(){@OverridepublicObjectcall(String s){return s;}});JavaRDD<String> words=lines.flatMap(newFlatMapFunction<String,String>(){@OverridepublicIterator<String>call(String s)throwsException{returnArrays.asList(SPACE.split(s)).iterator();}});JavaPairRDD<String,Integer> wordsOnes=words.mapToPair(newPairFunction<String,String,Integer>(){@OverridepublicTuple2<String,Integer>call(String s)throwsException{returnnewTuple2<String,Integer>(s,1);}});JavaPairRDD<String,Integer> wordsCounts=wordsOnes.reduceByKey(newFunction2<Integer,Integer,Integer>(){@OverridepublicIntegercall(Integer integer,Integer integer2)throwsException{return integer+integer2;}});

        wordsCounts.saveAsTextFile("/user/local/data1");}}

5、编译打包生成spark01-1.0-SNAPSHOT.jar
6、上传到spark的服务器
然后执行命令：
spark-submit --master local[3] --executor-memory 512m --class web.SparkApplication spark01-1.0-SNAPSHOT.jar
7、即可以在HDFS中看到处理结果。
删除结果目录

hdfs dfs -rm-R /user/local/data1

重跑后即可看到结果。

hdfs dfs -cat /user/local/data1/*