内容简介
本篇文章主要讲述了Springboot项目和Spark相结合的方法。
在Spark的学习过程中，我们大多数提交作业的方式都是使用Spark的命令来提交jar包并运行，但是本次相反，我们将使用Springboot来开发Spark的相关计算任务代码，开发完成后将项目打包部署到服务器上，项目启动以后会生成一个SparkSession的长服务，各种请求可通过http来调用项目中相关的接口，再来调用Spark来进行计算。
环境准备
安装好Spark环境，单节点、集群都可以，如果是集群，则要保证节点之间的端口通信正常。
Spark版本要和代码中的Spark版本相同。
如果使用集群，则需要一个Nginx，做反向代理，用来分发项目包到各个节点。
server {
listen 30443 default_server;
server_name _;
location /jar/ {
alias /opt/jar/; // 放项目包的目录
}
location / {
return 502;
}
}
代码实现
POM文件只放打包相关的，依赖的包各不相同。
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>2.3.7.RELEASE</version>
<configuration>
<includeSystemScope>true</includeSystemScope>
<mainClass>com.xxx.xxx.TestApplication</mainClass>
</configuration>
<executions>
<execution>
<id>repackage</id>
<goals>
<goal>repackage</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
创建SparkSession长服务实现代码：
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.UDFRegistration;
import org.apache.spark.sql.types.StructType;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.Map;
@Service
@Slf4j
public class SparkService implements DisposableBean, InitializingBean {
private SparkSession session = null;
public SparkSession session() {
return session;
}
@Override
public void afterPropertiesSet() {
try {
log.info("正在创建Spark Session......");
SparkConf config = new SparkConf();
config.setAppName(getClass().getSimpleName());
config.setMaster("spark://bigdata1:7077");
config.setJars(new String[]{ "http://bigdata1:30543/jar/Test-1.0-SNAPSHOT.jar.original" });
config.set("spark.executor.cores", "2"); // 每个executor占用的核心数
config.set("spark.driver.memory", "2g");
config.set("spark.driver.cores", "2");
config.set("spark.executor.memory", "2g");
config.set("spark.default.parallelism", "8"); // 任务并行度
session = SparkSession.builder().config(config).getOrCreate();
} catch (Exception e) {
log.error("Spark Session创建失败！！！");
e.printStackTrace();
}
}
@Override
public void destroy() {
FileUtil.close(session);
}
// 注册自定义UDF
public SparkSession udf(SparkSession session) {
UDFRegistration udf = session.udf();
udf.register(ToCharUDF.NAME, new ToCharUDF(), ToCharUDF.TYPE);
udf.register(AgeUDF.NAME, new AgeUDF(), AgeUDF.TYPE);
udf.register(NullIfUDF.NAME,new NullIfUDF(), NullIfUDF.TYPE);
udf.register(DatePartEpochUDF.NAME,new DatePartEpochUDF(), DatePartEpochUDF.TYPE);
udf.register(ToNumberUDF.NAME,new ToNumberUDF(), ToNumberUDF.TYPE);
return session;
}
}
当需要用到Spark时先将上面的类注入：
@Autowired
public SparkService sparkService;
再创建一个新的session：
SparkSession session = sparkService.session().newSession(); // 没有注册自定义UDF函数 SparkSession sparkSession = sparkService.udf(session); // 注册没有自定义UDF函数
部署运行
再说项目打包相关事宜，一共会打两个项目包，一个是以“.jar”结尾，另一个是以“.jar.original”结尾。
还要导出所有项目相关依赖的jar包放到Spark /jars目录下，使用下面命令即可：
mvn dependency:copy-dependencies
但是需要注意，自己导出来的包可能会和Spark的jars目录下的包有冲突，此时要以项目里的依赖版本为准，如果删除Spark中的冲突依赖之后导致Spark服务无法启动，则需要调整项目中引用的依赖版本向Spark服务靠齐。
做完以上几步之后就可以启动spark服务，启动项目（命令：java -jar xxx）即可。
标签：session,集群,org,spark,import,Spark,config,远程 From： https://www.cnblogs.com/chuijingjing/p/16995280.html