文档首页> 数据湖探索 DLI> 开发指南> Spark Jar作业开发指南> 使用Spark Jar作业读取和查询OBS数据
更新时间:2024-01-09 GMT+08:00
分享

使用Spark Jar作业读取和查询OBS数据

操作场景

DLI完全兼容开源的Apache Spark,支持用户开发应用程序代码来进行作业数据的导入、查询以及分析处理。本示例从编写Spark程序代码读取和查询OBS数据、编译打包到提交Spark Jar作业等完整的操作步骤说明来帮助您在DLI上进行作业开发。

环境准备

在进行Spark Jar作业开发前,请准备以下开发环境。

表1 Spark Jar作业开发环境

准备项

说明

操作系统

Windows系统,支持Windows7以上版本。

安装JDK

JDK使用1.8版本。

安装和配置IntelliJ IDEA

IntelliJ IDEA为进行应用开发的工具,版本要求使用2019.1或其他兼容版本。

安装Maven

开发环境的基本配置。用于项目管理,贯穿软件开发生命周期。

开发流程

DLI进行Spark Jar作业开发流程参考如下:
图1 Spark Jar作业开发流程
表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”的队列,队列类型选择为“通用队列”。创建队列详细介绍请参考创建队列
    图2 创建队列
  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”。
    图3 新建Project
  2. 选择Maven,Project SDK选择1.8,单击“Next”。
    图4 新建Project
  3. 定义样例工程名和配置样例工程存储路径,单击“Finish”完成工程创建。
    图5 创建工程

    如上图所示,本示例创建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>
    图6 修改pom.xml文件
  5. 在工程路径的“src > main > java”文件夹上鼠标右键,选择“New > Package”,新建Package和类文件。
    图7 新建Package
    Package根据需要定义,本示例定义为:“com.huawei.dli.demo”,完成后回车。
    图8 输入包名
    在包路径下新建Java Class文件,本示例定义为:SparkDemoObs。
    图9 新建Java Class文件

步骤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”对代码进行打包。
    图10 编译打包
    打包成功后,生成的Jar包会放到target目录下,以备后用。本示例将会生成到:“D:\DLITest\SparkJarObs\target”下名为“SparkJarObs-1.0-SNAPSHOT.jar”。
    图11 导出jar包

步骤6:上传Jar包到OBS和DLI下

  1. 登录OBS控制台,将生成的“SparkJarObs-1.0-SNAPSHOT.jar”Jar包文件上传到OBS路径下。
  2. 将Jar包文件上传到DLI的程序包管理中,方便后续统一管理。
    1. 登录DLI管理控制台,单击“数据管理 > 程序包管理”。
    2. 在“程序包管理”页面,单击右上角的“创建”创建程序包。
    3. 在“创建程序包”对话框,配置以下参数。
      1. 包类型:选择“JAR”。
      2. OBS路径:程序包所在的OBS路径。
      3. 分组设置和组名称根据情况选择设置,方便后续识别和管理程序包。
    4. 单击“确定”,完成创建程序包。
      图12 创建程序包

步骤7:创建Spark Jar作业

  1. 登录DLI控制台,单击“作业管理 > Spark作业”。
  2. 在“Spark作业”管理界面,单击“创建作业”。
  3. 在作业创建界面,配置对应作业运行参数。具体说明如下:
    • 所属队列:选择已创建的DLI通用队列。例如当前选择步骤1:创建DLI通用队列创建的通用队列“sparktest”。
    • 在下拉列表中选择支持的Spark版本,推荐使用最新版本。
    • 作业名称(--name):自定义Spark Jar作业运行的名称。当前定义为:SparkTestObs。
    • 应用程序:选择步骤6:上传Jar包到OBS和DLI下中上传到DLI程序包。例如当前选择为:“SparkJarObs-1.0-SNAPSHOT.jar”。
    • 主类:格式为:程序包名+类名。例如当前为:com.huawei.dli.demo.SparkDemoObs。

    其他参数可暂不选择。

    了解更多Spark Jar作业提交说明可以参考创建Spark作业

    图13 创建Spark Jar作业
  4. 单击“执行”,提交该Spark Jar作业。在Spark作业管理界面显示已提交的作业运行状态。
    图14 作业运行状态

步骤8:查看作业运行结果

  1. 在Spark作业管理界面显示已提交的作业运行状态。初始状态显示为“启动中”。
  2. 如果作业运行成功则作业状态显示为“已成功”,单击“操作”列“更多”下的“Driver日志”,显示当前作业运行的日志。
    图15 diver日志
    图16 “Driver日志”中的作业执行日志
  3. 如果作业运行成功,本示例进入OBS桶下的“result/parquet”目录,查看已生成预期的parquet文件。
    图17 obs桶文件
  4. 如果作业运行失败,单击“操作”列“更多”下的“Driver日志”,显示具体的报错日志信息,根据报错信息定位问题原因。
    例如,如下截图信息因为创建Spark Jar作业时主类名没有包含包路径,报找不到类名“SparkDemoObs”。
    图18 报错信息

    可以在“操作”列,单击“编辑”,修改“主类”参数为正确的:com.huawei.dli.demo.SparkDemoObs,单击“执行”重新运行该作业即可。

后续指引

完整样例代码参考

认证用的access.key和secret.key硬编码到代码中或者明文存储都有很大的安全风险,建议在配置文件或者环境变量中密文存放,使用时解密,确保安全。

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|
        // +----+-------+

        // Select only the "name" column
        df.select("name").show();
        // +-------+
        // |   name|
        // +-------+
        // |Michael|
        // |   Andy|
        // | Justin|
        // +-------+

        // Select people older than 21
        df.filter(col("age").gt(21)).show();
        // +---+----+
        // |age|name|
        // +---+----+
        // | 30|Andy|
        // +---+----+

        // Count people by age
        df.groupBy("age").count().show();
        // +----+-----+
        // | age|count|
        // +----+-----+
        // |  19|    1|
        // |null|    1|
        // |  30|    1|
        // +----+-----+

        // Register the DataFrame as a SQL temporary view
        df.createOrReplaceTempView("people");

        Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
        sqlDF.show();
        // +----+-------+
        // | age|   name|
        // +----+-------+
        // |null|Michael|
        // |  30|   Andy|
        // |  19| Justin|
        // +----+-------+

        sqlDF.write().mode(SaveMode.Overwrite).parquet("obs://dli-test-obs01/result/parquet");
        spark.read().parquet("obs://dli-test-obs01/result/parquet").show();

        spark.stop();
    }
}

分享:

    相关文档

    相关产品