Spark 离线开发框架设计与实现

2.3.2 Java 应用

当 SQL 无法满足复杂数据处理需求时，支持 Java 程序编写。开发者需创建新类继承 Application 父类并实现 run 方法。

在 run 方法中，你只需关注数据处理逻辑。通用的 SparkSession 创建及关闭、输入输出逻辑均由框架封装。框架支持 HDFS 文件输入、SQL 输入等多种类型，调用相关处理函数即可。

以下是一个简单的 Java 数据处理应用示例。从配置可以看出，仍需配置资源大小，但相比 SQL 应用，需要定制化编写对应的 Java 类，并指定输入（input）和输出（output）参数。此例中输入为 SQL 代码，输出为 HDFS 文件。

<?xml version="1.0" encoding="UTF-8"?>
<project name="ecommerce_dwd_hanwuji_click_incr_day_domain">
    <class>com.way.app.instance.ecommerce.Test</class>
    <input>
        <type>table</type>
        <sql>select clk_url, clk_num from test_table where event_day='{DATE}' and click_pv > 0 and is_ubs_spam=0</sql>
    </input>
    <output>
        <type>afs_kp</type>
        <path>test/event_day={DATE}</path>
    </output>
    <conf>
        <spark.executor.memory>2G</spark.executor.memory>
        <spark.executor.cores>2</spark.executor.cores>
        <spark.driver.memory>2G</spark.driver.memory>
        <spark.executor.instances>10</spark.executor.instances>
    </conf>
</project>

对应 Java 实现类如下，开发者只需三步走：获取输入数据、逻辑处理、结果输出。

package com.way.app.instance.ecommerce;

import com.way.app.Application;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.api.java.function.FilterFunction;

import java.util.Map;

public class Test extends Application {
    @Override
    public void run() {
        // 输入
        Map<String, String> input = (Map<String, String>) property.get("input");
        Dataset<Row> ds = sparkSession.sql(getInput(input)).toDF("url", "num");

        // 逻辑处理（筛选特定站点日志）
        JavaRDD<String> outRdd = ds.filter((FilterFunction<Row>) row -> {
            String url = row.getAs("url").toString();
            return url.contains(".jd.com") || url.contains(".suning.com") || 
                   url.contains("pin.suning.com") || url.contains(".taobao.com") || 
                   url.contains("detail.tmall.hk") || url.contains(".amazon.cn") || 
                   url.contains(".kongfz.com") || url.contains(".gome.com.cn") || 
                   url.contains(".kaola.com") || url.contains(".dangdang.com") || 
                   url.contains("aisite.wejianzhan.com") || url.contains("w.weipaitang.com");
        }).toJavaRDD().map(row -> row.mkString("\001"));

        // 输出
        Map<String, String> output = (Map<String, String>) property.get("output");
        outRdd.saveAsTextFile(getOutPut(output));
    }
}

2.3.3 数据回溯应用

数据回溯应用是为了解决快速回溯、释放人力而研发的。开发者无需重构任务代码，与 SQL 应用相同，回溯应用类为唯一类，支持多种回溯方案。

方案设计

日常回溯中发现，单次任务存在严重的时间浪费。无论何种提交方式，都需经历执行环境申请及准备过程：Client 向 RS 申请启动 ApplicationMaster，AM 向 RS 申请 Executor 资源，注册 Task 等。这个过程统称为'执行环境准备'，通常耗时 5-20 分钟，失败率约 10%。

整个流程分为三个阶段：执行环境准备、开始执行代码、释放资源。

离线开发框架回溯应用从节省时间和人力两方面考虑，设计方案如下：

（注：此处展示回溯时间优化方案图）

时间节省： 将所有回溯子任务的第一、第三步（环境准备及释放）压缩为一次。若回溯 30 个子任务，可节省 5-20 分钟 × 29。子任务越多，提效越明显。

人力节省： 开发者无需额外开发或添加回溯配置。框架启动的任务数量远小于传统方案，只需关注一个任务的执行状态，而非维护 N 个任务。

实测数据显示，回溯一年的串行任务，不使用框架在最理想情况下需 2.5 天；使用框架在最坏情况下仅需 6 小时，提效 90% 以上，且基本无需人力关注。

功能介绍

断点续回 Spark 计算虽快，但节点宕机、网络波动导致任务失败的情况难免。回溯任务动辄数月，失败后从断点处继续回溯至关重要。框架记录了已成功部分，重启后自动续回。

回溯顺序 根据业务需求确定顺序。依赖历史数据的增量数据通常从历史到现在；无依赖需求时，从现在到历史可更快满足指标了解。框架支持灵活配置。

并行回溯 回溯任务优先级通常低于例行任务，资源有限时需控制并发。框架默认为串行回溯，但也支持一定程度的并行度，开发者可根据资源情况调整。

创建一个回溯任务

回溯应用使用非常方便，无需新开发代码，配置回溯方案即可。

• class：回溯应用唯一类，必填。
• type：应用类型，默认 sql，Java 应用填类名。
• path：回溯代码路径，必填。
• limitdate：截止日期，必填。
• startdate：开始日期，必填（断点续回或并行时可能无效）。
• order：回溯顺序，默认倒序（-1），正序为 1。
• distance：步长，用于并行回溯。
• file：日志文件，记录成功日期，失败重启时以此为准。
• conf：资源占用配置。

<?xml version="1.0" encoding="UTF-8"?>
<project name="ecommerce_ads_others_order_retain_incr_day">
    <class>com.way.app.instance.ecommerce.Huisu</class>
    <type>sql</type>
    <path>/sql/ecommerce/ecommerce_ads_others_order_retain_incr_day.sql</path>
    <limitdate>20220404</limitdate>
    <startdate>20210101</startdate>
    <order>1</order>
    <distance>-1</distance>
    <file>/user/ecommerce_ads_others_order_retain_incr_day/process</file>
    <conf>
        <spark.executor.memory>1G</spark.executor.memory>
        <spark.executor.cores>2</spark.executor.cores>
        <spark.executor.instances>30</spark.executor.instances>
        <spark.yarn.maxAppAttempts>1</spark.yarn.maxAppAttempts>
    </conf>
</project>

三、使用方式

3.1 使用介绍

使用离线框架开发时，开发者只需重点关注数据逻辑处理。打包提交后，每个应用主类相同（即 Application 父类），唯一变化的是接收的配置文件的相对路径。

3.2 使用对比

使用离线框架前后对比效果显著，开发效率大幅提升。

（注：此处展示使用前后对比图）

四、展望

目前框架仅支持 SQL 和 Java 语言。未来计划升级支持更多 Spark 原生语言，让开发者更方便、快速地进行大数据开发。