Spark Packages寻宝(一):简单易用的数据准备工具Optimus

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: 本文主要介绍了Optimus项目,作为一个Spark的第三方库,Optimus基于PySpark,为用户提供了一套完整的数据质量探查和数据清理工具集,接口参考Pandas设计,易用且强大,非常适合大规模数据的清理准备工作。限于篇幅,还有很多Optimus的清理接口和Profile功能没有介绍,感兴趣的同学可以访问[Optimus官网](https://hi-optimus.com/)探索更多功能和用法。

作者:李呈祥,花名司麟 ,阿里云智能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的接口对比:

Description Pandas Spark Optimus
Read csv file pd.read_csv() spark.read.csv() op.read.csv()
Create Dataframe pd.Dataframe df.createdataframe() op.create.df()
Append Row df.append df.union() df.row().append()
Column Mean df.mean df1.agg({"x": "max"}) df.cols().mean()
Show Rows from Dataframe df.head() df.show() df.show()
Drop Columns df.drop() df.drop() df.cols().drop()
Sum all values in a Column df.sum() df1.agg({"x": "sum"}) function df.cols().sum()
Save Dataframe to csv df.to_csv() df.write.csv() df.save().csv()
Get a value by index df.get() NA NA
Get the mode of a column df.mode() NI df.cols().mode()
Cast a Column df.astype() df.column.cast() df.cols().cast(), astype() as alias
Substract 2 dataframes df.sub() NI NI
Merge to dataframes pd.concat() df.union() optimus.concat()
Apply a user defined fucntion to a column df.apply(func) fn = F.udf(labmbda x:x+1, DoubleType()) df.withColumn('disp1', n(df.disp)) df.cols().apply(func)
Group rows df.groupby() df.groupby() df.groupby()
Joint operation between to dataframes df.join() df.join() df.join()
Fill Null values with x df.fillna() df.fillna() df.fillna()
Get the max number of a Column df.max() df1.agg({"x": "max"}) df.cols().max()
Reset index reset_index() NA NA

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)

1.png

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

Profile由于计算量比较大,可能会比较耗时,特别是对于distinct count这种操作,使用relative_errorapprox_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氛围,欢迎钉钉扫码加入!
image.png

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

image.png

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

image.png

相关实践学习
数据湖构建DLF快速入门
本教程通过使⽤数据湖构建DLF产品对于淘宝用户行为样例数据的分析,介绍数据湖构建DLF产品的数据发现和数据探索功能。
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
相关文章
|
2月前
|
SQL 分布式计算 API
Spark学习------SparkSQL(概述、编程、数据的加载和保存)
Spark学习------SparkSQL(概述、编程、数据的加载和保存)
104 2
|
28天前
|
存储 分布式计算 Hadoop
Spark和Hadoop都是大数据处理领域的重要工具
【6月更文挑战第17天】Spark和Hadoop都是大数据处理领域的重要工具
124 59
|
16天前
|
弹性计算 分布式计算 DataWorks
DataWorks产品使用合集之spark任务如何跨空间取表数据
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
12 1
|
29天前
|
SQL 分布式计算 HIVE
实时计算 Flink版产品使用问题之同步到Hudi的数据是否可以被Hive或Spark直接读取
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
1月前
|
SQL 分布式计算 NoSQL
使用Spark高效将数据从Hive写入Redis (功能最全)
使用Spark高效将数据从Hive写入Redis (功能最全)
|
2月前
|
SQL 分布式计算 关系型数据库
使用 Spark 抽取 MySQL 数据到 Hive 时某列字段值出现异常(字段错位)
在 MySQL 的 `order_info` 表中,包含 `order_id` 等5个字段,主要存储订单信息。执行按 `create_time` 降序的查询,显示了部分结果。在 Hive 中复制此表结构时,所有字段除 `order_id` 外设为 `string` 类型,并添加了 `etl_date` 分区字段。然而,由于使用逗号作为字段分隔符,当 `address` 字段含逗号时,数据写入 Hive 出现错位,导致 `create_time` 值变为中文字符串。问题解决方法包括更换字段分隔符或使用 Hive 默认分隔符 `\u0001`。此案例提醒在建表时需谨慎选择字段分隔符。
|
2月前
|
机器学习/深度学习 数据采集 分布式计算
【机器学习】Spark ML 对数据进行规范化预处理 StandardScaler 与向量拆分
标准化Scaler是数据预处理技术,用于将特征值映射到均值0、方差1的标准正态分布,以消除不同尺度特征的影响,提升模型稳定性和精度。Spark ML中的StandardScaler实现此功能,通过`.setInputCol`、`.setOutputCol`等方法配置并应用到DataFrame数据。示例展示了如何在Spark中使用StandardScaler进行数据规范化,包括创建SparkSession,构建DataFrame,使用VectorAssembler和StandardScaler,以及将向量拆分为列。规范化有助于降低特征重要性,提高模型训练速度和计算效率。
|
2月前
|
机器学习/深度学习 分布式计算 算法
【机器学习】Spark ML 对数据特征进行 One-Hot 编码
One-Hot 编码是机器学习中将离散特征转换为数值表示的方法,每个取值映射为一个二进制向量,常用于避免特征间大小关系影响模型。Spark ML 提供 OneHotEncoder 进行编码,输入输出列可通过 `inputCol` 和 `outputCol` 参数设置。在示例中,先用 StringIndexer 对类别特征编码,再用 OneHotEncoder 转换,最后展示编码结果。注意 One-Hot 编码可能导致高维问题,可结合实际情况选择编码方式。
|
1月前
|
分布式计算 定位技术 Scala
使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析
使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析
|
1月前
|
消息中间件 分布式计算 关系型数据库
使用Apache Spark从MySQL到Kafka再到HDFS的数据转移
使用Apache Spark从MySQL到Kafka再到HDFS的数据转移