Spark MLib 每月一算法 —— QuantileDiscretizer

2018-03-29

QuantileDiscretizer

QuantileDiscretizer是一个Estimator，用来将某个连续值类型的属性转化成离散的数字分类编码属性，以便后续其他适用于数字编码的算法使用。离散化的过程也被称作分箱(Bucketizer)。
从算法名称可以看出QuantileDiscretizer是通过分位数来确定每一个箱体(bucket)的范围，比如希望将数据分成2组，那么这2组的范围分别会是[-Infinity,中位数]和[中位数,+Infinity]。但由于实际中严格获取分位点比较耗时，所以Spark采取近似算法，用户预先给出可以接受的近似误差。
QuantileDiscretizer需要根据已有训练集数据来进行fit，得到数据集对应的箱体编码model，后续有新的属性数据会被transform到对应的箱体中，针对Null值数据可以设置handleInvalid参数来处理。

参数

inputCol: 需要进行离散化的属性名称
inputCols: 需要进行离散化的一组属性名称，一次处理多个属性的便捷方式
outputCol: 离散化后的属性名称 (默认值: quantileDiscretizer_######_output)
outputCols: 离散化后的属性名称，一次处理多个属性的便捷方式，次序和上面的inputCols一致
numBuckets: 需要分成多少个类别(bucket)，分完箱之后每个类别的编号取值范围为0～numBuckets-1 (>=2的整数，默认值: 2)
numBucketsArray: 针对每一个属性需要分成多少个类别(bucket)的值组成的数组，一次处理多个属性的便捷方式，次序和上面的inputCols一致
handleInvalid: 表示在transform期间如何处理Null值。 (默认值: error)
可选项1. skip-放弃有Null值的row；
可选项2. error-抛出异常org.apache.spark.SparkException: Failed to execute user defined function($anonfun$apply$1: (double) => double)；
可选项3. keep-将Null值都放入一个额外的类别(bucket)，该额外类别编号为numBuckets定义的数字。
relativeError: 分位数近似算法的误差精度 ([0,1]之间的实数，0表示没有误差，默认值0.001）

例子

val training = spark.createDataFrame(
  Seq((0, 18.0, 1000), (1, 19.0, 2300), (2, 8.0, 500), (3, 5.0, 1500), (4, 2.2, 1500))
).toDF("id", "age", "income")
training.show()
//    +---+----+------+
//    | id| age|income|
//    +---+----+------+
//    |  0|18.0|  1000|
//    |  1|19.0|  2300|
//    |  2| 8.0|   500|
//    |  3| 5.0|  1500|
//    |  4| 2.2|  1500|
//    +---+----+------+


val discretizer = new QuantileDiscretizer()
  .setInputCols(Array("age", "income"))
  .setOutputCols(Array("ageBucket", "incomeBucket"))
  .setNumBucketsArray(Array(3, 2))

val model = discretizer.fit(training)
model.transform(training).show()
//    +---+----+------+---------+------------+
//    | id| age|income|ageBucket|incomeBucket|
//    +---+----+------+---------+------------+
//    |  0|18.0|  1000|      2.0|         0.0|
//    |  1|19.0|  2300|      2.0|         1.0|
//    |  2| 8.0|   500|      1.0|         0.0|
//    |  3| 5.0|  1500|      1.0|         1.0|
//    |  4| 2.2|  1500|      0.0|         1.0|
//    +---+----+------+---------+------------+


val test = spark.createDataFrame(
  Seq((0, Double.NaN, 188.0), (1, 19.0, -2300.0),
    (2, 34.4, Double.NaN),
    (3, -55.0, 345.0), (4, 2.2, 1500.0))
).toDF("id", "age", "income")

try {
  model.transform(test).show()
} catch {  //error from NaN value.
  case ex: SparkException => println(ex)
}

model.setHandleInvalid("keep")
model.transform(test).show()
//    +---+-----+-------+---------+------------+
//    | id|  age| income|ageBucket|incomeBucket|
//    +---+-----+-------+---------+------------+
//    |  0|  NaN|  188.0|      3.0|         0.0|
//    |  1| 19.0|-2300.0|      2.0|         0.0|
//    |  2| 34.4|    NaN|      2.0|         2.0|
//    |  3|-55.0|  345.0|      0.0|         0.0|
//    |  4|  2.2| 1500.0|      0.0|         1.0|
//    +---+-----+-------+---------+------------+