数据预处理

 前面我们介绍了探索数据的一些方法，通过对数据的探索，可以帮助我们发现一些奇异数据、缺失数据、特征的类别及其分布情况等信息。而这些信息正是对数据预处理的重要依据。在数据分析、机器学习中，数据的预处理非常关键，尤其是涉及大数据的处理，往往是比较费时、费神的过程，有时还需要往返多次。当然，如果数据预处理做得好，除提高数据质量外，更能极大提高模型的性能。  数据的预处理一般包括数据清理、数据变换、数据集成、数据归约等

1.1、数据清理

 数据清理的主要任务是填补缺失值、光滑噪声数据、处理奇异数据、纠正错误数据、删除重复数据、删除唯一性属性、 去除不相关字段或特征、处理不一致数据等。 噪声数据的处理方法为分箱、聚类等。以下分别以处理缺失数据、异常数据为例，说明在Spark中如何进行数据清理。

1.1.1、处理缺失数据

def test5():
    df = pd.read_csv("C:/Users/12285/Downloads/catering_sale.csv", header=0, encoding = 'gb2312')
    # 定位数据中的空值
    print(df[df.isnull().values == True])
    print(df.count())   # 空值
    print(df.describe())    # 统计信息
    # 空值填补0
    fill_df = df.fillna(0)
    print(fill_df[fill_df.isnull().values == True])

    fill_df1 = df['sale_amt'].fillna(df['sale_amt'].count())
    print(fill_df1)

    plt.figure()
    # 画箱型图， 通过箱型图查看异常值
    bp = fill_df.boxplot(return_type='dict')

    # fliers 为异常值的标签
    x = bp['fliers'][0].get_xdata()
    y = bp['fliers'][0].get_ydata()
    y.sort()
    # 用annotate添加注释
    for i in range(len(x)):
        plt.annotate(y[i], xy=(x[i], y[i]), xytext=(x[i] + 0.1 - 0.8 / (y[i] - y[i - 1]), y[i]))

    plt.show()

1.1.2、处理奇异数据

 在数据探索阶段，我们发现销售数据文件catering_sale.csv中有6个可能的奇异数据，假设与相关人员核实后，只有22.0为奇异数据或错误数据。对错误数据我们一般采用删除或替换的方法，这里我们采用Spark SQL来处理奇异数据。

spark = SparkSession.builder.appName("test01") \
    .getOrCreate()
sc = spark.sparkContext

def test6():
    df = spark.read.csv("hdfs://192.168.179.14:8020/mlDataSet/catering_sale.csv", header=True)
    df.show
    df1 = df.select(df['sale_date'], df['sale_amt'].cast("Double"))
    df1.replace(22.0, 200.0, 'sale_amt')
    df1.show()

1.2、数据变换

  数据变换是数据预处理中的-项重要内容，如对数据进行规范化、离散化、衍生指标、类别特征数值化、平滑噪声等都属于数据变换。在Spark ML中有很多现成的数据变换算

  这里我们以卡方检验为例，分析如何根据特征的贡献率来选择特征。假设我们有很多特征，如表示时间特征的季节( season)、年月(yr)、 月份( mnth)、是否节假日( holiday)、是否周末( weekday),表示天气的特征weathersit、temp 等。为了使用卡方检验来选择这些特征，首先需要把各特征组合为一个特征向量，然后，把整合后的特征向量及选择特征个数等代入卡方模型中，详细代码如下:

    featuresArray = ["season", "yr", "mnth", "hr", "holiday", "weekday", "workingday",
                     "weathersit", "temp", " atemp", "hum", "windspeed"]
    ###把各特征组合成特征向量features
    assembler = VectorAssembler(inputCols=featuresArray, outputCol=" features")
    ###选择贡献度较大的前5个特征
    selectorfeature = ChiSqSelector(numTopFeatures=5, featuresCol=" features ", outputCol=" selectedFeatures", labelCol="label")

1.3、数据集成

   数据集成是将多文件或者多数据库中的数据进行合并，然后存放在一个一致的数据存储中。数据集成一般通过join、 union或merge等关键字把两个(或多个)数据集连接在一起，Spark SQL (包括DataFrame) 有join方法，Pandas 下有merge方法。数据集成往往需要耗费很多资源，尤其是大数据间的集成涉及shuffle 过程，有时需要牵涉多个节点，所以除了数据一致性外，性能问题常常不请自来，需要我们特别留心。传统数据库一般是在单机上采用hash join方法，如果在分布式环境中，采用join时，可以考虑充分利用分布式资源进行平行化。当然，在进行join之前，对数据过滤或归约也是常用的优化方法。

1.4、数据规约

  大数据是机器学习的基础，但大数据往往数据量非常大，有时我们可以通过数据归约技术删除或减少冗余属性(或维)、精简数据集等，使归约后的数据比原数据小，甚至小很多，但仍然接近于保持原数据的完整性，且结果与归约前结果相同或几乎相同。下图列举了Spark ML自带的特征选择或降维算法。

  选择特征或降维是机器学习中重要的处理方法，我们可以使用上述这些方法在减少特征个数、消除噪声等问题的同时，维持原始数据的内在结构或主要特征。尤其是降维，在大数据、机器学习中具有重要作用。