浅谈Spark在大数据开发中的一些最佳实践

1.前  言

eBay 智能营销部门在长时间的生产实践中，我们总结了一套基于Scala开发Spark任务的可行规范，来帮助我们写出高可读性、高可维护性和高质量的代码，提升整体开发效率

2.基本开发规范

一、字段名规范

字段名应足够简洁清晰，使阅读者能快速理解字段内容。

PARTNER_CLICK,INTERNAL_CLICK

二、业务字典

对于公司已有统一命名的专业术语，应采用此命名方式，例如 GMB。

对于公司级别命名并未统一的专业术语，在 domain 或 team 内部应有统一的命名规范。比如你的ETL任务中用到了多个系统的数据，对于用户ID，系统A里面叫user_id，系统B里面叫u_id，系统C里面叫mapped_id，当经过我们数据清洗流程后我们应该将这些字段统一成同个概念，比如USER_ID。

数据 schema 中字段名应用下划线分割，而代码变量仍旧应该采用驼峰命名法，字段与变量应该有默认对应关系。

建议维护一个业务名词库用来统一定义专业概念和术语，注明是公司级别的术语或是 domain/team 级别的术语，级别名称应在字段名上体现。

三、幂等性三、幂等性

一个spark任务应该是幂等的，这个任务在有同样的输入时被执行多次输出是恒定的，不应该产生副作用。一个spark任务应该是幂等的，这个任务在有同样的输入时被执行多次输出是恒定的，不应该产生副作用。

四、数值类型四、

在分析计算需求的时候，需要先对数值类型进行分类，不同的数值类型的计算方式也会不同。

原始数值指标：

🚩 示例

🚩 对于一个广告系列中，我们可以直接将广告系列中的产品的GMB直接相加得到总GMB

衍生数值指标：

🚩 示例

🚩 对于一个广告系列，我们不能直接将广告系列中的CPC相加得到总CPC

定类数据 (Nominal level)

定类数据不作为数值指标，不可进行算数计算。

3

基本编码规范

一、建议将建表DDL和写数据分离，并且不要在编码中使用drop+create来覆写表数据一、

因为 drop table 和 create table 是非原子性操作，

如下sql，如果create table失败，table将处于不可用状态：

更佳的方式应该如下：

当数据重新生成完以后只需要使用原子操作更新hive的location即可，这样就可以保证每次写入数据时不影响表的使用。

二、特殊的逻辑应该要有注释，比如   ，应该说明这个字段和对应的值的作用，或者定义一个常量来语义化这个魔法值，二、

三、在hive中没有布尔值，禁止使用true/false，它在hive中会变成字符串‘true’/’false’，所以请使用数值类型代替布尔类型。三、

四、若使用Spark SQL编写代码，那么最好不要将核心的SQL逻辑拆分成片段，这样会使可读性下降。对于有多段逻辑的Job，需要让代码更多的暴露出表操作的核心逻辑。四、若

过度的函数分装会使代码可读性下降，从而无法快速直观的了解表操作的逻辑，

稍微修改一下，以下是伪代码， 我们可以将表操作的逻辑暴露出来，而把非核心逻辑封装起来，这样我们可以轻松理解这段代码到底是在做什么：

4

Spark开发最佳实践

一、使用Spark cache时，需要考虑它能否带来计算时间上的提升。Spark cache是使用给定的存储级别来缓存表的内容或查询的输出内容，常用于未来查询中复用原始文件的场景。

Cache的存储级别分为以下几种：

如果所需要 cache的数据集过大，使用 MEMORY_ONLY 容易导致OOM；而使用默认的MEMORY_AND_DISK，当内存不适合写入时则会写入磁盘，这时I/O会占用大量时间，并且由于内存减少导致频繁GC，反而使效率下降。在使用 cache 的时候需要平衡好数据 I/O 的开销和计算资源的使用。如果一个数据集cache消耗的I/O时间不是明显小于直接重计算消耗的时间，不建议使用cache。

是由一张小表 join大表生成的，cache数据时会产生额外的磁盘写入开销；而考虑到这个 join 操作本身所需要的计算时间并不多，如果从时间性能的角度考虑，这个case我们就不应该使用cache

二、二、二、DataFrame的 API 和Spark SQL中的 union 行为是不一致的

三、两个DataFrame来源于同一个数据源，如果直接将它们join则会报以下错：三、三、

Detected implicit cartesian product for LEFT(INNER/RIGHT) OUTER join between logical plans

由于来自同一个数据源的DataFrame join很容易产生笛卡尔积，所以Spark默认禁止这种行为。但是在一些业务场景中的确有这种join的情况，解决方案有两种：

在join前将数据存储到临时目录(一般是HDFS)，再重新加载进来，用来截断血缘。

四、写入分区表时，Spark会默认覆盖所有分区，四、写入分区表时，四、写入分区表时，

五、DataFrame中使用udf时，需要注意udf的参数如果是基础类型则必须不为空，否则不会被执行。五、

示例：如下代码，一个简单根据int值生成对应的flag，但是如果norb是null的话，那么这段udf不会被执行，对应的norbFlag为null。对于这种由于null值导致的逻辑不一样问题，可以借助DataFrameNaFunctions 来协助处理null值情况。

六、Spark原生不支持数据更改，所以对一些非分区表更新数据是有挑战的。六、

5

后  记

此篇文章总结了我们在使用 Spark 过程中所遇到的挑战和技术案例，希望能够抛砖引玉，引出更多更好的实践方案。杨青波刘炼刘轶

参考

「1」https://github.com/delta-io/delta/blob/73ca6fcea0a25f302ee655f9849f86832bbe5f23/examples/scala/src/main/scala/example/QuickstartSQL.scala

–END–