无监督领域有一个准度和效率双佳的异常点检测算法，我在实践中使用过几次，效果奇好，就是最近几年非常流行的isolation forest（孤立森林）。该算法在sklearn中有现成的包，但是如果大数据的集群上跑的话，目前没有封装好的接口，给分布式任务的部署带来了很多不便（话说spark mllib中集成的算法真心太少了），本文用scala从头进行该算法在spark上的分布式实现，并演示任务在集群上的执行全过程。

一、算法简介

先说一下算法的最少必要知识，细节部分会揉在代码里进行讲解。

1、训练过程：构建森林的树木

iForest由iTree组成。构建每一颗iTree时，从训练数据中抽取N个样本，然后在这些样本中，随机选择一个特征，再随机选择该特征下的一个值，对样本进行二叉划分，然后分别在左右两边的数据集上重复上面的过程，直接达到终止条件，一颗树构建完成。

2、预测过程：计算样本的异常得分

把测试数据在每棵树上沿对应的条件分支往下走，直到达到叶子节点，并记录这过程中经过的路径长度path length(用h(x)表示)。并由此得出异常分数，当分数超过某一阈值，即可判定为异常样本。

二、scala实现

代码主体非原创，参考自国外的一位大神：https://github.com/hsperr/first_steps_in_scala，有部分修改

1、首先，import编写spark程序所需的包，以及scala的Random模块，用于随机选取功能。

import org.apache.spark.SparkContext._
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.Random

2、定义单颗树iTree，第二、三行意味着，每棵树的左右分支ITreeBranch和叶子节点ITreeLeaf都属于iTree的子类。

sealed trait ITree
case class ITreeBranch(left: ITree, right: ITree, split_column: Int, split_value: Double) extends ITree
case class ITreeLeaf(size: Long) extends ITree

3、定义孤立森林的类，完成算法的训练部分，即全部树的构建。

3.1、从样本中抽样，用于构建单个iTree

object IsolationForest {
    def getRandomSubsample(data: RDD[Array[Double]], sampleRatio: Double, seed: Long = Random.nextLong): RDD[Array[Double]] = {
        data.sample(false, sampleRatio, seed=seed)
    }

3.2 、递归构建生成单颗iTree。
参数：
data：上步抽出的样本数据；
maxHeight：树的最大高度即树终止生长的条件；
numColumns：data的特征数量；
currentHeight：树的当前高度。
返回：
一颗完整的ITree

    def growTree(data: RDD[Array[Double]], maxHeight:Int, numColumns:Int, currentHeight:Int = 0): ITree = {
        val numSamples = data.count()
        //递归终止条件，当前树高大于maxHeight或数据量不大于1
        if(currentHeight>=maxHeight || numSamples <= 1){
            return new ITreeLeaf(numSamples)
        }
        //随机选择特征列
        val split_column = Random.nextInt(numColumns)
        val column = data.map(s => s(split_column))
        //随机选择该特征列中的值split_value，用于分割样本
        val col_min = column.min()
        val col_max = column.max()
        val split_value = col_min + Random.nextDouble()*(col_max-col_min)
        //小于分割值的成为左子树，反之右子树
        val X_left = data.filter(s => s(split_column) < split_value).cache()
        val X_right = data.filter(s => s(split_column) >= split_value).cache()

        //递归
        new ITreeBranch(growTree(X_left, maxHeight, numColumns, currentHeight + 1),
            growTree(X_right, maxHeight, numColumns, currentHeight + 1),
            split_column,
            split_value)
    }
}

3.3、将多棵iTree组建成完整森林iforest
参数：
data：全部训练数据；
numTrees：森林中树的个数；
subSampleSize：每棵树采样的大小；
seed：随机种子。
返回：
孤立森林

def buildForest(data: RDD[Array[Double]], numTrees: Int = 2, subSampleSize: Int = 256, seed: Long = Random.nextLong) : IsolationForest = {
        val numSamples = data.count()
        val numColumns = data.take(1)(0).size
        val maxHeight = math.ceil(math.log(subSampleSize)).toInt
        val trees = Array.fill[ITree](numTrees)(ITreeLeaf(1))

        val trainedTrees = trees.map(s=>growTree(getRandomSubsample(data, subSampleSize/numSamples.toDouble, seed), maxHeight, numColumns))

        IsolationForest(numSamples, trainedTrees)
    }

4、定义预测功能类

4.1 预测功能类定义为IsolationForest的样例类，
参数
num_samples：单课iTree的样本数目
trees：已经构建好的孤立森林iforest

主函数predict，
参数x：要预测的单条样本数组，
返回：异常得分Anomaly Score
步骤：
在每一棵iTree上，计算样本达到叶子节点走过的路径长度，然后将得到的不同路径长度按照如下公式进行计算，得到异常得分，走过的路径越短，得分越高，代表越异常。

公式中，h(x)代表路径长度，E(h(x))代表在不同的iTree上路径长度的均值，即群体决策，分母是用来归一化的。

case class IsolationForest(num_samples: Long, trees: Array[ITree]) {
    def predict(x:Array[Double]): Double = {
        val predictions = trees.map(s => pathLength(x, s, 0)).toList
        println(predictions.mkString(","))
        math.pow(2, -(predictions.sum/predictions.size)/cost(num_samples)) //Anomaly Score
    }

上面代码用到的cost 方法和pathLength方法定义如下，
cost方法参数为二叉树中的样本个数，范围该二叉树的平均路径长度，公式为：

    def cost(num_items:Long): Int =
        //二叉搜索树的平均路径长度。0.5772156649:欧拉常数
        (2*(math.log(num_items-1) + 0.5772156649)-(2*(num_items-1)/num_items)).toInt

pathLength方法是一个递归计算，因为每走一步，接下来面对的仍然是一颗树，分支树或者叶子节点。
参数：样本x，单颗树tree，当前的路径长度path_length，初始值应传入0。
返回：最终的路径长度

    @scala.annotation.tailrec
    final def pathLength(x:Array[Double], tree:ITree, path_length:Int): Double ={
        tree match{ //match方法，让tree进行如下两种模式匹配
            //如果ITree匹配到的类型是叶子节点，那么，查看该节点的样本数size，如果size大于1，则加上该size对应的二叉搜索树的平均路径长度，如果size等于1，则直接加1
            case ITreeLeaf(size) => 
                if (size > 1)
                    path_length + cost(size)
                else 
                    path_length + 1

            //如果ITree匹配到的类型是一颗分支子树，该子树还会有left分支，right分支，以及分类的依据特征列split_column，和该特征列的分割值split_value
            case ITreeBranch(left, right, split_column, split_value) => 
                val sample_value = x(split_column)  //传入的样本x在该特征上的取值

                if (sample_value < split_value)  //如果小于分割值则在左子树上进行递归计算，如果大于分割值则在右子树上进行递归计算
                    pathLength(x, left, path_length + 1)
                else
                    pathLength(x, right, path_length + 1)
        }
    }
}

5、读取数据进行预测

本节定义最终要调用运行的main方法，我把样例数据放在了本地，也可以放到hdfs上，csv格式，已经做好了标准化，概览如下

5.1、一些对spark的基本设置

object Runner{
    def main(args:Array[String]): Unit ={
        Random.setSeed(1337)

        val conf = new SparkConf()
            .setAppName("IsolationTree")
            .setMaster("local")

        val sc = new SparkContext(conf)
        //禁止对输出文件进行压缩
        sc.hadoopConfiguration.set("mapred.output.compress", "false")  

5.2、读入csv数据并预处理，lines为RDD格式，这是spark处理数据的基本单元

        val lines = sc.textFile("file:///tmp/spark_data/spark_if_train.csv") //本地路径
        val data =  //对每一行数据以逗号为分隔符进行拆分，从第二个数据开始取，因为第一个数字是索引
            lines
                .map(line => line.split(","))
                .map(s => s.slice(1,s.length))
        val header = data.first() // 取第一行的数据作为列名

        // 去掉列名行并将数据转化为double类型
        val rows = data.filter(line => line(0) != header(0)).map(s => s.map(_.toDouble)) 

        println("Loaded CSV File...")
        println(header.mkString("\n"))  // 看一下列名
        println(rows.take(5).deep.mkString("\n"))  // 看一下前5行数据

5.3、进行iforest的构建和对样本的预测

        // 构建森林，训练数据rows，森林里树的棵树，这里写10，数据量大的话一般是100
        val forest = IsolationForest.buildForest(rows, numTrees=10)

        // 对每一行数据进行预测
        val result_rdd = rows.map(row => row ++ Array(forest.predict(row)))

        // 将结果存入本地文件
        result_rdd.map(lines => lines.mkString(",")).repartition(1).saveAsTextFile("file:///tmp/predict_label")

        // 看一下前10条数据的预测结果
        val local_rows = rows.take(10)
        for(row <- local_rows){
            println("ForestScore", forest.predict(row))
        }
        println("Finished Isolation")
    }
}

以上，isolation forest训练部分和预测部分都做好了。

三、部署到spark上并运行

（图片给自己的机器打了码，略丑😢）

1、基础环境配置

前提1：配置好spark集群，能成功进入下图所示的交互状态。此部分教程自行google～

$ spark-shell

前提2：配置好sbt ，用于管理项目依赖，构建项目
参考教程：http://blog.csdn.net/zcf1002797280/article/details/49677881

sbt sbtVersion

2、部署脚本

2.1、将上节代码文件命名为Runner.scala
2.2、创建目录结构

cd ~
mkdir -p mysparkapp/iforest_model/src/main/scala

2.3、将Runner.scala移动到~/mysparkapp/iforest_model/src/main/scala文件夹下

mv Runner.scala ~/mysparkapp/iforest_model/src/main/scala/

2.4、新建配置文件conf.sbt，声明我们项目的名称以及对相关版本的依赖信息

cd ~/mysparkapp/iforest_model
vim conf.sbt

在conf.sbt中，添加如下内容，版本信息根据你配置的真实信息来写哦：

name := "IsolationForest"
version := "1.0"
scalaVersion := "2.11.8"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.2.1"

现在看一下我们的项目结构是否如图所示

find .

2.5、将程序打包，仍然在~/mysparkapp/iforest_model下，执行：

sbt package

注意黄色箭头指向的文件地址，这是打包好的jar包，供我们稍后提交任务使用。

2.6、正式提交spark任务
在提交spark任务之前，要确保输出目录不存在：

rm -r /tmp/predict_label

然后用spark-submit命令提交任务，需要传入刚刚打包好的jar包路径：

spark-submit --class "Runner" ~/mysparkapp/iforest_model/target/scala-2.11/isolationforest_2.11-1.0.jar

开始运行～～🤗️
我们打印出了数据的列名、前5条数据、以及前10条数据的异常得分如图所示：

任务执行完毕，看一下输出文件，图示捞出了前五行，最后一个字段即为预测得分，接下来就可以设定一个阈值，原作论文推荐为0.6，大于阈值的即判定为异常啦。

图中的第二行数据，得分0.69，其他数据得分均为0.5以下，观察一下它前面的字段，比其他数据都要大出很多，确实为一个异常点～

四、小结

isolation forest由多棵树构成，而树的生长过程并不受其他树影响，所以是一个非常完美的适合分布式并行的算法。样例数据和代码都放到了https://github.com/scarlettgin/isolation_spark