Spark Packages寻宝（一）：简单易用的数据准备工具Optimus

作者：李呈祥，花名司麟 ，阿里云智能EMR团队高级技术专家，Apache Hive Committer， Apache Flink Committer，目前主要专注于EMR产品中开源计算引擎的优化工作。

Spark社区在Spark Packages网站中索引了许多第三方库，这些第三方库由不同的开发者贡献，作为Spark生态圈的一部分，扩充了Spark的使用范围和使用场景，其中很多对于我们日常的使用可能有帮助，我们准备开启一个系列文章介绍Spark Packages中一些有意思的第三方库，作为系列的第一篇，本文主要介绍Optimus，一个基于PySpark的简单易用的数据准备工具。

本文的部分内容源自Optimus官网和相关介绍文章，原文链接参考文末引用部分。

在Spark（Pyspark）的支持下，Optimus允许用户使用自己的或一组预先创建的数据转换功能来清理数据，对其进行概要分析并应用与数据分析和机器学习等场景，可以轻松地利用python语言进行所有这些操作。Optimus主要关注与以下几个方面：

• 创建一个可靠的API来访问和处理数据。
• 让用户轻松地从Pandas迁移。
• 使数据探索更加容易。

创建一个可靠的API来访问和处理数据

首先，我们来看看Optimus的基本使用方式

from pyspark.sql import SparkSession
from optimus import Optimus

// 创建context
spark = SparkSession.builder.appName('optimus').getOrCreate()
op= Optimus(spark)

// 加载数据
df = op.load.csv("../examples/data/foo.csv")

// 执行清理
new_df = df\
    .rows.sort("rank","desc")\
    .withColumn('new_age', df.age)\
    .cols.lower(["names","function"])\
    .cols.date_transform("date arrival", "yyyy/MM/dd", "dd-MM-YYYY")\
    .cols.years_between("date arrival", "dd-MM-YYYY", output_cols = "from arrival")\
    .cols.remove_accents("names")\
    .cols.remove_special_chars("names")\
    .rows.drop(df["rank"]>8)\
    .cols.rename(str.lower)\
    .cols.trim("*")\
    .cols.unnest("japanese name", output_cols="other names")\
    .cols.unnest("last position seen",separator=",", output_cols="pos")\
    .cols.drop(["last position seen", "japanese name","date arrival", "cybertronian", "nulltype"])

//保存结果
new_df.save.csv("data/foo.csv")

Optimus基本PySpark框架，重新组织了对数据进行清理/准备的相关API，把数据处理整理为针对dataframe.rows和dataframe.cols两类操作，基于rows和cols实现了非常丰富的针对数据清理和准备相关的接口，用户可以使用这些接口非常方便高效地完成相关工作。在Optimus中，核心的数据操作可以归纳为如下几类：

• 创建一个DataFrame
• 用append()追加行或列
• 使用select()选择行或列
• 使用apply()更新或转换列数据
• 使用drop()删除行或列
• 使用read()加载数据
• 使用write()保存数据

针对列的操作

对于数据集的操作主要是针对列进行的，所以这里主要介绍一些典型的针对列的操作类型：

Aggregation

Optimus扩展了PySpark的操作，创建了一种更简单的方式来对数据集进行统计。

print（df.cols.min（“ height”））
print（df.cols.percentile（['height'，'rank']，[0.05，0.25，0.5，0.75，0.95]））
print（df.cols.max（“ height”））
print（df.cols.median（[“ height”，“ rank”]））
print（df.cols.range（[“ height”，“ rank”]）））
print（df.cols.std（[“ height”，“ rank”]））17.5    
{'height'：{0.05：17.5，0.25：17.5，0.5：26.0，0.75：28.0，0.95：28.0}，'rank'：{0.05：7.0，0.25：7.0，0.5：7.0，0.75：10.0，0.95 ：10.0}} 
28.0 
{'height'：26.0，'rank'：7.0} 
{'height'：{'min'：17.5，'max'：28.0}，'rank'：{'min'：7，'max '：10}} 
{'height'：{'stddev'：5.575242894559244}，'rank'：{'stddev'：1.7320508075688772}}

Transformation and Chaining

类似PySpark DataFrame的操作，Optimus的数据转换操作也可以链接起来，甚至还可以和PySpark的DataFrame操作链接起来。利用Spark的延迟计算的特性（在示例中show()才会触发计算），使用Catalyst优化执行计划。

df = df\
    .rows.sort(["rank","height"])\
    .cols.lower(["names","function"])\
    .cols.remove_accents("names")\
    .cols.remove_special_chars("names")\
    .cols.trim("names")\
    .show()
+---------+------+---------+----+
|    names|height| function|rank|
+---------+------+---------+----+
|  optimus|  28.0|   leader|  10|
| ironhide|  26.0| security|   7|
|bumblebee|  17.5|espionage|   7|
+---------+------+---------+----+

Nest/Unnest

使用nest和unnest操作可以将多列合并成一个新列或者将一列拆分为多列，如下所示：

df.cols.nest(["names", "function"], output_col = "new_col", shape ="string").show()
+---------+------+---------+----+-------------------+
|    names|height| function|rank|            new_col|
+---------+------+---------+----+-------------------+
|  optimus|  28.0|   leader|  10|     optimus leader|
| ironhide|  26.0| security|   7|  ironhide security|
|bumblebee|  17.5|espionage|   7|bumblebee espionage|
+---------+------+---------+----+-------------------+

df.cols.unnest("new_col", " ").cols.drop("new_col")
+---------+------+---------+----+---------+---------+
|    names|height| function|rank|new_col_0|new_col_1|
+---------+------+---------+----+---------+---------+
|  optimus|  28.0|   leader|  10|  optimus|   leader|
| ironhide|  26.0| security|   7| ironhide| security|
|bumblebee|  17.5|espionage|   7|bumblebee|espionage|
+---------+------+---------+----+---------+---------+

自定义转换

Optimus具有两个函数apply（）和apply_expr（），用户可以在其中实现函数（UDF或Pandas UDF）或列表达式。

from pyspark.sql import functions as F
def func(value, args):
    return value + 1
  
df.cols.apply("height", func, "int")\
  .cols.apply_expr("rank", F.col("rank")+1)\
  .show()
    
+---------+------+---------+----+
|    names|height| function|rank|
+---------+------+---------+----+
|  optimus|    29|   leader|  11|
| ironhide|    27| security|   8|
|bumblebee|    18|espionage|   8|
+---------+------+---------+----+

让用户轻松地从Pandas迁移

在数据分析领域，Python是通用语言，而Pandas是最常用的库，所以Optimus在设计时尽量和Pandas的接口保持一致，以下是Optimus和Pandas以及PySpark的接口对比：

NI= Not implemented
NA= Not

除了在Spark无法实现的功能（如reset_index），Optimus实现了几乎所有Pandas可应用于Spark的功能，而且两个接口基本一致，大大方便了Pandas用户的迁移。

使数据探索更轻松

Optimus具有功能强大的内置数据探查器，除所有基本操作外，它还提供了独特的数据探查功能。用户可以查看特定列中存在多少种数据类型。例如，有一百万行的颜色值为白色，黑色，红色以及数百种颜色，如何确定一百万行中没有“ 1”数字？数据探查使得用户可以了解数据集的质量，是否有脏数据，为数据清理和准备提供前提信息和验证方式，使数据集以合适的状态用于后续的数据分析和ML/DL处理。

df = op.load.csv("https://raw.githubusercontent.com/ironmussa/Optimus/master/examples/data/Meteorite_Landings.csv").h_repartition()
// 对name列进行profile
op.profiler.run(df, "name", infer=False)

可以看到profile提供了详细的数据统计信息，包括name列的数据类型，null数量，distinct count数量，topN出现次数，histogram分布等等，基于这些信息，用户可以准确了解数据的质量。

Profile由于计算量比较大，可能会比较耗时，特别是对于distinct count这种操作，使用relative_error 和 approx_count参数可以以降低精度为代价加速profile速度。

op.profiler.run(df, "name", infer=False, relative_error =1, approx_count=True)

总结

本文主要介绍了Optimus项目，作为一个Spark的第三方库，Optimus基于PySpark，为用户提供了一套完整的数据质量探查和数据清理工具集，接口参考Pandas设计，易用且强大，非常适合大规模数据的清理准备工作。限于篇幅，还有很多Optimus的清理接口和Profile功能没有介绍，感兴趣的同学可以访问Optimus官网探索更多功能和用法。

引用

1. https://hi-optimus.com/
2. https://github.com/ironmussa/Optimus
3. http://docs.hioptimus.com/en/latest/sections/overview.html
4. https://towardsdatascience.com/announcing-optimus-v2-agile-data-science-workflows-made-easy-c127a12d9e13

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区，定期推送精彩案例，技术专家直播，问答区近万人Spark技术同学在线提问答疑，只为营造纯粹的Spark氛围，欢迎钉钉扫码加入！

对开源大数据和感兴趣的同学可以加小编微信（下图二维码，备注“进群”）进入技术交流微信群。

Apache Spark技术交流社区公众号，微信扫一扫关注