spark初步学习

1.1 下载数据集

在此，我们将使用National Health and Nutrition Health Survey数据集。

图3-1 National Health and Nutrition Health Survey数据集

1.2 理解数据

National Health and Nutrition Health Survey数据集出现在 2019年由 An Dinh、Amber Young和 Stacey Miertschin撰写并发表在《BMC医学信息学与决策制定》杂志上的题为 《基于机器学习的数据驱动方法预测糖尿病和心血管疾病》的论文中。

NHANES数据集旨在通过访谈、体检和实验室测试评估美国成人和儿童的健康和营养状况。该数据集由美国疾病控制和预防中心（CDC）的国家卫生统计中心（NCHS）资助，并包括了来自美国各地的调查受访者。针对这个特定的子集，年龄65岁及以上的受访者被标记为“老年人”，而所有年龄在65岁以下的个体则被标记为“非老年人”

这些数据包含6287个观测值，每个方价都对应7个特征数量：

SEQN ID：表示受访者序列号

age_group：表示受访者的年龄组别（老年人/非老年人）

RIDAGEYR：表示受访者年龄

RIAGENDR ：表示受访者性别

PAQ605：表示受访者是否在典型的一周内参与中等或剧烈强度的运动、健身或娱乐活动

BMXBMI：表示受访者的体重指数

LBXGLU：表示受访者禁食后的血糖水平

DIQ010：表示受访者是否患糖尿病

LBXGLT：表示受访者口腔中的血糖水平

LBXIN：表示受访者的血胰岛素水平

1.3 读入数据

在将数据加载到 DataFrame 时指定数据的类型将提供比类型推断更好的性能。

读取数据 dataframe：

val df: DataFrame = spark.read

  .format("csv")

  .option("header", "true")

  .load(filePath)

显示DataFrame的内容：

df.show()

   查看描述性统计信息：

          df.describe().show()

   显示spark dataframe的列名：![](https://img-blog.csdnimg.cn/5037669393f644feb99e04fc644948c7.png)

1.4探索数据

通过spark选取指定列，并展示前10行数据：

val selectedColumns = Seq("age_group", "RIDAGEYR", "RIAGENDR", "PAQ605", "BMXBMI", "LBXGLU", "DIQ010", "LBXGLT", "LBXIN")

val selectedDF = df.select(selectedColumns.map(col): _*)

selectedDF.show(10)

接下来看下该数据集中位数年龄分布情况：

// 构建直方图数据集

val binIndex = frequencies._1.indices

for (i <- binIndex.dropRight(1)) { // 使用 dropRight(1) 避免超出索引范围

  dataset.addValue(frequencies._2(i).toDouble, "", s"${frequencies._1(i)}-${frequencies._1(i + 1)}")

}

   从上图可以看到该数据集青少年居多。

Spark DataFrames 包含一些用于统计处理的内置函数。describe() 函数对所有数值列执行汇总统计计算，并将它们作为 DataFrame 返回。

val numericColumns = df.dtypes.filter{ case (_, dataType) => dataType == "DoubleType" }  // 过滤出 DoubleType 类型的列

  .map{ case (column, _) => column }

// 使用 describe 函数统计数值列

val summaryStats = df.describe(numericColumns: _*)

// 打印汇总统计结果

summaryStats.show()

从上面的统计结果中，我们可以看到所有数据的最小值和最大值。这些数据范围很大，因此我们需要标准化这个数据集。

1.5数据预处理

// 缺失值处理

val imputer = new Imputer()

  .setInputCols(Array("RIDAGEYR", "BMXBMI", "LBXGLU", "LBXGLT", "LBXIN"))

  .setOutputCols(Array("imputed_RIDAGEYR", "imputed_BMXBMI", "imputed_LBXGLU", "imputed_LBXGLT", "imputed_LBXIN"))

val imputedData = imputer.fit(df).transform(df)

// 标签编码

val labelIndexer = new StringIndexer()

  .setInputCol("age_group")

  .setOutputCol("label")

val encodedData = labelIndexer.fit(imputedData).transform(imputedData)

// 特征向量化

val featureCols = Array("RIDAGEYR", "RIAGENDR", "PAQ605", "BMXBMI", "LBXGLU", "DIQ010", "LBXGLT", "LBXIN")

val assembler = new VectorAssembler()

  .setInputCols(featureCols)

  .setOutputCol("features")

val assembledData = assembler.transform(encodedData)

// 独热编码

val encoder = new OneHotEncoder()

  .setInputCol("RIAGENDR")

  .setOutputCol("encoded_gender")

val encodedData = encoder.transform(assembledData)

如上图所示，所有的特征都变成了一个密集向量features。

1.6标准化

接下来，我们终于可以使用 StandardScaler 缩放数据了。输入列是特征，将输出列将命名为"features_scaled"：

// 特征规范化

val scaler = new StandardScaler()

  .setInputCol("features")

  .setOutputCol("features_scaled")

  .setWithMean(true)

  .setWithStd(true)

val scaledData = scaler.fit(assembledData).transform(assembledData)

在上面特征处理中，我们使用了标准化，那是因为我们用到的特征都是数值特征，但是在有些数据集中还会有类别类特征、文本特征、以及其他特征。

其中，类别特征，它们的取值只能是可能状态集合中的某一种。比如说用户性别、职业等便是这类。当类别特征处于原始形式的时候，其取值来自所有可能构成的集合而不是一个数字，故不能作为输入。因此我们需要把类别变量进过编码处理。

文本特征的处理方法有很多，最简单且标准的方法就是词袋法（bag-of-word）。该方法将一段文本视为由其中的文本或数字组成的集合，其处理过程一版分为：分词、删除停用词、提取词干、向量化等步骤。