利用Spark 实现数据的采集、清洗、存储和分析

学习本文，你将了解spark是干啥的，以及他的核心的特性是什么，然后了解这些核心特性的情况下，我们会继续学习，如何使用spark进行数据的采集/清洗/存储/和分析。

spark是干啥的

Apache Spark 是一个用于大规模数据处理的统一分析引擎。它提供了高级的数据分析能力和支持多种数据源的灵活性。Spark 的主要特点包括：

• 快速处理：利用内存计算，Spark 能够快速处理大量数据。
• 易于使用：提供了 Scala、Java、Python 和 R 等多种编程语言的接口，本文为了简单，使用Python进行示例的讲解，因为我已经装了Python的环境。
• 多组件支持：包括 Spark SQL（用于处理结构化数据）、Spark Streaming（用于处理实时数据）、MLlib（机器学习库）和 GraphX（图计算框架）。

当然，做大规模数据处理的引擎可不止spark一个，还有 hadoop MapReduce,Flink等等，了解一下他们的特性，对于我们做项目选型也是非常重要的。

一个demo，使用spark做数据采集，清洗，存储，分析

好吧，废话也不在多说了，开始我们的demo环节了，Spark 可以从多种数据源（例如 HDFS、Cassandra、HBase 和 S3）读取数据，对于数据的清洗包括过滤、合并、格式化转换，处理后的数据可以存储回文件系统、数据库或者其他数据源，最后的工序就是用存储的清洗过的数据进行分析了。

假设我们有一个 CSV 格式的数据文件，其中包含了用户的信息，比如姓名、年龄和国籍。我们的目标是读取这个文件，清洗数据（比如去除无效或不完整的记录），并对年龄进行平均值计算，最后将处理后的数据存储到一个新的文件中。

其中有一些异常数据是需要我们清洗的，数据格式如下图所示：

代码环节：数据读取，从一个原始的 csv 文件里面读取，清洗是对一些脏数据进行清洗，这里是清理掉年龄为负数的项目，数据分析是看看这些人群的平均年龄。

from pyspark.sql import SparkSession
from pyspark.sql.functions import mean
# 初始化 Spark 会话
spark = SparkSession.builder.appName("UserDataAnalysis").getOrCreate()
# 读取 CSV 文件
df = spark.read.csv("users.csv", header=True, inferSchema=True)

print(df.show())
# 清洗数据，例如去除年龄为 null 或非正常值的行
df_clean = df.filter(df.Age > 0)

print(df_clean.show())

# 计算年龄的平均值
avg_age = df_clean.select(mean("Age")).collect()[0][0]
print(f"Average age: {avg_age}")

# 将处理后的数据存储为新的 CSV 文件
# df_clean.write.csv("result.csv", header=True)

# 关闭 Spark 会话
spark.stop()

执行一下看看：

这里，可以看到，我们讲异常数据首先讲异常数据清理掉，然后使用 avg_age = df_clean.select(mean("Age")).collect()[0][0] 计算了一下平均年龄，符合预期。至于数据的存储，我们可以直接以csv的方式存在本地。df_clean.write.csv("result.csv", header=True)

以下是我存储的清洗后的数据的一个示例：

总结

本文这个例子对于 spark 来说应该算是高射炮打文字了，spark 在做数据清洗上绝对不是仅仅这么点刷子，我们这里使用 spark sql 对结构化数据做了简单的清洗，你可能了解过，我们还可以使用 Spark MLlib 或 Spark ML 来进行数据质量检查和数据 profiling，以识别数据中的异常值、离群值、噪声等问题。另外对于数据分析，我们可以使用 Spark MLlib 或 Spark ML 来进行机器学习和统计分析，如回归、分类、聚类、降维等，甚至使用 Spark GraphX 来进行图数据分析，如社区检测、页面排名等。另外，在数据可视化方面， Spark 连接外部可视化工具，如 Tableau、PowerBI、QlikView 等，来可视化数据。

因此本文就是一个 spark 入门而已，门槛很低的那种，一些高端的玩法还需要下点功夫去探索。