当前位置：首页 > news >正文

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

news 来源：原创 2024/9/22 1:42:48

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文件内容：

<project xmlns="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><scope>provided</scope></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><transformer implementation="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。

package web;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.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;import java.util.Arrays;
import java.util.Iterator;
import java.util.regex.Pattern;public class SparkApplication {private static final Pattern SPACE=Pattern.compile(" ");public static void main(String[] args) {SparkConf sparkConf=new SparkConf().setAppName("sparkBoot").setMaster("local");JavaSparkContext sparkContext=new JavaSparkContext(sparkConf);JavaRDD<String> lines=sparkContext.textFile("/user/input/").cache();lines.map(new Function<String, Object>() {@Overridepublic Object call(String s) {return s;}});JavaRDD<String> words=lines.flatMap(new FlatMapFunction<String, String>() {@Overridepublic Iterator<String> call(String s) throws Exception {return Arrays.asList(SPACE.split(s)).iterator();}});JavaPairRDD<String,Integer> wordsOnes=words.mapToPair(new PairFunction<String, String, Integer>() {@Overridepublic Tuple2<String, Integer> call(String s) throws Exception {return new Tuple2<String,Integer>(s,1);}});JavaPairRDD<String,Integer> wordsCounts=wordsOnes.reduceByKey(new Function2<Integer, Integer, Integer>() {@Overridepublic Integer call(Integer integer, Integer integer2) throws Exception {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中看到处理结果。
删除结果目录