导读：本文介绍了开发框架的整体设计，随后对各模块进行了拆解，重点介绍了如何疾速实现应用程序的开发，并从设计思路、实现形式、性能介绍及创立形式等角度对通用的数据回溯利用进行了全面介绍，实现了一次环境筹备，多数据回溯工作的启动计划。总之，框架对开发效率、回溯工作的效率与保护老本及代码治理便捷性都会有显著的成果。

全文3308字，预计浏览工夫10分钟。

一、背景

随着 Spark 以及其社区的一直倒退，Spark自身技术也在一直成熟，Spark在技术架构和性能上的劣势越来越显著，目前大多数公司在大数据处理中都偏向应用Spark。Spark反对多种语言的开发，如Scala、Java、Sql、Python等。

Spark SQL应用规范的数据连贯，与Hive兼容，易与其它语言API整合，表达清晰、简略易上手、学习成本低，是开发者开发简略数据处理的首选语言，但对于简单的数据处理、数据分析的开发，应用SQL开发显得力不从心，保护老本也十分高，应用高级语言解决会更高效。

在日常的数据仓库开发工作中，咱们除了开发工作外，也波及大量的数据回溯工作。对于创新型业务来说，口径变动频繁、业务迅速迭代，数据仓库的回溯十分常见，通过回溯几个月甚至一年是十分广泛的，但传统的回溯工作形式效率极低，而且须要人力亲密关注各工作状态。

针对目前现状，咱们开发了一套Spark离线开发框架，如下表所示，咱们例举了目前存在的问题及解决方案。框架的实现不仅让开发变得简略高效，而且对于数据的回溯工作在不须要任何开发的状况下，疾速高效地实现大量的回溯工作。

二、框架设计

框架旨在封装反复的工作，让开发变得简略。框架如图2-1所示，次要分为三个局部，根底框架、可扩大工具及应用程序，开发者只需关注应用程序即可简略疾速实现代码开发。

2.1 根底框架

根底框架中，咱们对于所有类型的利用实现代码与配置拆散机制，资源配置对立以XML文件模式保留并由框架解析解决。框架会依据开发者配置的工作应用资源大小，实现了SparkSession、SparkContext、SparkConf的创立，同时加载了罕用环境变量，开发了通用的UDF函数（如罕用的url参数解析等）。其中Application为所有利用的父类，解决流程如图所示，开发者只需编写关注绿色局部即可。

目前，离线框架所反对的罕用环境变量如下表所示。

2.2 可扩大工具

可扩大工具中蕴含了大量的工具类，服务于应用程序及根底框架，罕用有，配置文件解析类，如解析工作资源参数等；数据库工具类，用于读写数据库；日期工具类，用于日期加减、转换、辨认并解析环境变量等。服务于应用程序的通用工具模块可统称为可扩大工具，这里不再赘述。

2.3 应用程序

2.3.1 SQL利用

对于SQL利用，只须要创立SQL代码及资源配置即可，利用类为惟一类（已实现），有且只有一个，供所有SQL利用应用，开发者无需关怀。如下配置所示，class为所有利用的惟一类名，开发者要关怀的是path中的sql代码及conf中该sql所应用的资源大小。

<?xml version="1.0" encoding="UTF-8"?><project name="test">    <class>com.way.app.instance.SqlExecutor</class>    <path>sql文件门路</path>  <!--    sparksession conf   -->    <conf>        <spark.executor.memory>1G</spark.executor.memory>        <spark.executor.cores>2</spark.executor.cores>        <spark.driver.memory>1G</spark.driver.memory>        <spark.executor.instances>20</spark.executor.instances>    </conf></project>

2.3.2 Java利用

对于简单的数据处理，SQL代码不能满足需要时，咱们也反对Java程序的编写，与SQL不同的是，开发者须要创立新的利用类，继承Application父类并实现run办法即可，run办法中开发者只须要关注数据的解决逻辑，对于通用的SparkSession、SparkContext等创立及敞开无需关注，框架还帮忙开发者封装了代码的输出、输入逻辑，对于输出类型，框架反对HDFS文件输出、SQL输出等多种输出类型，开发者只需调用相干处理函数即可。

如下为一个简略的Java数据处理利用，从配置文件能够看出，仍需配置资源大小，但与SQL不同的是，开发者须要定制化编写对应的Java类（class参数），以及利用的输出（input参数）和输入参数（output参数），此利用中输出为SQL代码，输入为HDFS文件。从Test类实现能够看出，开发者只需三步走：获取输出数据、逻辑解决、后果输入，即可实现代码编写。

<?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>
package com.way.app.instance.ecommerce;import com.way.app.Application;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.sql.Row;import java.util.Map;import org.apache.spark.api.java.function.FilterFunction;import org.apache.spark.sql.Dataset;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");        // 逻辑解决（简略的筛选出url带有局部站点的日志）        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利用雷同，回溯利用类为惟一类（已实现），有且只有一个，供所有回溯工作应用，且反对多种回溯计划。

2.3.3.1 方案设计

在日常回溯过程中发现，一次回溯工作存在重大的工夫节约，无论以何种形式提交工作，都须要经验以下执行环境申请及筹备的过程：

  1. 在client提交application，首先client向RS申请启动ApplicationMaster
  2. RS先随机找到一台NodeManager启动ApplicationMaster
  3. ApplicationMaster向RS申请启动Executor的资源
  4. RS返回一批资源给ApplicationMaster
  5. ApplicationMaster连贯Executor
  6. 各个Executor反向注册给ApplicationMaster
  7. ApplicationMaster发送task、监控task执行，回收后果

这个过程占用的工夫咱们统称为执行环境筹备，咱们提交工作后，经验如下三个过程：

  1. 执行环境筹备
  2. 开始执行代码
  3. 开释资源

执行环境筹备通常会有5-20分钟的等待时间，以队列过后的资源状况高低稳定，失败率为10%左右，失败起因因为队列、网络、资源有余等造成的不可抗力因素；代码执行过程通常失败率在5%左右，通常因为节点不稳固、网络等因素导致。离线开发框架回溯利用从节省时间和人力两个方面思考，设计方案图2-3所示。

从回溯工夫方面来看：将所有回溯子工作的第一、第三步的工夫压缩为一次，即环境筹备及开释各一次，执行屡次回溯代码。若开发者回溯工作为30个子工作，则节俭的工夫为5-20分钟乘29，可见，回溯子工作越多，回溯提效越显著。

从人工染指方面来看，第一，开发者无需额定开发、增加回溯配置即可。第二，离线框架回溯利用启动的工作数量远远小于传统回溯计划，以图2-3为例，该回溯工作为串行回溯形式，应用框架后只需关注一个工作的执行状态，而传统形式则需人工保护N个工作的执行状态。

最初，咱们在应用离线开发框架回溯一个一年的串行工作中，代码的执行只须要5分钟左右，咱们发现，不应用离线开发框架回溯的工作在最现实的状况下（即最短时间调配到资源、所有子工作均无失败状况、一次能够串行启动365天），须要的工夫为2.5天，但应用离线开发框架回溯的工作，在最坏的状况下（即最长工夫调配到资源，工作失败状况呈现10%），只须要6个小时就可实现，提效90%以上，且根本无需人力关注。

2.3.3.2 性能介绍
断点续回

应用Spark计算，咱们在享受其计算带来的飞快速度时，难免会遭逢其中的不稳定性，节点宕机、网络连接失败、资源问题带来的工作失败不足为奇，回溯工作动辄几个月、甚至一年，任务量微小，失败后能够持续从断点处回溯显得尤为重要。在离线框架设计中，记录了工作回溯过程中已胜利的局部，工作失败重启后会进行断点续回。

回溯程序

在回溯工作中，通常咱们会依据业务须要确定回溯程序，如对于有新老用户的增量数据，因为以后的日期数据依赖历史数据，所以咱们通常会从历史到当初开始回溯。但没有这种须要时，一般来说，先回溯当初能够疾速满足业务方对当初数据指标的理解，咱们通常会从当初到历史回溯。在离线框架设计中，开发者可依据业务须要抉择回溯程序。

并行回溯

通常，回溯工作优先级低于例行工作，在资源无限的状况下，回溯过程中不能一次性全副开启，免得占用大量资源影响例行工作，所以离线框架默认为串行回溯。当然在资源充沛的时间段，咱们能够抉择适当的并行回溯。离线开发框架反对肯定的并发度，开发者在回溯工作时熟能生巧。

2.3.3.3 创立一个回溯工作

回溯利用的应用十分不便，开发者无需新开发代码，应用例行的代码，配置回溯计划即可，如下代码所示，

  • class参数为回溯利用的惟一类，必填参数，所有回溯工作无需变动。
  • type参数为回溯利用类型，默认为sql，若利用类型为java，则type值应为java类名。
  • path参数为回溯代码门路，必填参数，无默认值，通常与例行工作代码雷同，无需批改。
  • limitdate参数为回溯的截止日期，必填参数，无默认值。
  • startdate参数为回溯开始日期，必填参数，无默认值，若工作进入断点续回或开启并行回溯时，则该参数有效。
  • order参数为回溯程序，默认为倒序。当值为1时为正序，为值为-1时为倒序。
  • distance参数为回溯步长，框架默认为串行回溯，但也反对并行回溯，该参数次要用于反对并行回溯，当该参数存在且值不为-1时，回溯开始日期取值为基准日期。如启动两个并行任务，工作的执行范畴为基准日期至基准日期加步长或limitdate，若基准日期加步长后日期大于limitdate，则是取limitdate，否则反之。
  • file参数为回溯日志文件，必填参数，无默认值，用于记录已回溯胜利的日期，当失败再次重启工作时，startdate会以日志文件中日期的下一个日期为准。
  • 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反对的语言远不止这两种，咱们须要持续对框架降级反对多语言开发等，让开发者更不便、疾速地进行大数据开发。

----------  END  ----------

举荐浏览：

云原生时代的搜寻服务算力治理

浅谈小程序开源业务架构建设之路

百度小程序包流式下载安装优化

前端工程化之FaaS SSR计划

日志中台不重不丢实现浅谈

百度ToB垂类账号权限平台的设计与实际