操作场景
DLI完全兼容开源的Apache Spark，支持用户开发应用程序代码来进行作业数据的导 入、查询以及分析处理。本示例从编写Spark程序代码读取和查询OBS数据、编译打包 到提交Spark Jar作业等完整的操作步骤说明来帮助您在DLI上进行作业开发。
环境准备
在进行Spark Jar作业开发前，请准备以下开发环境。
表2-1 Spark Jar 作业开发环境
准备项 说明
操作系统 Windows系统，支持Windows7以上版本。
安装JDK JDK使用1.8版本。
安装和配置IntelliJ
IDEA IntelliJ IDEA为进行应用开发的工具，版本要求使用2019.1 或其他兼容版本。
安装Maven 开发环境的基本配置。用于项目管理，贯穿软件开发生命周 期。
开发流程
DLI进行Spark Jar作业开发流程参考如下：
图2-1 Spark Jar 作业开发流程
表2-2 开发流程说明 序
号
阶段 操作界
面
说明
1 创建DLI通用队列 DLI控
制台 创建作业运行的DLI队列。
2 上传数据到OBS
桶 OBS控
制台 将测试数据上传到OBS桶下。
3 新建Maven工 程，配置pom文 件
IntelliJ
IDEA 参考样例代码说明，编写程序代码读取OBS数 据。
4 编写程序代码 5 调试，编译代码
并导出Jar包 6 上传Jar包到OBS
和DLI OBS控
制台
将生成的Spark Jar包文件上传到OBS目录下和 DLI程序包中。
7 创建Spark Jar作 业
DLI控 制台
在DLI控制台创建Spark Jar作业并提交运行作 业。
8 查看作业运行结 果
DLI控 制台
查看作业运行状态和作业运行日志。
步骤 1：创建 DLI 通用队列
第一次提交Spark作业，需要先创建队列，例如创建名为“sparktest”的队列，队列类 型选择为“通用队列”。
1. 在DLI管理控制台的左侧导航栏中，选择“队列管理”。
2. 单击“队列管理”页面右上角“购买队列”进行创建队列。
3. 创建名为“sparktest”的队列，队列类型选择为“通用队列”。创建队列详细介 绍请参考创建队列。
4. 单击“立即购买”，确认配置。
5. 配置确认无误，单击“提交”完成队列创建。
步骤 2：上传数据到 OBS 桶
1. 根据如下数据，创建people.json文件。
{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}
2. 进入OBS管理控制台，在“桶列表”下，单击已创建的OBS桶名称，本示例桶名 为“dli-test-obs01”，进入“概览”页面。
3. 单击左侧列表中的“对象”，选择“上传对象”，将people.json文件上传到OBS 桶根目录下。
4. 在OBS桶根目录下，单击“新建文件夹”，创建名为“result”的文件夹。
5. 单击“result”的文件夹，在“result”下单击“新建文件夹”，创建名为
“parquet”的文件夹。
步骤 3：新建 Maven 工程，配置 pom 依赖
以下通过IntelliJ IDEA 2020.2工具操作演示。
1. 打开IntelliJ IDEA，选择“File > New > Project”。
图2-2 新建 Project
2. 选择Maven，Project SDK选择1.8，单击“Next”。
3. 定义样例工程名和配置样例工程存储路径，单击“Finish”完成工程创建。
如上图所示，本示例创建Maven工程名为：SparkJarObs，Maven工程路径为：
“D:\DLITest\SparkJarObs”。
4. 在pom.xml文件中添加如下配置。
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.3.2</version>
</dependency>
</dependencies>
图2-3 修改 pom.xml 文件
5. 在工程路径的“src > main > java”文件夹上鼠标右键，选择“New >
Package”，新建Package和类文件。
Package根据需要定义，本示例定义为：“com.huawei.dli.demo”，完成后回 车。
在包路径下新建Java Class文件，本示例定义为：SparkDemoObs。
步骤 4：编写代码
编写SparkDemoObs程序读取OBS桶下的1的“people.json”文件，并创建和查询临时 表“people”。
完整的样例请参考完整样例代码参考，样例代码分段说明如下：
1. 导入依赖的包。
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.col;
2. 通过当前帐号的AK和SK创建SparkSession会话spark 。
SparkSession spark = SparkSession .builder()
.config("spark.hadoop.fs.obs.access.key", "xxx") .config("spark.hadoop.fs.obs.secret.key", "yyy") .appName("java_spark_demo")
.getOrCreate();
– "spark.hadoop.fs.obs.access.key"参数对应的值"xxx"需要替换为帐号的AK 值。
– "spark.hadoop.fs.obs.secret.key"参数对应的值“yyy”需要替换为帐号的SK 值。
AK和SK值获取请参考：如何获取AK和SK。
3. 读取OBS桶中的“people.json”文件数据。
其中“dli-test-obs01”为演示的OBS桶名，请根据实际的OBS桶名替换。
Dataset<Row> df = spark.read().json("obs://dli-test-obs01/people.json");
df.printSchema();
4. 通过创建临时表“people”读取文件数据。
df.createOrReplaceTempView("people");
5. 查询表“people”数据。
Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
sqlDF.show();
6. 将表“people”数据以parquet格式输出到OBS桶的“result/parquet”目录下。
sqlDF.write().mode(SaveMode.Overwrite).parquet("obs://dli-test-obs01/result/parquet");
spark.read().parquet("obs://dli-test-obs01/result/parquet").show();
7. 关闭SparkSession会话spark。
spark.stop();
步骤 5：调试、编译代码并导出 Jar 包
1. 单击IntelliJ IDEA工具右侧的“Maven”，参考下图分别单击“clean”、
“compile”对代码进行编译。
编译成功后，单击“package”对代码进行打包。
打包成功后，生成的Jar包会放到target目录下，以备后用。本示例将会生成到：
“D:\DLITest\SparkJarObs\target”下名为“SparkJarObs-1.0-SNAPSHOT.jar”。
步骤 6：上传 Jar 包到 OBS 和 DLI 下
1. 登录OBS控制台，将生成的“SparkJarObs-1.0-SNAPSHOT.jar”Jar包文件上传到 OBS路径下。
2. 将Jar包文件上传到DLI的程序包管理中，方便后续统一管理。
a. 登录DLI管理控制台，单击“数据管理 > 程序包管理”。
b. 在“程序包管理”页面，单击右上角的“创建”创建程序包。
c. 在“创建程序包”对话框，配置以下参数。
i. 包类型：选择“JAR”。
ii. OBS路径：程序包所在的OBS路径。
iii. 分组设置和组名称根据情况选择设置，方便后续识别和管理程序包。
d. 单击“确定”，完成创建程序包。
步骤 7：创建 Spark Jar 作业
1. 登录DLI控制台，单击“作业管理 > Spark作业”。
2. 在“Spark作业”管理界面，单击“创建作业”。
3. 在作业创建界面，配置对应作业运行参数。具体说明如下：
– 所属队列：选择已创建的DLI通用队列。例如当前选择步骤1：创建DLI通用队 列创建的通用队列“sparktest”。
– 作业名称（--name）：自定义Spark Jar作业运行的名称。当前定义为：
SparkTestObs。
– 应用程序：选择步骤6：上传Jar包到OBS和DLI下中上传到DLI程序包。例如 当前选择为：“SparkJarObs-1.0-SNAPSHOT.jar”。
– 主类：格式为：程序包名+类名。例如当前为：
com.huawei.dli.demo.SparkDemoObs。
其他参数可暂不选择，想了解更多Spark Jar作业提交说明可以参考创建Spark作 业。
图2-4 创建 Spark Jar 作业
4. 单击“执行”，提交该Spark Jar作业。在Spark作业管理界面显示已提交的作业运 行状态。
步骤 8：查看作业运行结果
1. 在Spark作业管理界面显示已提交的作业运行状态。初始状态显示为“启动中”。
2. 如果作业运行成功则作业状态显示为“已成功”，单击“操作”列“更多”下的
“Driver日志”，显示当前作业运行的日志。
图2-5 “Driver 日志”中的作业执行日志
3. 如果作业运行成功，本示例进入OBS桶下的“result/parquet”目录，查看已生成 预期的parquet文件。
4. 如果作业运行失败，单击“操作”列“更多”下的“Driver日志”，显示具体的报 错日志信息，根据报错信息定位问题原因。
例如，如下截图信息因为创建Spark Jar作业时主类名没有包含包路径，报找不到 类名“SparkDemoObs”。
可以在“操作”列，单击“编辑”，修改“主类”参数为正确的：
com.huawei.dli.demo.SparkDemoObs，单击“执行”重新运行该作业即可。
后续指引
● 如果您想通过Spark Jar作业访问其他数据源，请参考《使用Spark作业跨源访问 数据源》。
● 如果您想通过Spark Jar作业在DLI创建数据库和表，请参考《使用Spark作业访问 DLI元数据》。
完整样例代码参考
package com.huawei.dli.demo;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.col;
public class SparkDemoObs {
public static void main(String[] args) { SparkSession spark = SparkSession .builder()
.config("spark.hadoop.fs.obs.access.key", "xxx") .config("spark.hadoop.fs.obs.secret.key", "yyy") .appName("java_spark_demo")
.getOrCreate();
// can also be used --conf to set the ak sk when submit the app // test json data:
// {"name":"Michael"}
// {"name":"Andy", "age":30}
// {"name":"Justin", "age":19}
Dataset<Row> df = spark.read().json("obs://dli-test-obs01/people.json");
df.printSchema();
// root
// |-- age: long (nullable = true) // |-- name: string (nullable = true)
// Displays the content of the DataFrame to stdout df.show();
// +----+---+
// | age| name|
// +----+---+
// |null|Michael|
// | 30| Andy|
// | 19| Justin| df.createOrReplaceTempView("people");
Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
sqlDF.show();
sqlDF.write().mode(SaveMode.Overwrite).parquet("obs://dli-test-obs01/result/parquet");
spark.read().parquet("obs://dli-test-obs01/result/parquet").show();
spark.stop();
} }