2小时入门Spark之MLlib.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T06:53:16.533008Z",
     "start_time": "2019-09-30T06:53:16.355Z"
    }
   },
   "source": [
    "## 2小时入门Spark之MLlib"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "MLlib是Spark的机器学习库，包括以下主要功能。\n",
    "\n",
    "* 实用工具：线性代数，统计，数据处理等工具\n",
    "* 特征工程：特征提取，特征转换，特征选择\n",
    "* 常用算法：分类，回归，聚类，协同过滤，降维\n",
    "* 模型优化：模型评估，参数优化。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "MLlib库包括两个不同的部分\n",
    "\n",
    "spark.mllib 包含基于rdd的机器学习算法API，目前不再更新，在3.0版本后将会丢弃，不建议使用。\n",
    "\n",
    "spark.ml 包含基于DataFrame的机器学习算法API，可以用来构建机器学习工作流Pipeline，推荐使用。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T02:18:17.140760Z",
     "start_time": "2019-10-08T02:18:12.746Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rdd = ParallelCollectionRDD[0] at parallelize at <console>:27\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "100"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rdd = sc.parallelize(1 to 100)\n",
    "rdd.count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Marking org.apache.spark:spark_mllib_2.11:2.3.1 for download\n",
      "-> Failed to resolve org.apache.spark:spark_mllib_2.11:2.3.1\n",
      "    -> not found: /var/folders/bs/pw656yts35qb1dpr1myl3wd80000gn/T/toree-tmp-dir1813761629305680892/toree_add_deps/cache/org.apache.spark/spark_mllib_2.11/ivy-2.3.1.xml\n",
      "    -> not found: https://repo1.maven.org/maven2/org/apache/spark/spark_mllib_2.11/2.3.1/spark_mllib_2.11-2.3.1.pom\n",
      "Obtained 0 files\n"
     ]
    }
   ],
   "source": [
    "%AddDeps org.apache.spark spark_mllib_2.11 2.3.1  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 一，MLlib基本概念"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "DataFrame: MLlib中数据的存储形式，其列可以存储特征向量，标签，以及原始的文本，图像。\n",
    "\n",
    "Transformer：转换器。具有transform方法。通过附加一个或多个列将一个DataFrame转换成另外一个DataFrame。\n",
    "\n",
    "Estimator：估计器。具有fit方法。它接受一个DataFrame数据作为输入后经过训练，产生一个转换器Transformer。\n",
    "\n",
    "Pipeline：流水线。具有setStages方法。顺序将多个Transformer和1个Estimator串联起来，得到一个流水线模型。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 二，Pipeline流水线范例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "任务描述：用逻辑回归模型预测句子中是否包括”spark“这个单词。\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T02:18:21.413905Z",
     "start_time": "2019-10-08T02:18:21.293Z"
    }
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.sql.SparkSession\n",
    "import org.apache.spark.ml.feature._\n",
    "import org.apache.spark.ml.classification.LogisticRegression\n",
    "import org.apache.spark.ml.evaluation.{MulticlassClassificationEvaluator,BinaryClassificationEvaluator}\n",
    "import org.apache.spark.ml.{Pipeline,PipelineModel}\n",
    "import org.apache.spark.ml.linalg.Vector\n",
    "import org.apache.spark.sql.Row\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T02:18:25.196060Z",
     "start_time": "2019-10-08T02:18:23.540Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "spark = org.apache.spark.sql.SparkSession@717bac32\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "org.apache.spark.sql.SparkSession@717bac32"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val spark = SparkSession.builder()\n",
    "   .master(\"local[4]\").appName(\"ml\")\n",
    "   .getOrCreate()\n",
    "\n",
    "import spark.implicits._"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**1，准备数据**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T09:15:04.899205Z",
     "start_time": "2019-09-30T09:15:04.513Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------------+-----+\n",
      "| id|             text|label|\n",
      "+---+-----------------+-----+\n",
      "|  0|  a b c d e spark|  1.0|\n",
      "|  1|            a c f|  0.0|\n",
      "|  2|spark hello world|  1.0|\n",
      "|  3| hadoop mapreduce|  0.0|\n",
      "|  4|     I love spark|  1.0|\n",
      "|  5|         big data|  0.0|\n",
      "+---+-----------------+-----+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dftrain = [id: bigint, text: string ... 1 more field]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[id: bigint, text: string ... 1 more field]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val dftrain = Seq((0L,\"a b c d e spark\",1.0),\n",
    "                (1L,\"a c f\",0.0),\n",
    "                (2L,\"spark hello world\",1.0),\n",
    "                (3L,\"hadoop mapreduce\",0.0),\n",
    "                (4L,\"I love spark\", 1.0),\n",
    "                (5L,\"big data\",0.0)).toDF(\"id\",\"text\",\"label\")\n",
    "dftrain.show"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**2，构建模型**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T11:05:48.275051Z",
     "start_time": "2019-09-30T11:05:47.425Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "class org.apache.spark.ml.feature.Tokenizer\n",
      "class org.apache.spark.ml.feature.HashingTF\n",
      "class org.apache.spark.ml.classification.LogisticRegression\n",
      "class org.apache.spark.ml.Pipeline\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "tokenizer = tok_adaedd5a0732\n",
       "hashingTF = hashingTF_7142c60ae88f\n",
       "lr = logreg_3d5b60586b76\n",
       "pipe = pipeline_4a935e4be5f1\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "pipeline_4a935e4be5f1"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val tokenizer = new Tokenizer().setInputCol(\"text\").setOutputCol(\"words\")\n",
    "println(tokenizer.getClass)\n",
    "\n",
    "val hashingTF = new HashingTF().setNumFeatures(100)\n",
    "   .setInputCol(tokenizer.getOutputCol)\n",
    "   .setOutputCol(\"features\")\n",
    "println(hashingTF.getClass)\n",
    "\n",
    "val lr = new LogisticRegression().setLabelCol(\"label\")\n",
    "//println(lr.explainParams)\n",
    "lr.setFeaturesCol(\"features\").setMaxIter(10).setRegParam(0.01)\n",
    "println(lr.getClass)\n",
    "\n",
    "val pipe = new Pipeline().setStages(Array(tokenizer,hashingTF,lr))\n",
    "println(pipe.getClass)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**3，训练模型**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T09:21:10.812706Z",
     "start_time": "2019-09-30T09:21:10.123Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "class org.apache.spark.ml.PipelineModel"
     ]
    },
    {
     "data": {
      "text/plain": [
       "model = pipeline_4a935e4be5f1\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "pipeline_4a935e4be5f1"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val model = pipe.fit(dftrain)\n",
    "print(model.getClass)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T09:08:05.789848Z",
     "start_time": "2019-09-30T09:08:05.703Z"
    }
   },
   "source": [
    "**4，使用模型**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T09:27:35.556496Z",
     "start_time": "2019-09-30T09:27:34.887Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+-----+\n",
      "| id|                text|label|\n",
      "+---+--------------------+-----+\n",
      "|  7|           spark job|  1.0|\n",
      "|  9|         hello world|  0.0|\n",
      "| 10|           a b c d e|  0.0|\n",
      "| 11|      you can you up|  0.0|\n",
      "| 12|spark is easy to ...|  1.0|\n",
      "+---+--------------------+-----+\n",
      "\n",
      "+--------------------+--------------------+--------------------+----------+\n",
      "|                text|            features|         probability|prediction|\n",
      "+--------------------+--------------------+--------------------+----------+\n",
      "|           spark job|(100,[5,70],[1.0,...|[0.35046042897666...|       1.0|\n",
      "|         hello world|(100,[48,50],[1.0...|[0.33560921515506...|       1.0|\n",
      "|           a b c d e|(100,[22,61,70,78...|[0.19082246657270...|       1.0|\n",
      "|      you can you up|(100,[25,28,33],[...|[0.81519423235139...|       0.0|\n",
      "|spark is easy to ...|(100,[5,21,60,81,...|[0.47768327161200...|       1.0|\n",
      "+--------------------+--------------------+--------------------+----------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dftest = [id: bigint, text: string ... 1 more field]\n",
       "dfresult = [id: bigint, text: string ... 6 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[id: bigint, text: string ... 6 more fields]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val dftest = Seq((7L,\"spark job\",1.0),(9L,\"hello world\",0.0),\n",
    "                 (10L,\"a b c d e\",0.0),(11L,\"you can you up\",0.0),\n",
    "                (12L,\"spark is easy to use.\",1.0)).toDF(\"id\",\"text\",\"label\")\n",
    "dftest.show\n",
    "\n",
    "val dfresult = model.transform(dftest)\n",
    "\n",
    "dfresult.selectExpr(\"text\",\"features\",\"probability\",\"prediction\").show"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T09:08:10.813260Z",
     "start_time": "2019-09-30T09:08:10.532Z"
    }
   },
   "source": [
    "**5，评估模型**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: long (nullable = false)\n",
      " |-- text: string (nullable = true)\n",
      " |-- label: double (nullable = false)\n",
      " |-- words: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- features: vector (nullable = true)\n",
      " |-- rawPrediction: vector (nullable = true)\n",
      " |-- probability: vector (nullable = true)\n",
      " |-- prediction: double (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfresult.printSchema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T09:36:42.815369Z",
     "start_time": "2019-09-30T09:36:42.376Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "accuracy = 0.6\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "evaluator = mcEval_3be0310cbf8f\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "mcEval_3be0310cbf8f"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val evaluator = new MulticlassClassificationEvaluator().setMetricName(\"accuracy\")\n",
    "    .setPredictionCol(\"prediction\").setLabelCol(\"label\")\n",
    "\n",
    "//println(evaluator.explainParams())\n",
    "\n",
    "evaluator.evaluate(dfresult)\n",
    "\n",
    "println(s\"\\naccuracy = ${evaluator.evaluate(dfresult)}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**6，保存模型**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T11:11:36.792176Z",
     "start_time": "2019-09-30T11:11:36.206Z"
    }
   },
   "outputs": [],
   "source": [
    "model.write.overwrite().save(\"mymodel.model\")\n",
    "\n",
    "// Now we can optionally save the fitted pipeline to disk\n",
    "//model.write.overwrite().save(\"spark-logistic-regression-model\")\n",
    "\n",
    "// We can also save this unfit pipeline to disk\n",
    "//pipeline.write.overwrite().save(\"unfit-lr-model\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T09:52:05.433590Z",
     "start_time": "2019-09-30T09:52:04.533Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+----------+\n",
      "|                text|label|prediction|\n",
      "+--------------------+-----+----------+\n",
      "|           spark job|  1.0|       1.0|\n",
      "|         hello world|  0.0|       1.0|\n",
      "|           a b c d e|  0.0|       1.0|\n",
      "|      you can you up|  0.0|       0.0|\n",
      "|spark is easy to ...|  1.0|       1.0|\n",
      "+--------------------+-----+----------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "model_loaded = pipeline_4a935e4be5f1\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "pipeline_4a935e4be5f1"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//重新载入模型\n",
    "val model_loaded = PipelineModel.load(\"mymodel.model\")\n",
    "model_loaded.transform(dftest).select(\"text\",\"label\",\"prediction\").show"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 三，导入数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "可以使用spark.read导入csv，image，libsvm，txt等格式数据。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T11:52:35.459487Z",
     "start_time": "2019-09-30T11:52:35.114Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- image: struct (nullable = true)\n",
      " |    |-- origin: string (nullable = true)\n",
      " |    |-- height: integer (nullable = true)\n",
      " |    |-- width: integer (nullable = true)\n",
      " |    |-- nChannels: integer (nullable = true)\n",
      " |    |-- mode: integer (nullable = true)\n",
      " |    |-- data: binary (nullable = true)\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dfimage = [image: struct<origin: string, height: int ... 4 more fields>]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[image: struct<origin: string, height: int ... 4 more fields>]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//导入图片\n",
    "\n",
    "val dfimage = spark.read.format(\"image\")\n",
    "    .option(\"dropInvalid\", true).load(\"data/mllib/images/origin/kittens\")\n",
    "dfimage.printSchema\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-09-30T11:52:46.987458Z",
     "start_time": "2019-09-30T11:52:40.230Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----+---------+----+--------------------+\n",
      "|              origin|height|width|nChannels|mode|                data|\n",
      "+--------------------+------+-----+---------+----+--------------------+\n",
      "|file:///Users/lia...|   311|  300|        3|  16|[C1 C1 C1 C2 C2 C...|\n",
      "|file:///Users/lia...|   313|  199|        3|  16|[D0 E5 ED CA DF E...|\n",
      "|file:///Users/lia...|   200|  300|        3|  16|[58 5D 60 58 5D 6...|\n",
      "|file:///Users/lia...|   296|  300|        3|  16|[CB E6 F4 CA E5 F...|\n",
      "+--------------------+------+-----+---------+----+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfimage.selectExpr(\"image.*\").show"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 四，特征工程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "spark的特征处理功能主要在 spark.ml.feature 模块中，包括以下一些功能。\n",
    "\n",
    "* 特征提取：Tf-idf, Word2Vec, CountVectorizer, FeatureHasher\n",
    "\n",
    "* 特征转换：OneHotEncoderEstimator, Normalizer, Imputer(缺失值填充), StandardScaler, MinMaxScaler, Tokenizer(构建词典), \n",
    "  StopWordsRemover, SQLTransformer, Bucketizer, Interaction(交叉项), Binarizer(二值化), n-gram,……\n",
    "\n",
    "* 特征选择：VectorSlicer(向量切片), RFormula, ChiSqSelector(卡方检验)\n",
    "\n",
    "* LSH转换：局部敏感哈希广泛用于海量数据中求最邻近，聚类等算法。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**1，CountVectorizer**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "CountVectorizer可以提取文本中的词频特征。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---------------+--------------------+\n",
      "| id|          words|            features|\n",
      "+---+---------------+--------------------+\n",
      "|  0|      [a, b, c]|(3,[0,1,2],[1.0,1...|\n",
      "|  1|[a, b, b, c, a]|(3,[0,1,2],[2.0,2...|\n",
      "+---+---------------+--------------------+\n",
      "\n",
      "+---+---------------+--------------------+\n",
      "| id|          words|            features|\n",
      "+---+---------------+--------------------+\n",
      "|  0|      [a, b, c]|(3,[0,1,2],[1.0,1...|\n",
      "|  1|[a, b, b, c, a]|(3,[0,1,2],[2.0,2...|\n",
      "+---+---------------+--------------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "df = [id: int, words: array<string>]\n",
       "cvModel = cntVec_324b7869593e\n",
       "cvm = cntVecModel_4947f6142271\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "cntVecModel_4947f6142271"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.{CountVectorizer, CountVectorizerModel}\n",
    "\n",
    "val df = spark.createDataFrame(Seq(\n",
    "  (0, Array(\"a\", \"b\", \"c\")),\n",
    "  (1, Array(\"a\", \"b\", \"b\", \"c\", \"a\"))\n",
    ")).toDF(\"id\", \"words\")\n",
    "\n",
    "// fit a CountVectorizerModel from the corpus\n",
    "val cvModel: CountVectorizerModel = new CountVectorizer()\n",
    "  .setInputCol(\"words\")\n",
    "  .setOutputCol(\"features\")\n",
    "  .setVocabSize(3)\n",
    "  .setMinDF(2)\n",
    "  .fit(df)\n",
    "\n",
    "// alternatively, define CountVectorizerModel with a-priori vocabulary\n",
    "val cvm = new CountVectorizerModel(Array(\"a\", \"b\", \"c\"))\n",
    "  .setInputCol(\"words\")\n",
    "  .setOutputCol(\"features\")\n",
    "\n",
    "cvModel.transform(df).show()\n",
    "\n",
    "cvm.transform(df).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**2，Tf-IDF**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Tf-IDF可以降低文本频率过高的常用词的权重。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------------+\n",
      "|label|            features|\n",
      "+-----+--------------------+\n",
      "|  0.0|(20,[0,5,9,17],[0...|\n",
      "|  0.0|(20,[2,7,9,13,15]...|\n",
      "|  1.0|(20,[4,6,13,15,18...|\n",
      "+-----+--------------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "sentenceData = [label: double, sentence: string]\n",
       "tokenizer = tok_a56305ad5983\n",
       "wordsData = [label: double, sentence: string ... 1 more field]\n",
       "hashingTF = hashingTF_3041530f877f\n",
       "featurizedData = [label: double, sentence: string ... 2 more fields]\n",
       "idf = idf_512ac68468f2\n",
       "idfModel = idf_512ac68468f2\n",
       "rescaledData = [label: double, sentence: string ... 3 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[label: double, sentence: string ... 3 more fields]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}\n",
    "\n",
    "val sentenceData = spark.createDataFrame(Seq(\n",
    "  (0.0, \"Hi I heard about Spark\"),\n",
    "  (0.0, \"I wish Java could use case classes\"),\n",
    "  (1.0, \"Logistic regression models are neat\")\n",
    ")).toDF(\"label\", \"sentence\")\n",
    "\n",
    "val tokenizer = new Tokenizer().setInputCol(\"sentence\").setOutputCol(\"words\")\n",
    "val wordsData = tokenizer.transform(sentenceData)\n",
    "\n",
    "val hashingTF = new HashingTF()\n",
    "  .setInputCol(\"words\").setOutputCol(\"rawFeatures\").setNumFeatures(20)\n",
    "\n",
    "val featurizedData = hashingTF.transform(wordsData)\n",
    "// alternatively, CountVectorizer can also be used to get term frequency vectors\n",
    "\n",
    "val idf = new IDF().setInputCol(\"rawFeatures\").setOutputCol(\"features\")\n",
    "val idfModel = idf.fit(featurizedData)   \n",
    "\n",
    "val rescaledData = idfModel.transform(featurizedData)\n",
    "rescaledData.select(\"label\", \"features\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**3，Word2Vec**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Word2Vec可以使用浅层神经网络提取文本中词的相似语义信息。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Text: [Hi, I, heard, about, Spark] => \n",
      "Vector: [-0.008142343163490296,0.02051363289356232,0.03255096450448036]\n",
      "\n",
      "Text: [I, wish, Java, could, use, case, classes] => \n",
      "Vector: [0.043090314205203734,0.035048123182994974,0.023512658663094044]\n",
      "\n",
      "Text: [Logistic, regression, models, are, neat] => \n",
      "Vector: [0.038572299480438235,-0.03250147425569594,-0.01552378609776497]\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "documentDF = [text: array<string>]\n",
       "word2Vec = w2v_478394473f71\n",
       "model = w2v_478394473f71\n",
       "result = [text: array<string>, result: vector]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[text: array<string>, result: vector]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.Word2Vec\n",
    "import org.apache.spark.ml.linalg.Vector\n",
    "import org.apache.spark.sql.Row\n",
    "\n",
    "val documentDF = spark.createDataFrame(Seq(\n",
    "  \"Hi I heard about Spark\".split(\" \"),\n",
    "  \"I wish Java could use case classes\".split(\" \"),\n",
    "  \"Logistic regression models are neat\".split(\" \")\n",
    ").map(Tuple1.apply)).toDF(\"text\")\n",
    "\n",
    "val word2Vec = new Word2Vec()\n",
    "  .setInputCol(\"text\")\n",
    "  .setOutputCol(\"result\")\n",
    "  .setVectorSize(3)\n",
    "  .setMinCount(0)\n",
    "val model = word2Vec.fit(documentDF)\n",
    "\n",
    "val result = model.transform(documentDF)\n",
    "result.collect().foreach { case Row(text: Seq[_], features: Vector) =>\n",
    "  println(s\"Text: [${text.mkString(\", \")}] => \\nVector: $features\\n\") }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**4， OnHotEncoderEstimator**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "OneHotEncoderEstimator可以将类别特征转换成OneHot编码。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+--------------+-------------+-------------+\n",
      "|categoryIndex1|categoryIndex2| categoryVec1| categoryVec2|\n",
      "+--------------+--------------+-------------+-------------+\n",
      "|           0.0|           1.0|(2,[0],[1.0])|(2,[1],[1.0])|\n",
      "|           1.0|           0.0|(2,[1],[1.0])|(2,[0],[1.0])|\n",
      "|           2.0|           1.0|    (2,[],[])|(2,[1],[1.0])|\n",
      "|           0.0|           2.0|(2,[0],[1.0])|    (2,[],[])|\n",
      "|           0.0|           1.0|(2,[0],[1.0])|(2,[1],[1.0])|\n",
      "|           2.0|           0.0|    (2,[],[])|(2,[0],[1.0])|\n",
      "+--------------+--------------+-------------+-------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "df = [categoryIndex1: double, categoryIndex2: double]\n",
       "encoder = oneHotEncoder_53d7e1227a6e\n",
       "model = oneHotEncoder_53d7e1227a6e\n",
       "encoded = [categoryIndex1: double, categoryIndex2: double ... 2 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[categoryIndex1: double, categoryIndex2: double ... 2 more fields]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.OneHotEncoderEstimator\n",
    "\n",
    "val df = spark.createDataFrame(Seq(\n",
    "  (0.0, 1.0),\n",
    "  (1.0, 0.0),\n",
    "  (2.0, 1.0),\n",
    "  (0.0, 2.0),\n",
    "  (0.0, 1.0),\n",
    "  (2.0, 0.0)\n",
    ")).toDF(\"categoryIndex1\", \"categoryIndex2\")\n",
    "\n",
    "val encoder = new OneHotEncoderEstimator()\n",
    "  .setInputCols(Array(\"categoryIndex1\", \"categoryIndex2\"))\n",
    "  .setOutputCols(Array(\"categoryVec1\", \"categoryVec2\"))\n",
    "val model = encoder.fit(df)\n",
    "\n",
    "val encoded = model.transform(df)\n",
    "encoded.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**5，FeatureHasher**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当特征数量过多时，可以用FeatureHasher来进行降维。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+---------+------+--------------------------------------------------------+\n",
      "|real|bool |stringNum|string|features                                                |\n",
      "+----+-----+---------+------+--------------------------------------------------------+\n",
      "|2.2 |true |1        |foo   |(262144,[174475,247670,257907,262126],[2.2,1.0,1.0,1.0])|\n",
      "|3.3 |false|2        |bar   |(262144,[70644,89673,173866,174475],[1.0,1.0,1.0,3.3])  |\n",
      "|4.4 |false|3        |baz   |(262144,[22406,70644,174475,187923],[1.0,1.0,4.4,1.0])  |\n",
      "|5.5 |false|4        |foo   |(262144,[70644,101499,174475,257907],[1.0,1.0,5.5,1.0]) |\n",
      "+----+-----+---------+------+--------------------------------------------------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dataset = [real: double, bool: boolean ... 2 more fields]\n",
       "hasher = featureHasher_c4a5311853be\n",
       "featurized = [real: double, bool: boolean ... 3 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[real: double, bool: boolean ... 3 more fields]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.FeatureHasher\n",
    "\n",
    "val dataset = spark.createDataFrame(Seq(\n",
    "  (2.2, true, \"1\", \"foo\"),\n",
    "  (3.3, false, \"2\", \"bar\"),\n",
    "  (4.4, false, \"3\", \"baz\"),\n",
    "  (5.5, false, \"4\", \"foo\")\n",
    ")).toDF(\"real\", \"bool\", \"stringNum\", \"string\")\n",
    "\n",
    "val hasher = new FeatureHasher()\n",
    "  .setInputCols(\"real\", \"bool\", \"stringNum\", \"string\")\n",
    "  .setOutputCol(\"features\")\n",
    "\n",
    "val featurized = hasher.transform(dataset)\n",
    "featurized.show(false)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**6, IndexToString 和 StringIndexer**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "StringIndexer和IndexToString互为逆操作。\n",
    "\n",
    "前者可以将字符串表示的类别特征转换成数字表示的类别特征。\n",
    "\n",
    "后者可以将数字表示的类别特征转换成字符串表示的类别特征。\n",
    "\n",
    "两者一般会同时使用，先用StringIndexer将字符串编码成数字。\n",
    "\n",
    "经过模型转换后，再调用IndexToString将数字转换成字符串。\n",
    "\n",
    "StringIndexer是一个Estimator需要被fit。\n",
    "\n",
    "而IndexToString是一个Transformer可以直接使用。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Transformed string column 'category' to indexed column 'categoryIndex'\n",
      "+---+--------+-------------+\n",
      "| id|category|categoryIndex|\n",
      "+---+--------+-------------+\n",
      "|  0|       a|          0.0|\n",
      "|  1|       b|          2.0|\n",
      "|  2|       c|          1.0|\n",
      "|  3|       a|          0.0|\n",
      "|  4|       a|          0.0|\n",
      "|  5|       c|          1.0|\n",
      "+---+--------+-------------+\n",
      "\n",
      "StringIndexer will store labels in output column metadata: {\"vals\":[\"a\",\"c\",\"b\"],\"type\":\"nominal\",\"name\":\"categoryIndex\"}\n",
      "\n",
      "Transformed indexed column 'categoryIndex' back to original string column 'originalCategory' using labels in metadata\n",
      "+---+-------------+----------------+\n",
      "| id|categoryIndex|originalCategory|\n",
      "+---+-------------+----------------+\n",
      "|  0|          0.0|               a|\n",
      "|  1|          2.0|               b|\n",
      "|  2|          1.0|               c|\n",
      "|  3|          0.0|               a|\n",
      "|  4|          0.0|               a|\n",
      "|  5|          1.0|               c|\n",
      "+---+-------------+----------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "df = [id: int, category: string]\n",
       "indexer = strIdx_ea624b6a7998\n",
       "indexed = [id: int, category: string ... 1 more field]\n",
       "inputColSchema = StructField(categoryIndex,DoubleType,false)\n",
       "converter = idxToStr_d63f9da12729\n",
       "converted = [id: int, category: string ... 2 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[id: int, category: string ... 2 more fields]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.attribute.Attribute\n",
    "import org.apache.spark.ml.feature.{IndexToString, StringIndexer}\n",
    "\n",
    "val df = spark.createDataFrame(Seq(\n",
    "  (0, \"a\"),\n",
    "  (1, \"b\"),\n",
    "  (2, \"c\"),\n",
    "  (3, \"a\"),\n",
    "  (4, \"a\"),\n",
    "  (5, \"c\")\n",
    ")).toDF(\"id\", \"category\")\n",
    "\n",
    "val indexer = new StringIndexer()\n",
    "  .setInputCol(\"category\")\n",
    "  .setOutputCol(\"categoryIndex\")\n",
    "  .fit(df)\n",
    "val indexed = indexer.transform(df)\n",
    "\n",
    "println(s\"Transformed string column '${indexer.getInputCol}' \" +\n",
    "    s\"to indexed column '${indexer.getOutputCol}'\")\n",
    "indexed.show()\n",
    "\n",
    "val inputColSchema = indexed.schema(indexer.getOutputCol)\n",
    "println(s\"StringIndexer will store labels in output column metadata: \" +\n",
    "    s\"${Attribute.fromStructField(inputColSchema).toString}\\n\")\n",
    "\n",
    "val converter = new IndexToString()\n",
    "  .setInputCol(\"categoryIndex\")\n",
    "  .setOutputCol(\"originalCategory\")\n",
    "\n",
    "val converted = converter.transform(indexed)\n",
    "\n",
    "println(s\"Transformed indexed column '${converter.getInputCol}' back to original string \" +\n",
    "    s\"column '${converter.getOutputCol}' using labels in metadata\")\n",
    "converted.select(\"id\", \"categoryIndex\", \"originalCategory\").show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{\"vals\":[\"a\",\"c\",\"b\"],\"type\":\"nominal\",\"name\":\"categoryIndex\"}"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "indexed.schema(indexer.getOutputCol)\n",
    "\n",
    "Attribute.fromStructField(inputColSchema).toString"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**7，StandardScaler 正态标准化**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------------+--------------------+\n",
      "|label|            features|      scaledFeatures|\n",
      "+-----+--------------------+--------------------+\n",
      "|  0.0|(692,[127,128,129...|(692,[127,128,129...|\n",
      "|  1.0|(692,[158,159,160...|(692,[158,159,160...|\n",
      "|  1.0|(692,[124,125,126...|(692,[124,125,126...|\n",
      "|  1.0|(692,[152,153,154...|(692,[152,153,154...|\n",
      "|  1.0|(692,[151,152,153...|(692,[151,152,153...|\n",
      "+-----+--------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dataFrame = [label: double, features: vector]\n",
       "scaler = stdScal_c9687c90f418\n",
       "scalerModel = stdScal_c9687c90f418\n",
       "scaledData = [label: double, features: vector ... 1 more field]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[label: double, features: vector ... 1 more field]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.StandardScaler\n",
    "\n",
    "val dataFrame = spark.read.format(\"libsvm\")\n",
    "    .load(\"data/mllib/sample_libsvm_data.txt\")\n",
    "\n",
    "val scaler = new StandardScaler()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"scaledFeatures\")\n",
    "  .setWithStd(true)\n",
    "  .setWithMean(false)\n",
    "\n",
    "\n",
    "val scalerModel = scaler.fit(dataFrame)\n",
    "\n",
    "\n",
    "val scaledData = scalerModel.transform(dataFrame)\n",
    "scaledData.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**8, MinMax标准化**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Features scaled to range: [0.0, 1.0]\n",
      "+--------------+--------------+\n",
      "|      features|scaledFeatures|\n",
      "+--------------+--------------+\n",
      "|[1.0,0.1,-1.0]| [0.0,0.0,0.0]|\n",
      "| [2.0,1.1,1.0]| [0.5,0.1,0.5]|\n",
      "|[3.0,10.1,3.0]| [1.0,1.0,1.0]|\n",
      "+--------------+--------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "df = [id: int, features: vector]\n",
       "scaler = minMaxScal_c35a76973fd4\n",
       "scalerModel = minMaxScal_c35a76973fd4\n",
       "scaledData = [id: int, features: vector ... 1 more field]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[id: int, features: vector ... 1 more field]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.MinMaxScaler\n",
    "import org.apache.spark.ml.linalg.Vectors\n",
    "\n",
    "val df = spark.createDataFrame(Seq(\n",
    "  (0, Vectors.dense(1.0, 0.1, -1.0)),\n",
    "  (1, Vectors.dense(2.0, 1.1, 1.0)),\n",
    "  (2, Vectors.dense(3.0, 10.1, 3.0))\n",
    ")).toDF(\"id\", \"features\")\n",
    "\n",
    "val scaler = new MinMaxScaler()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"scaledFeatures\")\n",
    "\n",
    "// Compute summary statistics and generate MinMaxScalerModel\n",
    "val scalerModel = scaler.fit(df)\n",
    "\n",
    "// rescale each feature to range [min, max].\n",
    "val scaledData = scalerModel.transform(df)\n",
    "println(s\"Features scaled to range: [${scaler.getMin}, ${scaler.getMax}]\")\n",
    "scaledData.select(\"features\", \"scaledFeatures\").show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**9，MaxAbsScaler标准化**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+----------------+\n",
      "|      features|  scaledFeatures|\n",
      "+--------------+----------------+\n",
      "|[1.0,0.1,-8.0]|[0.25,0.01,-1.0]|\n",
      "|[2.0,1.0,-4.0]|  [0.5,0.1,-0.5]|\n",
      "|[4.0,10.0,8.0]|   [1.0,1.0,1.0]|\n",
      "+--------------+----------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "df = [id: int, features: vector]\n",
       "scaler = maxAbsScal_8f99413b2cdd\n",
       "scalerModel = maxAbsScal_8f99413b2cdd\n",
       "scaledData = [id: int, features: vector ... 1 more field]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[id: int, features: vector ... 1 more field]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.MaxAbsScaler\n",
    "import org.apache.spark.ml.linalg.Vectors\n",
    "\n",
    "val df = spark.createDataFrame(Seq(\n",
    "  (0, Vectors.dense(1.0, 0.1, -8.0)),\n",
    "  (1, Vectors.dense(2.0, 1.0, -4.0)),\n",
    "  (2, Vectors.dense(4.0, 10.0, 8.0))\n",
    ")).toDF(\"id\", \"features\")\n",
    "\n",
    "val scaler = new MaxAbsScaler()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"scaledFeatures\")\n",
    "\n",
    "// Compute summary statistics and generate MaxAbsScalerModel\n",
    "val scalerModel = scaler.fit(df)\n",
    "\n",
    "// rescale each feature to range [-1, 1]\n",
    "val scaledData = scalerModel.transform(df)\n",
    "scaledData.select(\"features\", \"scaledFeatures\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**10，SQLTransformer**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "可以使用SQL语法将DataFrame进行转换，等效于注册表的作用。\n",
    "\n",
    "但它可以用于Pipeline中作为Transformer."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+---+---+----+\n",
      "| id| v1| v2| v3|  v4|\n",
      "+---+---+---+---+----+\n",
      "|  0|1.0|3.0|4.0| 3.0|\n",
      "|  2|2.0|5.0|7.0|10.0|\n",
      "+---+---+---+---+----+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "df = [id: int, v1: double ... 1 more field]\n",
       "sqlTrans = sql_0bf0e28b93cd\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "sql_0bf0e28b93cd"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.SQLTransformer\n",
    "\n",
    "val df = spark.createDataFrame(\n",
    "  Seq((0, 1.0, 3.0), (2, 2.0, 5.0))).toDF(\"id\", \"v1\", \"v2\")\n",
    "\n",
    "val sqlTrans = new SQLTransformer().setStatement(\n",
    "  \"SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM __THIS__\")\n",
    "\n",
    "sqlTrans.transform(df).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**11, Imputer**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Imputer转换器可以填充缺失值，缺失值可以用 Double.NaN来表示。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+-----+-----+\n",
      "|  a|  b|out_a|out_b|\n",
      "+---+---+-----+-----+\n",
      "|1.0|NaN|  1.0|  4.0|\n",
      "|2.0|NaN|  2.0|  4.0|\n",
      "|NaN|3.0|  3.0|  3.0|\n",
      "|4.0|4.0|  4.0|  4.0|\n",
      "|5.0|5.0|  5.0|  5.0|\n",
      "+---+---+-----+-----+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "df = [a: double, b: double]\n",
       "imputer = imputer_e1ab83578e21\n",
       "model = imputer_e1ab83578e21\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "imputer_e1ab83578e21"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.Imputer\n",
    "\n",
    "val df = spark.createDataFrame(Seq(\n",
    "  (1.0, Double.NaN),\n",
    "  (2.0, Double.NaN),\n",
    "  (Double.NaN, 3.0),\n",
    "  (4.0, 4.0),\n",
    "  (5.0, 5.0)\n",
    ")).toDF(\"a\", \"b\")\n",
    "\n",
    "val imputer = new Imputer()\n",
    "  .setInputCols(Array(\"a\", \"b\"))\n",
    "  .setOutputCols(Array(\"out_a\", \"out_b\"))\n",
    "\n",
    "val model = imputer.fit(df)\n",
    "model.transform(df).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**12 ChiSqSelector**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当label是离散值时，ChiSqSelector选择器可以根据Chi2检验统计量筛选特征。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ChiSqSelector output with top 1 features selected\n",
      "+---+------------------+-------+----------------+\n",
      "| id|          features|clicked|selectedFeatures|\n",
      "+---+------------------+-------+----------------+\n",
      "|  7|[0.0,0.0,18.0,1.0]|    1.0|          [18.0]|\n",
      "|  8|[0.0,1.0,12.0,0.0]|    0.0|          [12.0]|\n",
      "|  9|[1.0,0.0,15.0,0.1]|    0.0|          [15.0]|\n",
      "+---+------------------+-------+----------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = List((7,[0.0,0.0,18.0,1.0],1.0), (8,[0.0,1.0,12.0,0.0],0.0), (9,[1.0,0.0,15.0,0.1],0.0))\n",
       "df = [id: int, features: vector ... 1 more field]\n",
       "selector = chiSqSelector_b8d478ed2aee\n",
       "result = [id: int, features: vector ... 2 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[id: int, features: vector ... 2 more fields]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.ChiSqSelector\n",
    "import org.apache.spark.ml.linalg.Vectors\n",
    "\n",
    "val data = Seq(\n",
    "  (7, Vectors.dense(0.0, 0.0, 18.0, 1.0), 1.0),\n",
    "  (8, Vectors.dense(0.0, 1.0, 12.0, 0.0), 0.0),\n",
    "  (9, Vectors.dense(1.0, 0.0, 15.0, 0.1), 0.0)\n",
    ")\n",
    "\n",
    "val df = spark.createDataset(data).toDF(\"id\", \"features\", \"clicked\")\n",
    "\n",
    "val selector = new ChiSqSelector()\n",
    "  .setNumTopFeatures(1)\n",
    "  .setFeaturesCol(\"features\")\n",
    "  .setLabelCol(\"clicked\")\n",
    "  .setOutputCol(\"selectedFeatures\")\n",
    "\n",
    "val result = selector.fit(df).transform(df)\n",
    "\n",
    "println(s\"ChiSqSelector output with top ${selector.getNumTopFeatures} features selected\")\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**13，局部敏感哈希（LSH）**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Locality Sensitive Hashing 是一种广泛使用的哈希技巧，常用于求海量数据的最邻近，聚类和离群值检测等任务中。\n",
    "\n",
    "局部敏感哈希的特性是将距离较近的点以很大的概率通过哈希作用映射到相同的桶内，而将距离较远的点以很大的概率通过哈希作用映射到不同的桶中。\n",
    "\n",
    "BucketedRandomProjectionLSH 使用Euclidean distance作为距离度量，取点在某个随机方向的投影结果作为哈希值。\n",
    "\n",
    "MinHashLSH 使用Jaccard distance作为距离度量，它是衡量两个集合差异性的一个指标，MinHashLSH取集合中所有元素的哈希值的最小值作为哈希值。\n",
    "\n",
    "杰卡德距离计算公式如下:$$d(A,B)=1−\\frac{|A∩B|}{|A∪B|}$$\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The hashed dataset where hashed values are stored in the column 'hashes':\n",
      "+---+-----------+--------------------+\n",
      "| id|   features|              hashes|\n",
      "+---+-----------+--------------------+\n",
      "|  0|  [1.0,1.0]|[[0.0], [0.0], [-...|\n",
      "|  1| [1.0,-1.0]|[[-1.0], [-1.0], ...|\n",
      "|  2|[-1.0,-1.0]|[[-1.0], [-1.0], ...|\n",
      "|  3| [-1.0,1.0]|[[0.0], [0.0], [-...|\n",
      "+---+-----------+--------------------+\n",
      "\n",
      "Approximately joining dfA and dfB on Euclidean distance smaller than 1.5:\n",
      "+---+---+-----------------+\n",
      "|idA|idB|EuclideanDistance|\n",
      "+---+---+-----------------+\n",
      "|  1|  4|              1.0|\n",
      "|  0|  6|              1.0|\n",
      "|  1|  7|              1.0|\n",
      "|  3|  5|              1.0|\n",
      "|  0|  4|              1.0|\n",
      "|  3|  6|              1.0|\n",
      "|  2|  7|              1.0|\n",
      "|  2|  5|              1.0|\n",
      "+---+---+-----------------+\n",
      "\n",
      "Approximately searching dfA for 2 nearest neighbors of the key:\n",
      "+---+----------+--------------------+-------+\n",
      "| id|  features|              hashes|distCol|\n",
      "+---+----------+--------------------+-------+\n",
      "|  0| [1.0,1.0]|[[0.0], [0.0], [-...|    1.0|\n",
      "|  1|[1.0,-1.0]|[[-1.0], [-1.0], ...|    1.0|\n",
      "+---+----------+--------------------+-------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dfA = [id: int, features: vector]\n",
       "dfB = [id: int, features: vector]\n",
       "key = [1.0,0.0]\n",
       "brp = brp-lsh_9d4f769e11c9\n",
       "model = brp-lsh_9d4f769e11c9\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "brp-lsh_9d4f769e11c9"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.BucketedRandomProjectionLSH\n",
    "import org.apache.spark.ml.linalg.Vectors\n",
    "import org.apache.spark.sql.SparkSession\n",
    "import org.apache.spark.sql.functions.col\n",
    "\n",
    "val dfA = spark.createDataFrame(Seq(\n",
    "  (0, Vectors.dense(1.0, 1.0)),\n",
    "  (1, Vectors.dense(1.0, -1.0)),\n",
    "  (2, Vectors.dense(-1.0, -1.0)),\n",
    "  (3, Vectors.dense(-1.0, 1.0))\n",
    ")).toDF(\"id\", \"features\")\n",
    "\n",
    "val dfB = spark.createDataFrame(Seq(\n",
    "  (4, Vectors.dense(1.0, 0.0)),\n",
    "  (5, Vectors.dense(-1.0, 0.0)),\n",
    "  (6, Vectors.dense(0.0, 1.0)),\n",
    "  (7, Vectors.dense(0.0, -1.0))\n",
    ")).toDF(\"id\", \"features\")\n",
    "\n",
    "val key = Vectors.dense(1.0, 0.0)\n",
    "\n",
    "val brp = new BucketedRandomProjectionLSH()\n",
    "  .setBucketLength(2.0)\n",
    "  .setNumHashTables(3)\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"hashes\")\n",
    "\n",
    "val model = brp.fit(dfA)\n",
    "\n",
    "// Feature Transformation\n",
    "println(\"The hashed dataset where hashed values are stored in the column 'hashes':\")\n",
    "model.transform(dfA).show()\n",
    "\n",
    "// Compute the locality sensitive hashes for the input rows, then perform approximate\n",
    "// similarity join.\n",
    "// We could avoid computing hashes by passing in the already-transformed dataset, e.g.\n",
    "// `model.approxSimilarityJoin(transformedA, transformedB, 1.5)`\n",
    "println(\"Approximately joining dfA and dfB on Euclidean distance smaller than 1.5:\")\n",
    "model.approxSimilarityJoin(dfA, dfB, 1.5, \"EuclideanDistance\")\n",
    "  .select(col(\"datasetA.id\").alias(\"idA\"),\n",
    "    col(\"datasetB.id\").alias(\"idB\"),\n",
    "    col(\"EuclideanDistance\")).show()\n",
    "\n",
    "// Compute the locality sensitive hashes for the input rows, then perform approximate nearest\n",
    "// neighbor search.\n",
    "// We could avoid computing hashes by passing in the already-transformed dataset, e.g.\n",
    "// `model.approxNearestNeighbors(transformedA, key, 2)`\n",
    "println(\"Approximately searching dfA for 2 nearest neighbors of the key:\")\n",
    "model.approxNearestNeighbors(dfA, key, 2).show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The hashed dataset where hashed values are stored in the column 'hashes':\n",
      "+---+--------------------+--------------------+\n",
      "| id|            features|              hashes|\n",
      "+---+--------------------+--------------------+\n",
      "|  0|(6,[0,1,2],[1.0,1...|[[2.25592966E8], ...|\n",
      "|  1|(6,[2,3,4],[1.0,1...|[[2.25592966E8], ...|\n",
      "|  2|(6,[0,2,4],[1.0,1...|[[2.25592966E8], ...|\n",
      "+---+--------------------+--------------------+\n",
      "\n",
      "Approximately joining dfA and dfB on Jaccard distance smaller than 0.6:\n",
      "+---+---+---------------+\n",
      "|idA|idB|JaccardDistance|\n",
      "+---+---+---------------+\n",
      "|  1|  4|            0.5|\n",
      "|  0|  5|            0.5|\n",
      "|  1|  5|            0.5|\n",
      "|  2|  5|            0.5|\n",
      "+---+---+---------------+\n",
      "\n",
      "Approximately searching dfA for 2 nearest neighbors of the key:\n",
      "+---+--------------------+--------------------+-------+\n",
      "| id|            features|              hashes|distCol|\n",
      "+---+--------------------+--------------------+-------+\n",
      "|  1|(6,[2,3,4],[1.0,1...|[[2.25592966E8], ...|   0.75|\n",
      "+---+--------------------+--------------------+-------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dfA = [id: int, features: vector]\n",
       "dfB = [id: int, features: vector]\n",
       "key = (6,[1,3],[1.0,1.0])\n",
       "mh = mh-lsh_dee13c9ca2f9\n",
       "model = mh-lsh_dee13c9ca2f9\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "mh-lsh_dee13c9ca2f9"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.MinHashLSH\n",
    "import org.apache.spark.ml.linalg.Vectors\n",
    "import org.apache.spark.sql.SparkSession\n",
    "import org.apache.spark.sql.functions.col\n",
    "\n",
    "val dfA = spark.createDataFrame(Seq(\n",
    "  (0, Vectors.sparse(6, Seq((0, 1.0), (1, 1.0), (2, 1.0)))),\n",
    "  (1, Vectors.sparse(6, Seq((2, 1.0), (3, 1.0), (4, 1.0)))),\n",
    "  (2, Vectors.sparse(6, Seq((0, 1.0), (2, 1.0), (4, 1.0))))\n",
    ")).toDF(\"id\", \"features\")\n",
    "\n",
    "val dfB = spark.createDataFrame(Seq(\n",
    "  (3, Vectors.sparse(6, Seq((1, 1.0), (3, 1.0), (5, 1.0)))),\n",
    "  (4, Vectors.sparse(6, Seq((2, 1.0), (3, 1.0), (5, 1.0)))),\n",
    "  (5, Vectors.sparse(6, Seq((1, 1.0), (2, 1.0), (4, 1.0))))\n",
    ")).toDF(\"id\", \"features\")\n",
    "\n",
    "val key = Vectors.sparse(6, Seq((1, 1.0), (3, 1.0)))\n",
    "\n",
    "val mh = new MinHashLSH()\n",
    "  .setNumHashTables(5)\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"hashes\")\n",
    "\n",
    "val model = mh.fit(dfA)\n",
    "\n",
    "// Feature Transformation\n",
    "println(\"The hashed dataset where hashed values are stored in the column 'hashes':\")\n",
    "model.transform(dfA).show()\n",
    "\n",
    "// Compute the locality sensitive hashes for the input rows, then perform approximate\n",
    "// similarity join.\n",
    "// We could avoid computing hashes by passing in the already-transformed dataset, e.g.\n",
    "// `model.approxSimilarityJoin(transformedA, transformedB, 0.6)`\n",
    "println(\"Approximately joining dfA and dfB on Jaccard distance smaller than 0.6:\")\n",
    "model.approxSimilarityJoin(dfA, dfB, 0.6, \"JaccardDistance\")\n",
    "  .select(col(\"datasetA.id\").alias(\"idA\"),\n",
    "    col(\"datasetB.id\").alias(\"idB\"),\n",
    "    col(\"JaccardDistance\")).show()\n",
    "\n",
    "// Compute the locality sensitive hashes for the input rows, then perform approximate nearest\n",
    "// neighbor search.\n",
    "// We could avoid computing hashes by passing in the already-transformed dataset, e.g.\n",
    "// `model.approxNearestNeighbors(transformedA, key, 2)`\n",
    "// It may return less than 2 rows when not enough approximate near-neighbor candidates are\n",
    "// found.\n",
    "println(\"Approximately searching dfA for 2 nearest neighbors of the key:\")\n",
    "model.approxNearestNeighbors(dfA, key, 2).show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 五，分类模型"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mllib支持常见的机器学习分类模型：逻辑回归，SoftMax回归，决策树，随机森林，梯度提升树，线性支持向量机，朴素贝叶斯，One-Vs-Rest，以及多层感知机模型。这些模型的接口使用方法基本大同小异，下面仅仅列举常用的决策树，随机森林和梯度提升树的使用作为示范。更多范例参见官方文档。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**1，决策树**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "data = [label: double, features: vector]\n",
       "labelIndexer = strIdx_f3f6e1f77fd5\n",
       "featureIndexer = vecIdx_f7dc3970a8a1\n",
       "trainingData = [label: double, features: vector]\n",
       "testData = [label: double, features: vector]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[label: double, features: vector]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.classification.DecisionTreeClassificationModel\n",
    "import org.apache.spark.ml.classification.DecisionTreeClassifier\n",
    "import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator\n",
    "import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}\n",
    "\n",
    "\n",
    "val data = spark.read.format(\"libsvm\").load(\"data/mllib/sample_libsvm_data.txt\")\n",
    "\n",
    "\n",
    "val labelIndexer = new StringIndexer()\n",
    "  .setInputCol(\"label\")\n",
    "  .setOutputCol(\"indexedLabel\")\n",
    "  .fit(data)\n",
    "\n",
    "val featureIndexer = new VectorIndexer()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"indexedFeatures\")\n",
    "  .setMaxCategories(4) // features with > 4 distinct values are treated as continuous.\n",
    "  .fit(data)\n",
    "\n",
    "val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-----+--------------------+\n",
      "|predictedLabel|label|            features|\n",
      "+--------------+-----+--------------------+\n",
      "|           0.0|  0.0|(692,[95,96,97,12...|\n",
      "|           0.0|  0.0|(692,[123,124,125...|\n",
      "|           0.0|  0.0|(692,[124,125,126...|\n",
      "|           0.0|  0.0|(692,[126,127,128...|\n",
      "|           0.0|  0.0|(692,[126,127,128...|\n",
      "+--------------+-----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "tree = dtc_d794122a02cf\n",
       "labelConverter = idxToStr_b0ce93d19a69\n",
       "pipeline = pipeline_5233752d6d77\n",
       "model = pipeline_5233752d6d77\n",
       "predictions = [label: double, features: vector ... 6 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[label: double, features: vector ... 6 more fields]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val tree = new DecisionTreeClassifier()\n",
    "  .setLabelCol(\"indexedLabel\")\n",
    "  .setFeaturesCol(\"indexedFeatures\")\n",
    "\n",
    "val labelConverter = new IndexToString()\n",
    "  .setInputCol(\"prediction\")\n",
    "  .setOutputCol(\"predictedLabel\")\n",
    "  .setLabels(labelIndexer.labels)\n",
    "\n",
    "val pipeline = new Pipeline()\n",
    "  .setStages(Array(labelIndexer, featureIndexer, tree, labelConverter))\n",
    "\n",
    "val model = pipeline.fit(trainingData)\n",
    "\n",
    "\n",
    "val predictions = model.transform(testData)\n",
    "\n",
    "predictions.select(\"predictedLabel\", \"label\", \"features\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test Error = 0.06896551724137934\n",
      "Learned classification tree model:\n",
      " DecisionTreeClassificationModel (uid=dtc_d794122a02cf) of depth 1 with 3 nodes\n",
      "  If (feature 406 <= 9.5)\n",
      "   Predict: 1.0\n",
      "  Else (feature 406 > 9.5)\n",
      "   Predict: 0.0\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "evaluator = mcEval_f5bd0f14f10e\n",
       "accuracy = 0.9310344827586207\n",
       "treeModel = DecisionTreeClassificationModel (uid=dtc_d794122a02cf) of depth 1 with 3 nodes\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "DecisionTreeClassificationModel (uid=dtc_d794122a02cf) of depth 1 with 3 nodes"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val evaluator = new MulticlassClassificationEvaluator()\n",
    "  .setLabelCol(\"indexedLabel\")\n",
    "  .setPredictionCol(\"prediction\")\n",
    "  .setMetricName(\"accuracy\")\n",
    "val accuracy = evaluator.evaluate(predictions)\n",
    "println(s\"Test Error = ${(1.0 - accuracy)}\")\n",
    "\n",
    "val treeModel = model.stages(2).asInstanceOf[DecisionTreeClassificationModel]\n",
    "println(s\"Learned classification tree model:\\n ${treeModel.toDebugString}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**2，随机森林**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-----+--------------------+\n",
      "|predictedLabel|label|            features|\n",
      "+--------------+-----+--------------------+\n",
      "|           0.0|  0.0|(692,[98,99,100,1...|\n",
      "|           0.0|  0.0|(692,[122,123,124...|\n",
      "|           0.0|  0.0|(692,[124,125,126...|\n",
      "|           0.0|  0.0|(692,[124,125,126...|\n",
      "|           0.0|  0.0|(692,[124,125,126...|\n",
      "+--------------+-----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Test Error = 0.0\n",
      "Learned classification forest model:\n",
      " RandomForestClassificationModel (uid=rfc_00687b80a80b) with 10 trees\n",
      "  Tree 0 (weight 1.0):\n",
      "    If (feature 412 <= 8.0)\n",
      "     If (feature 323 <= 7.5)\n",
      "      If (feature 630 <= 5.0)\n",
      "       Predict: 0.0\n",
      "      Else (feature 630 > 5.0)\n",
      "       Predict: 1.0\n",
      "     Else (feature 323 > 7.5)\n",
      "      Predict: 0.0\n",
      "    Else (feature 412 > 8.0)\n",
      "     Predict: 1.0\n",
      "  Tree 1 (weight 1.0):\n",
      "    If (feature 463 <= 2.0)\n",
      "     If (feature 605 <= 12.0)\n",
      "      If (feature 207 <= 2.0)\n",
      "       Predict: 0.0\n",
      "      Else (feature 207 > 2.0)\n",
      "       Predict: 1.0\n",
      "     Else (feature 605 > 12.0)\n",
      "      Predict: 1.0\n",
      "    Else (feature 463 > 2.0)\n",
      "     Predict: 0.0\n",
      "  Tree 2 (weight 1.0):\n",
      "    If (feature 540 <= 87.0)\n",
      "     If (feature 317 <= 8.0)\n",
      "      If (feature 377 <= 0.5)\n",
      "       Predict: 1.0\n",
      "      Else (feature 377 > 0.5)\n",
      "       Predict: 0.0\n",
      "     Else (feature 317 > 8.0)\n",
      "      Predict: 1.0\n",
      "    Else (feature 540 > 87.0)\n",
      "     Predict: 1.0\n",
      "  Tree 3 (weight 1.0):\n",
      "    If (feature 328 <= 24.0)\n",
      "     If (feature 350 <= 13.5)\n",
      "      Predict: 1.0\n",
      "     Else (feature 350 > 13.5)\n",
      "      Predict: 0.0\n",
      "    Else (feature 328 > 24.0)\n",
      "     Predict: 1.0\n",
      "  Tree 4 (weight 1.0):\n",
      "    If (feature 429 <= 7.0)\n",
      "     If (feature 358 <= 15.5)\n",
      "      Predict: 0.0\n",
      "     Else (feature 358 > 15.5)\n",
      "      Predict: 1.0\n",
      "    Else (feature 429 > 7.0)\n",
      "     Predict: 1.0\n",
      "  Tree 5 (weight 1.0):\n",
      "    If (feature 462 <= 62.5)\n",
      "     If (feature 240 <= 253.5)\n",
      "      Predict: 1.0\n",
      "     Else (feature 240 > 253.5)\n",
      "      If (feature 651 <= 0.5)\n",
      "       Predict: 0.0\n",
      "      Else (feature 651 > 0.5)\n",
      "       Predict: 1.0\n",
      "    Else (feature 462 > 62.5)\n",
      "     Predict: 0.0\n",
      "  Tree 6 (weight 1.0):\n",
      "    If (feature 512 <= 37.5)\n",
      "     If (feature 523 <= 44.5)\n",
      "      Predict: 0.0\n",
      "     Else (feature 523 > 44.5)\n",
      "      Predict: 1.0\n",
      "    Else (feature 512 > 37.5)\n",
      "     Predict: 1.0\n",
      "  Tree 7 (weight 1.0):\n",
      "    If (feature 512 <= 37.5)\n",
      "     If (feature 424 in {0.0})\n",
      "      Predict: 0.0\n",
      "     Else (feature 424 not in {0.0})\n",
      "      Predict: 1.0\n",
      "    Else (feature 512 > 37.5)\n",
      "     Predict: 1.0\n",
      "  Tree 8 (weight 1.0):\n",
      "    If (feature 462 <= 62.5)\n",
      "     If (feature 324 <= 251.5)\n",
      "      Predict: 1.0\n",
      "     Else (feature 324 > 251.5)\n",
      "      Predict: 0.0\n",
      "    Else (feature 462 > 62.5)\n",
      "     Predict: 0.0\n",
      "  Tree 9 (weight 1.0):\n",
      "    If (feature 377 <= 34.0)\n",
      "     If (feature 631 <= 5.5)\n",
      "      Predict: 0.0\n",
      "     Else (feature 631 > 5.5)\n",
      "      Predict: 1.0\n",
      "    Else (feature 377 > 34.0)\n",
      "     If (feature 413 <= 11.0)\n",
      "      Predict: 0.0\n",
      "     Else (feature 413 > 11.0)\n",
      "      Predict: 1.0\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = [label: double, features: vector]\n",
       "labelIndexer = strIdx_489d7a427841\n",
       "featureIndexer = vecIdx_b573479bab67\n",
       "trainingData = [label: double, features: vector]\n",
       "testData = [label: double, features: vector]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "rf: org.apache.spark.ml.classification.RandomForestClassifier...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[label: double, features: vector]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.classification.{RandomForestClassificationModel, RandomForestClassifier}\n",
    "import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator\n",
    "import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}\n",
    "\n",
    "// Load and parse the data file, converting it to a DataFrame.\n",
    "val data = spark.read.format(\"libsvm\").load(\"data/mllib/sample_libsvm_data.txt\")\n",
    "\n",
    "// Index labels, adding metadata to the label column.\n",
    "// Fit on whole dataset to include all labels in index.\n",
    "val labelIndexer = new StringIndexer()\n",
    "  .setInputCol(\"label\")\n",
    "  .setOutputCol(\"indexedLabel\")\n",
    "  .fit(data)\n",
    "// Automatically identify categorical features, and index them.\n",
    "// Set maxCategories so features with > 4 distinct values are treated as continuous.\n",
    "val featureIndexer = new VectorIndexer()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"indexedFeatures\")\n",
    "  .setMaxCategories(4)\n",
    "  .fit(data)\n",
    "\n",
    "// Split the data into training and test sets (30% held out for testing).\n",
    "val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))\n",
    "\n",
    "// Train a RandomForest model.\n",
    "val rf = new RandomForestClassifier()\n",
    "  .setLabelCol(\"indexedLabel\")\n",
    "  .setFeaturesCol(\"indexedFeatures\")\n",
    "  .setNumTrees(10)\n",
    "\n",
    "// Convert indexed labels back to original labels.\n",
    "val labelConverter = new IndexToString()\n",
    "  .setInputCol(\"prediction\")\n",
    "  .setOutputCol(\"predictedLabel\")\n",
    "  .setLabels(labelIndexer.labels)\n",
    "\n",
    "// Chain indexers and forest in a Pipeline.\n",
    "val pipeline = new Pipeline()\n",
    "  .setStages(Array(labelIndexer, featureIndexer, rf, labelConverter))\n",
    "\n",
    "// Train model. This also runs the indexers.\n",
    "val model = pipeline.fit(trainingData)\n",
    "\n",
    "// Make predictions.\n",
    "val predictions = model.transform(testData)\n",
    "\n",
    "// Select example rows to display.\n",
    "predictions.select(\"predictedLabel\", \"label\", \"features\").show(5)\n",
    "\n",
    "// Select (prediction, true label) and compute test error.\n",
    "val evaluator = new MulticlassClassificationEvaluator()\n",
    "  .setLabelCol(\"indexedLabel\")\n",
    "  .setPredictionCol(\"prediction\")\n",
    "  .setMetricName(\"accuracy\")\n",
    "val accuracy = evaluator.evaluate(predictions)\n",
    "println(s\"Test Error = ${(1.0 - accuracy)}\")\n",
    "\n",
    "val rfModel = model.stages(2).asInstanceOf[RandomForestClassificationModel]\n",
    "println(s\"Learned classification forest model:\\n ${rfModel.toDebugString}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**3，梯度提升树**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-----+--------------------+\n",
      "|predictedLabel|label|            features|\n",
      "+--------------+-----+--------------------+\n",
      "|           0.0|  0.0|(692,[123,124,125...|\n",
      "|           0.0|  0.0|(692,[123,124,125...|\n",
      "|           0.0|  0.0|(692,[124,125,126...|\n",
      "|           0.0|  0.0|(692,[124,125,126...|\n",
      "|           0.0|  0.0|(692,[126,127,128...|\n",
      "+--------------+-----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Test Error = 0.0\n",
      "Learned classification GBT model:\n",
      " GBTClassificationModel (uid=gbtc_3c7196e716c1) with 10 trees\n",
      "  Tree 0 (weight 1.0):\n",
      "    If (feature 434 <= 70.5)\n",
      "     If (feature 99 in {2.0})\n",
      "      Predict: -1.0\n",
      "     Else (feature 99 not in {2.0})\n",
      "      Predict: 1.0\n",
      "    Else (feature 434 > 70.5)\n",
      "     Predict: -1.0\n",
      "  Tree 1 (weight 0.1):\n",
      "    If (feature 490 <= 43.0)\n",
      "     If (feature 344 <= 253.5)\n",
      "      Predict: 0.4768116880884702\n",
      "     Else (feature 344 > 253.5)\n",
      "      Predict: -0.4768116880884694\n",
      "    Else (feature 490 > 43.0)\n",
      "     Predict: -0.47681168808846996\n",
      "  Tree 2 (weight 0.1):\n",
      "    If (feature 434 <= 70.5)\n",
      "     If (feature 627 <= 2.5)\n",
      "      Predict: -0.4381935810427206\n",
      "     Else (feature 627 > 2.5)\n",
      "      Predict: 0.43819358104272055\n",
      "    Else (feature 434 > 70.5)\n",
      "     Predict: -0.4381935810427206\n",
      "  Tree 3 (weight 0.1):\n",
      "    If (feature 490 <= 43.0)\n",
      "     If (feature 627 <= 2.5)\n",
      "      Predict: -0.4051496802845983\n",
      "     Else (feature 627 > 2.5)\n",
      "      Predict: 0.40514968028459825\n",
      "    Else (feature 490 > 43.0)\n",
      "     Predict: -0.40514968028459825\n",
      "  Tree 4 (weight 0.1):\n",
      "    If (feature 490 <= 43.0)\n",
      "     If (feature 100 <= 193.5)\n",
      "      If (feature 207 <= 18.0)\n",
      "       Predict: 0.3765841318352991\n",
      "      Else (feature 207 > 18.0)\n",
      "       Predict: 0.3765841318352993\n",
      "     Else (feature 100 > 193.5)\n",
      "      Predict: -0.3765841318352994\n",
      "    Else (feature 490 > 43.0)\n",
      "     If (feature 404 <= 1.0)\n",
      "      Predict: -0.37658413183529915\n",
      "     Else (feature 404 > 1.0)\n",
      "      Predict: -0.3765841318352994\n",
      "  Tree 5 (weight 0.1):\n",
      "    If (feature 434 <= 70.5)\n",
      "     If (feature 99 in {2.0})\n",
      "      Predict: -0.35166478958101005\n",
      "     Else (feature 99 not in {2.0})\n",
      "      If (feature 184 <= 60.5)\n",
      "       Predict: 0.35166478958101005\n",
      "      Else (feature 184 > 60.5)\n",
      "       Predict: 0.3516647895810101\n",
      "    Else (feature 434 > 70.5)\n",
      "     If (feature 295 <= 132.0)\n",
      "      If (feature 237 <= 31.5)\n",
      "       Predict: -0.35166478958101005\n",
      "      Else (feature 237 > 31.5)\n",
      "       If (feature 237 <= 207.5)\n",
      "        Predict: -0.35166478958101005\n",
      "       Else (feature 237 > 207.5)\n",
      "        Predict: -0.3516647895810101\n",
      "     Else (feature 295 > 132.0)\n",
      "      Predict: -0.35166478958101005\n",
      "  Tree 6 (weight 0.1):\n",
      "    If (feature 490 <= 43.0)\n",
      "     If (feature 295 <= 253.5)\n",
      "      If (feature 126 <= 39.0)\n",
      "       Predict: 0.32974984655529926\n",
      "      Else (feature 126 > 39.0)\n",
      "       Predict: 0.3297498465552993\n",
      "     Else (feature 295 > 253.5)\n",
      "      Predict: -0.32974984655530015\n",
      "    Else (feature 490 > 43.0)\n",
      "     Predict: -0.3297498465552992\n",
      "  Tree 7 (weight 0.1):\n",
      "    If (feature 434 <= 70.5)\n",
      "     If (feature 627 <= 2.5)\n",
      "      Predict: -0.3103372455197954\n",
      "     Else (feature 627 > 2.5)\n",
      "      If (feature 210 <= 55.5)\n",
      "       Predict: 0.3103372455197956\n",
      "      Else (feature 210 > 55.5)\n",
      "       If (feature 373 <= 11.5)\n",
      "        Predict: 0.3103372455197956\n",
      "       Else (feature 373 > 11.5)\n",
      "        Predict: 0.3103372455197957\n",
      "    Else (feature 434 > 70.5)\n",
      "     If (feature 405 <= 195.0)\n",
      "      If (feature 267 <= 173.0)\n",
      "       Predict: -0.3103372455197956\n",
      "      Else (feature 267 > 173.0)\n",
      "       Predict: -0.3103372455197957\n",
      "     Else (feature 405 > 195.0)\n",
      "      Predict: -0.31033724551979563\n",
      "  Tree 8 (weight 0.1):\n",
      "    If (feature 434 <= 70.5)\n",
      "     If (feature 627 <= 2.5)\n",
      "      Predict: -0.2930291649125432\n",
      "     Else (feature 627 > 2.5)\n",
      "      If (feature 156 <= 22.0)\n",
      "       Predict: 0.2930291649125433\n",
      "      Else (feature 156 > 22.0)\n",
      "       Predict: 0.2930291649125434\n",
      "    Else (feature 434 > 70.5)\n",
      "     If (feature 433 <= 119.0)\n",
      "      Predict: -0.2930291649125433\n",
      "     Else (feature 433 > 119.0)\n",
      "      Predict: -0.2930291649125434\n",
      "  Tree 9 (weight 0.1):\n",
      "    If (feature 462 <= 62.5)\n",
      "     If (feature 628 <= 0.5)\n",
      "      Predict: -0.27750666438358235\n",
      "     Else (feature 628 > 0.5)\n",
      "      Predict: 0.2775066643835825\n",
      "    Else (feature 462 > 62.5)\n",
      "     If (feature 405 <= 182.5)\n",
      "      Predict: -0.2775066643835825\n",
      "     Else (feature 405 > 182.5)\n",
      "      If (feature 322 <= 250.5)\n",
      "       If (feature 322 <= 31.0)\n",
      "        Predict: -0.27750666438358246\n",
      "       Else (feature 322 > 31.0)\n",
      "        Predict: -0.2775066643835825\n",
      "      Else (feature 322 > 250.5)\n",
      "       Predict: -0.27750666438358257\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = [label: double, features: vector]\n",
       "labelIndexer = strIdx_5aa2ccbbcbe3\n",
       "featureIndexer = vecIdx_084188469920\n",
       "trainingData = [label: double, features: vector]\n",
       "testData = [label: double, features: vector]\n",
       "gbt = gbtc_3c7196e716c1\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "label...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "gbtc_3c7196e716c1"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.classification.{GBTClassificationModel, GBTClassifier}\n",
    "import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator\n",
    "import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}\n",
    "\n",
    "// Load and parse the data file, converting it to a DataFrame.\n",
    "val data = spark.read.format(\"libsvm\").load(\"data/mllib/sample_libsvm_data.txt\")\n",
    "\n",
    "// Index labels, adding metadata to the label column.\n",
    "// Fit on whole dataset to include all labels in index.\n",
    "val labelIndexer = new StringIndexer()\n",
    "  .setInputCol(\"label\")\n",
    "  .setOutputCol(\"indexedLabel\")\n",
    "  .fit(data)\n",
    "// Automatically identify categorical features, and index them.\n",
    "// Set maxCategories so features with > 4 distinct values are treated as continuous.\n",
    "val featureIndexer = new VectorIndexer()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"indexedFeatures\")\n",
    "  .setMaxCategories(4)\n",
    "  .fit(data)\n",
    "\n",
    "// Split the data into training and test sets (30% held out for testing).\n",
    "val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))\n",
    "\n",
    "// Train a GBT model.\n",
    "val gbt = new GBTClassifier()\n",
    "  .setLabelCol(\"indexedLabel\")\n",
    "  .setFeaturesCol(\"indexedFeatures\")\n",
    "  .setMaxIter(10)\n",
    "  .setFeatureSubsetStrategy(\"auto\")\n",
    "\n",
    "// Convert indexed labels back to original labels.\n",
    "val labelConverter = new IndexToString()\n",
    "  .setInputCol(\"prediction\")\n",
    "  .setOutputCol(\"predictedLabel\")\n",
    "  .setLabels(labelIndexer.labels)\n",
    "\n",
    "// Chain indexers and GBT in a Pipeline.\n",
    "val pipeline = new Pipeline()\n",
    "  .setStages(Array(labelIndexer, featureIndexer, gbt, labelConverter))\n",
    "\n",
    "// Train model. This also runs the indexers.\n",
    "val model = pipeline.fit(trainingData)\n",
    "\n",
    "// Make predictions.\n",
    "val predictions = model.transform(testData)\n",
    "\n",
    "// Select example rows to display.\n",
    "predictions.select(\"predictedLabel\", \"label\", \"features\").show(5)\n",
    "\n",
    "// Select (prediction, true label) and compute test error.\n",
    "val evaluator = new MulticlassClassificationEvaluator()\n",
    "  .setLabelCol(\"indexedLabel\")\n",
    "  .setPredictionCol(\"prediction\")\n",
    "  .setMetricName(\"accuracy\")\n",
    "val accuracy = evaluator.evaluate(predictions)\n",
    "println(s\"Test Error = ${1.0 - accuracy}\")\n",
    "\n",
    "val gbtModel = model.stages(2).asInstanceOf[GBTClassificationModel]\n",
    "println(s\"Learned classification GBT model:\\n ${gbtModel.toDebugString}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------------+\n",
      "|label|            features|\n",
      "+-----+--------------------+\n",
      "|  0.0|(692,[127,128,129...|\n",
      "|  1.0|(692,[158,159,160...|\n",
      "|  1.0|(692,[124,125,126...|\n",
      "|  1.0|(692,[152,153,154...|\n",
      "|  1.0|(692,[151,152,153...|\n",
      "|  0.0|(692,[129,130,131...|\n",
      "|  1.0|(692,[158,159,160...|\n",
      "|  1.0|(692,[99,100,101,...|\n",
      "|  0.0|(692,[154,155,156...|\n",
      "|  0.0|(692,[127,128,129...|\n",
      "|  1.0|(692,[154,155,156...|\n",
      "|  0.0|(692,[153,154,155...|\n",
      "|  0.0|(692,[151,152,153...|\n",
      "|  1.0|(692,[129,130,131...|\n",
      "|  0.0|(692,[154,155,156...|\n",
      "|  1.0|(692,[150,151,152...|\n",
      "|  0.0|(692,[124,125,126...|\n",
      "|  0.0|(692,[152,153,154...|\n",
      "|  1.0|(692,[97,98,99,12...|\n",
      "|  1.0|(692,[124,125,126...|\n",
      "+-----+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.show"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**4, xgboost模型**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "spark也可以使用xgboost模型，既可以进行分布式训练，也可以进行分布式预测，支持各种参数设置和earlystopping。\n",
    "\n",
    "支持模型保存，并且保存后的模型和Python等语言是可以相互调用的。\n",
    "\n",
    "需要注意的是，输入xgboost的数据格式只能包含两列，features和label。\n",
    "\n",
    "可以用spark.ml.feature.VectorAssembler将不同特征转换成一个 features向量。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "spark中要使用xgboost模型，pom文件需要导入如下依赖"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "<dependency>\n",
    "  <groupId>ml.dmlc</groupId>\n",
    "  <artifactId>xgboost4j-spark</artifactId>\n",
    "  <version>latest_version_num</version>\n",
    "</dependency>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T08:16:32.679896Z",
     "start_time": "2019-10-08T08:16:31.790Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----------+------------+-----------+-----------+\n",
      "|sepal length|sepal width|petal length|petal width|      class|\n",
      "+------------+-----------+------------+-----------+-----------+\n",
      "|         4.7|        3.2|         1.3|        0.2|Iris-setosa|\n",
      "|         4.4|        2.9|         1.4|        0.2|Iris-setosa|\n",
      "|         5.7|        4.4|         1.5|        0.4|Iris-setosa|\n",
      "|         5.2|        3.5|         1.5|        0.2|Iris-setosa|\n",
      "+------------+-----------+------------+-----------+-----------+\n",
      "only showing top 4 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "spark = org.apache.spark.sql.SparkSession@717bac32\n",
       "schema = StructType(StructField(sepal length,DoubleType,true), StructField(sepal width,DoubleType,true), StructField(petal length,DoubleType,true), StructField(petal width,DoubleType,true), StructField(class,StringType,true))\n",
       "rawInput = [sepal length: double, sepal width: double ... 3 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[sepal length: double, sepal width: double ... 3 more fields]"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//导入数据\n",
    "\n",
    "import org.apache.spark.sql.SparkSession\n",
    "import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}\n",
    "\n",
    "val spark = SparkSession.builder().getOrCreate()\n",
    "val schema = new StructType(Array(\n",
    "  StructField(\"sepal length\", DoubleType, true),\n",
    "  StructField(\"sepal width\", DoubleType, true),\n",
    "  StructField(\"petal length\", DoubleType, true),\n",
    "  StructField(\"petal width\", DoubleType, true),\n",
    "  StructField(\"class\", StringType, true)))\n",
    "val rawInput = spark.read.schema(schema).csv(\"data/iris.csv\")\n",
    "\n",
    "rawInput.sample(false,0.1,1).show(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T08:16:51.944720Z",
     "start_time": "2019-10-08T08:16:51.250Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----------+------------+-----------+----------+\n",
      "|sepal length|sepal width|petal length|petal width|classIndex|\n",
      "+------------+-----------+------------+-----------+----------+\n",
      "|         5.4|        3.4|         1.7|        0.2|       0.0|\n",
      "|         5.5|        4.2|         1.4|        0.2|       0.0|\n",
      "|         5.1|        3.4|         1.5|        0.2|       0.0|\n",
      "|         5.1|        3.8|         1.9|        0.4|       0.0|\n",
      "|         5.3|        3.7|         1.5|        0.2|       0.0|\n",
      "|         6.3|        3.3|         4.7|        1.6|       1.0|\n",
      "|         6.0|        2.7|         5.1|        1.6|       1.0|\n",
      "|         5.6|        3.0|         4.1|        1.3|       1.0|\n",
      "|         5.7|        3.0|         4.2|        1.2|       1.0|\n",
      "|         7.3|        2.9|         6.3|        1.8|       2.0|\n",
      "|         7.7|        3.0|         6.1|        2.3|       2.0|\n",
      "+------------+-----------+------------+-----------+----------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "stringIndexer = strIdx_f24e615dbffd\n",
       "labelTransformed = [sepal length: double, sepal width: double ... 3 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[sepal length: double, sepal width: double ... 3 more fields]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//label转换成数字编码\n",
    "import org.apache.spark.ml.feature.StringIndexer\n",
    "val stringIndexer = new StringIndexer().\n",
    "  setInputCol(\"class\").\n",
    "  setOutputCol(\"classIndex\").\n",
    "  fit(rawInput)\n",
    "val labelTransformed = stringIndexer.transform(rawInput).drop(\"class\")\n",
    "labelTransformed.sample(false,0.1,0).show"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T08:17:10.165649Z",
     "start_time": "2019-10-08T08:17:09.597Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "vectorAssembler = vecAssembler_42bde9b2d00d\n",
       "xgbInput = [features: vector, classIndex: double]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[features: vector, classIndex: double]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//将特征组合成features向量\n",
    "import org.apache.spark.ml.feature.VectorAssembler\n",
    "val vectorAssembler = new VectorAssembler().\n",
    "  setInputCols(Array(\"sepal length\", \"sepal width\", \"petal length\", \"petal width\")).\n",
    "  setOutputCol(\"features\")\n",
    "val xgbInput = vectorAssembler.transform(labelTransformed).select(\"features\", \"classIndex\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T08:18:38.703238Z",
     "start_time": "2019-10-08T08:18:27.184Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Tracker started, with env={DMLC_NUM_SERVER=0, DMLC_TRACKER_URI=192.168.1.102, DMLC_TRACKER_PORT=9091, DMLC_NUM_WORKER=2}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "xgbParam = Map(num_workers -> 2, max_depth -> 2, num_class -> 3, objective -> multi:softprob, num_round -> 20, eta -> 0.1)\n",
       "xgbClassifier = xgbc_4508eebedb0a\n",
       "xgbClassificationModel = xgbc_4508eebedb0a\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "xgbc_4508eebedb0a"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//训练模型\n",
    "import ml.dmlc.xgboost4j.scala.spark.XGBoostClassifier\n",
    "val xgbParam = Map(\"eta\" -> 0.1f,\n",
    "      \"max_depth\" -> 2,\n",
    "      \"objective\" -> \"multi:softprob\",\n",
    "      \"num_class\" -> 3,\n",
    "      \"num_round\" -> 20,\n",
    "      \"num_workers\" -> 2)\n",
    "val xgbClassifier = new XGBoostClassifier(xgbParam).\n",
    "      setFeaturesCol(\"features\").\n",
    "      setLabelCol(\"classIndex\")\n",
    "\n",
    "//也可以通过方法进行设置参数\n",
    "xgbClassifier.setMaxDepth(3)\n",
    "xgbClassifier.setMinChildWeight(3.0)\n",
    "\n",
    "val xgbClassificationModel = xgbClassifier.fit(xgbInput)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T08:24:35.548168Z",
     "start_time": "2019-10-08T08:24:32.700Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+----------+--------------------+--------------------+----------+\n",
      "|         features|classIndex|       rawPrediction|         probability|prediction|\n",
      "+-----------------+----------+--------------------+--------------------+----------+\n",
      "|[5.7,4.4,1.5,0.4]|       0.0|[2.75150775909423...|[0.95917052030563...|       0.0|\n",
      "|[5.2,3.5,1.5,0.2]|       0.0|[2.75150775909423...|[0.98360401391983...|       0.0|\n",
      "|[5.1,3.8,1.9,0.4]|       0.0|[2.75150775909423...|[0.98360401391983...|       0.0|\n",
      "|[4.9,2.4,3.3,1.0]|       1.0|[-1.5969576835632...|[0.03399981185793...|       1.0|\n",
      "+-----------------+----------+--------------------+--------------------+----------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "results = [features: vector, classIndex: double ... 3 more fields]\n",
       "features = [5.1,3.5,1.4,0.2]\n",
       "single_result = 0.0\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "0.0"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//使用模型\n",
    "val results = xgbClassificationModel.transform(xgbInput)\n",
    "results.sample(false,0.01,1).show(5)\n",
    "\n",
    "//对单个样本进行预测\n",
    "val features = xgbInput.head().getAs[Vector](\"features\")\n",
    "val single_result = xgbClassificationModel.predict(features)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T08:28:30.672915Z",
     "start_time": "2019-10-08T08:28:30.219Z"
    }
   },
   "outputs": [],
   "source": [
    "//模型保存\n",
    "import ml.dmlc.xgboost4j.scala.spark.XGBoostClassificationModel\n",
    "xgbClassificationModel.write.overwrite().save(\"xgb.model\")\n",
    "\n",
    "\n",
    "//重新载入模型\n",
    "val xgbClassificationModel2 = XGBoostClassificationModel.load(\"xgb.model\")\n",
    "xgbClassificationModel2.transform(xgbInput).show\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 六，回归模型"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mllib支持常见的回归模型，如线性回归，广义线性回归，决策树回归，随机森林回归，梯度提升树回归，生存回归，保序回归。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**1，线性回归**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "numIterations: 7\n",
      "+------------------+\n",
      "|         residuals|\n",
      "+------------------+\n",
      "|-9.889232683103197|\n",
      "|0.5533794340053554|\n",
      "|-5.204019455758823|\n",
      "+------------------+\n",
      "only showing top 3 rows\n",
      "\n",
      "RMSE: 10.189077167598475\n",
      "r2: 0.022861466913958184\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "training = [label: double, features: vector]\n",
       "linear = linReg_e1c94ac8fd1b\n",
       "model = linReg_e1c94ac8fd1b\n",
       "trainingSummary = org.apache.spark.ml.regression.LinearRegressionTrainingSummary@5b43f1d8\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "org.apache.spark.ml.regression.LinearRegressionTrainingSummary@5b43f1d8"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.regression.LinearRegression\n",
    "\n",
    "val training = spark.read.format(\"libsvm\")\n",
    "  .load(\"data/mllib/sample_linear_regression_data.txt\")\n",
    "\n",
    "val linear = new LinearRegression()\n",
    "  .setMaxIter(10)\n",
    "  .setRegParam(0.3)\n",
    "  .setElasticNetParam(0.8)\n",
    "\n",
    "val model = linear.fit(training)\n",
    "\n",
    "//println(s\"Coefficients: ${model.coefficients} Intercept: ${model.intercept}\")\n",
    "\n",
    "val trainingSummary = model.summary\n",
    "println(s\"numIterations: ${trainingSummary.totalIterations}\")\n",
    "\n",
    "//println(s\"objectiveHistory: [${trainingSummary.objectiveHistory.mkString(\",\")}]\")\n",
    "\n",
    "trainingSummary.residuals.show(3)\n",
    "println(s\"RMSE: ${trainingSummary.rootMeanSquaredError}\")\n",
    "println(s\"r2: ${trainingSummary.r2}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**2，决策树回归**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "data = [label: double, features: vector]\n",
       "featureIndexer = vecIdx_6ffc9483212f\n",
       "trainingData = [label: double, features: vector]\n",
       "testData = [label: double, features: vector]\n",
       "tree = dtr_0e6117279ad5\n",
       "pipeline = pipeline_00a2585936c6\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "pipeline_00a2585936c6"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.evaluation.RegressionEvaluator\n",
    "import org.apache.spark.ml.feature.VectorIndexer\n",
    "import org.apache.spark.ml.regression.DecisionTreeRegressionModel\n",
    "import org.apache.spark.ml.regression.DecisionTreeRegressor\n",
    "\n",
    "val data = spark.read.format(\"libsvm\").load(\"data/mllib/sample_libsvm_data.txt\")\n",
    "\n",
    "val featureIndexer = new VectorIndexer()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"indexedFeatures\")\n",
    "  .setMaxCategories(4)\n",
    "  .fit(data)\n",
    "\n",
    "val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))\n",
    "\n",
    "val tree = new DecisionTreeRegressor()\n",
    "  .setLabelCol(\"label\")\n",
    "  .setFeaturesCol(\"indexedFeatures\")\n",
    "\n",
    "val pipeline = new Pipeline()\n",
    "  .setStages(Array(featureIndexer, tree))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root Mean Squared Error (RMSE) on test data = 0.0\n",
      "Learned regression tree model:\n",
      " DecisionTreeRegressionModel (uid=dtr_0e6117279ad5) of depth 2 with 5 nodes\n",
      "  If (feature 434 <= 70.5)\n",
      "   If (feature 99 in {0.0,3.0})\n",
      "    Predict: 0.0\n",
      "   Else (feature 99 not in {0.0,3.0})\n",
      "    Predict: 1.0\n",
      "  Else (feature 434 > 70.5)\n",
      "   Predict: 1.0\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "model = pipeline_00a2585936c6\n",
       "predictions = [label: double, features: vector ... 2 more fields]\n",
       "evaluator = regEval_ee3834d63b76\n",
       "rmse = 0.0\n",
       "treeModel = DecisionTreeRegressionModel (uid=dtr_0e6117279ad5) of depth 2 with 5 nodes\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "DecisionTreeRegressionModel (uid=dtr_0e6117279ad5) of depth 2 with 5 nodes"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val model = pipeline.fit(trainingData)\n",
    "val predictions = model.transform(testData)\n",
    "\n",
    "val evaluator = new RegressionEvaluator()\n",
    "  .setLabelCol(\"label\")\n",
    "  .setPredictionCol(\"prediction\")\n",
    "  .setMetricName(\"rmse\")\n",
    "val rmse = evaluator.evaluate(predictions)\n",
    "println(s\"Root Mean Squared Error (RMSE) on test data = $rmse\")\n",
    "\n",
    "val treeModel = model.stages(1).asInstanceOf[DecisionTreeRegressionModel]\n",
    "println(s\"Learned regression tree model:\\n ${treeModel.toDebugString}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 七，聚类模型"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mllib支持的聚类模型较少，主要有K均值聚类，高斯混合模型GMM，以及二分的K均值，隐含狄利克雷分布LDA模型等。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**1，K均值聚类**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Silhouette with squared euclidean distance = 0.9997530305375207\n",
      "Cluster Centers: \n",
      "[0.1,0.1,0.1]\n",
      "[9.1,9.1,9.1]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dfdata = [label: double, features: vector]\n",
       "kmeans = kmeans_d45e6549d6f9\n",
       "model = kmeans_d45e6549d6f9\n",
       "dfpredict = [label: double, features: vector ... 1 more field]\n",
       "evaluator = cluEval_91c4b9890b93\n",
       "silhouette = 0.9997530305375207\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "0.9997530305375207"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.clustering.KMeans\n",
    "import org.apache.spark.ml.evaluation.ClusteringEvaluator\n",
    "\n",
    "\n",
    "val dfdata = spark.read.format(\"libsvm\").load(\"data/mllib/sample_kmeans_data.txt\")\n",
    "\n",
    "// Trains a k-means model.\n",
    "val kmeans = new KMeans().setK(2).setSeed(1L)\n",
    "val model = kmeans.fit(dfdata)\n",
    "\n",
    "\n",
    "val dfpredict = model.transform(dfdata)\n",
    "\n",
    "\n",
    "val evaluator = new ClusteringEvaluator()\n",
    "\n",
    "val silhouette = evaluator.evaluate(dfpredict)\n",
    "println(s\"Silhouette with squared euclidean distance = $silhouette\")\n",
    "\n",
    "println(\"Cluster Centers: \")\n",
    "model.clusterCenters.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**2，高斯混合模型**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Gaussian 0:\n",
      "weight=0.5\n",
      "mu=[0.10000000000001552,0.10000000000001552,0.10000000000001552]\n",
      "sigma=\n",
      "0.006666666666806454  0.006666666666806454  0.006666666666806454  \n",
      "0.006666666666806454  0.006666666666806454  0.006666666666806454  \n",
      "0.006666666666806454  0.006666666666806454  0.006666666666806454  \n",
      "\n",
      "Gaussian 1:\n",
      "weight=0.5\n",
      "mu=[9.099999999999984,9.099999999999984,9.099999999999984]\n",
      "sigma=\n",
      "0.006666666666812185  0.006666666666812185  0.006666666666812185  \n",
      "0.006666666666812185  0.006666666666812185  0.006666666666812185  \n",
      "0.006666666666812185  0.006666666666812185  0.006666666666812185  \n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dataset = [label: double, features: vector]\n",
       "gmm = GaussianMixture_20780158aceb\n",
       "model = GaussianMixture_20780158aceb\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "GaussianMixture_20780158aceb"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.clustering.GaussianMixture\n",
    "\n",
    "val dataset = spark.read.format(\"libsvm\").load(\"data/mllib/sample_kmeans_data.txt\")\n",
    "\n",
    "\n",
    "val gmm = new GaussianMixture()\n",
    "  .setK(2)\n",
    "val model = gmm.fit(dataset)\n",
    "\n",
    "\n",
    "for (i <- 0 until model.getK) {\n",
    "  println(s\"Gaussian $i:\\nweight=${model.weights(i)}\\n\" +\n",
    "      s\"mu=${model.gaussians(i).mean}\\nsigma=\\n${model.gaussians(i).cov}\\n\")\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**3, 二分K均值 Bisecting k-means**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Bisecting k-means是一种自上而下的层次聚类算法。所有的样本点开始时属于一个cluster,然后不断通过K均值二分裂得到多个cluster。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Within Set Sum of Squared Errors = 0.11999999999994547\n",
      "Cluster Centers: \n",
      "[0.1,0.1,0.1]\n",
      "[9.1,9.1,9.1]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dataset = [label: double, features: vector]\n",
       "bkm = bisecting-kmeans_596ee31355ba\n",
       "model = bisecting-kmeans_596ee31355ba\n",
       "cost = 0.11999999999994547\n",
       "centers = Array([0.1,0.1,0.1], [9.1,9.1,9.1])\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array([0.1,0.1,0.1], [9.1,9.1,9.1])"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.clustering.BisectingKMeans\n",
    "\n",
    "// Loads data.\n",
    "val dataset = spark.read.format(\"libsvm\").load(\"data/mllib/sample_kmeans_data.txt\")\n",
    "\n",
    "// Trains a bisecting k-means model.\n",
    "val bkm = new BisectingKMeans().setK(2).setSeed(1)\n",
    "val model = bkm.fit(dataset)\n",
    "\n",
    "// Evaluate clustering.\n",
    "val cost = model.computeCost(dataset)\n",
    "println(s\"Within Set Sum of Squared Errors = $cost\")\n",
    "\n",
    "// Shows the result.\n",
    "println(\"Cluster Centers: \")\n",
    "val centers = model.clusterCenters\n",
    "centers.foreach(println)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 八，降维模型"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mllib中支持的降维模型只有主成分分析PCA算法。这个模型在spark.ml.feature中，通常作为特征预处理的一种技巧使用。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T02:23:51.623228Z",
     "start_time": "2019-10-08T02:23:50.237Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------+\n",
      "|pcaFeatures                                                |\n",
      "+-----------------------------------------------------------+\n",
      "|[1.6485728230883807,-4.013282700516296,-5.524543751369388] |\n",
      "|[-4.645104331781534,-1.1167972663619026,-5.524543751369387]|\n",
      "|[-6.428880535676489,-5.337951427775355,-5.524543751369389] |\n",
      "+-----------------------------------------------------------+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = Array((5,[1,3],[1.0,7.0]), [2.0,0.0,3.0,4.0,5.0], [4.0,0.0,0.0,6.0,7.0])\n",
       "df = [features: vector]\n",
       "pca = pca_a9997a4e554e\n",
       "result = [pcaFeatures: vector]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[pcaFeatures: vector]"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.PCA\n",
    "import org.apache.spark.ml.linalg.Vectors\n",
    "\n",
    "val data = Array(\n",
    "  Vectors.sparse(5, Seq((1, 1.0), (3, 7.0))),\n",
    "  Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),\n",
    "  Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)\n",
    ")\n",
    "val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF(\"features\")\n",
    "\n",
    "val pca = new PCA()\n",
    "  .setInputCol(\"features\")\n",
    "  .setOutputCol(\"pcaFeatures\")\n",
    "  .setK(3)\n",
    "  .fit(df)\n",
    "\n",
    "val result = pca.transform(df).select(\"pcaFeatures\")\n",
    "result.show(false)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 九，模型优化"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "模型优化一般也称作模型选择(Model selection)或者超参调优(hyperparameter tuning)。\n",
    "\n",
    "Mllib支持网格搜索方法进行超参调优，相关函数在spark.ml.tunning模块中。\n",
    "\n",
    "有两种使用网格搜索方法的模式，一种是通过交叉验证(cross-validation)方式进行使用，另外一种是通过留出法(hold-out)方法进行使用。\n",
    "\n",
    "交叉验证模式使用的是K-fold交叉验证，将数据随机等分划分成K份，每次将一份作为验证集，其余作为训练集，根据K次验证集的平均结果来决定超参选取，计算成本较高，但是结果更加可靠。\n",
    "\n",
    "而留出法只用将数据随机划分成训练集和验证集，仅根据验证集的单次结果决定超参选取，结果没有交叉验证可靠，但计算成本较低。\n",
    "\n",
    "如果数据规模较大，一般选择留出法，如果数据规模较小，则应该选择交叉验证模式。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**1，交叉验证模式**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T02:51:45.546527Z",
     "start_time": "2019-10-08T02:51:37.059Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "training = [id: bigint, text: string ... 1 more field]\n",
       "tokenizer = tok_dd95af63ad65\n",
       "hashingTF = hashingTF_347e37957a07\n",
       "lr = logreg_62eef1851979\n",
       "pipeline = pipeline_4638baa1de4e\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "pipeline_4638baa1de4e"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.classification.LogisticRegression\n",
    "import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator\n",
    "import org.apache.spark.ml.feature.{HashingTF, Tokenizer}\n",
    "import org.apache.spark.ml.linalg.Vector\n",
    "import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}\n",
    "import org.apache.spark.sql.Row\n",
    "\n",
    "// Prepare training data from a list of (id, text, label) tuples.\n",
    "val training = spark.createDataFrame(Seq(\n",
    "  (0L, \"a b c d e spark\", 1.0),\n",
    "  (1L, \"b d\", 0.0),\n",
    "  (2L, \"spark f g h\", 1.0),\n",
    "  (3L, \"hadoop mapreduce\", 0.0),\n",
    "  (4L, \"b spark who\", 1.0),\n",
    "  (5L, \"g d a y\", 0.0),\n",
    "  (6L, \"spark fly\", 1.0),\n",
    "  (7L, \"was mapreduce\", 0.0),\n",
    "  (8L, \"e spark program\", 1.0),\n",
    "  (9L, \"a e c l\", 0.0),\n",
    "  (10L, \"spark compile\", 1.0),\n",
    "  (11L, \"hadoop software\", 0.0)\n",
    ")).toDF(\"id\", \"text\", \"label\")\n",
    "\n",
    "\n",
    "val tokenizer = new Tokenizer()\n",
    "  .setInputCol(\"text\")\n",
    "  .setOutputCol(\"words\")\n",
    "val hashingTF = new HashingTF()\n",
    "  .setInputCol(tokenizer.getOutputCol)\n",
    "  .setOutputCol(\"features\")\n",
    "val lr = new LogisticRegression()\n",
    "  .setMaxIter(10)\n",
    "val pipeline = new Pipeline()\n",
    "  .setStages(Array(tokenizer, hashingTF, lr))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T02:51:45.546527Z",
     "start_time": "2019-10-08T02:51:37.059Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(4, spark i j k) --> prob=[0.1256626071135727,0.8743373928864273], prediction=1.0\n",
      "(5, l m n) --> prob=[0.995215441016286,0.0047845589837139536], prediction=0.0\n",
      "(6, mapreduce spark) --> prob=[0.3069689523262623,0.6930310476737377], prediction=1.0\n",
      "(7, apache hadoop) --> prob=[0.8040279442401415,0.19597205575985852], prediction=0.0\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "training = [id: bigint, text: string ... 1 more field]\n",
       "tokenizer = tok_7663763d8b06\n",
       "hashingTF = hashingTF_02ec887278ff\n",
       "lr = logreg_152f516e09f1\n",
       "pipeline = pipeline_f28af747c4ce\n",
       "paramGrid = \n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "val paramGrid = new ParamGridBuilder()\n",
    "  .addGrid(hashingTF.numFeatures, Array(10, 100, 1000))\n",
    "  .addGrid(lr.regParam, Array(0.1, 0.01))\n",
    "  .build()\n",
    "\n",
    "\n",
    "val cv = new CrossValidator()\n",
    "  .setEstimator(pipeline)\n",
    "  .setEvaluator(new BinaryClassificationEvaluator)\n",
    "  .setEstimatorParamMaps(paramGrid)\n",
    "  .setNumFolds(2)  // Use 3+ in practice\n",
    "  .setParallelism(2)  // Evaluate up to 2 parameter settings in parallel\n",
    "\n",
    "\n",
    "val cvModel = cv.fit(training)\n",
    "\n",
    "val test = spark.createDataFrame(Seq(\n",
    "  (4L, \"spark i j k\"),\n",
    "  (5L, \"l m n\"),\n",
    "  (6L, \"mapreduce spark\"),\n",
    "  (7L, \"apache hadoop\")\n",
    ")).toDF(\"id\", \"text\")\n",
    "\n",
    "cvModel.transform(test)\n",
    "  .select(\"id\", \"text\", \"probability\", \"prediction\")\n",
    "  .collect()\n",
    "  .foreach { case Row(id: Long, text: String, prob: Vector, prediction: Double) =>\n",
    "    println(s\"($id, $text) --> prob=$prob, prediction=$prediction\")\n",
    "  }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**2，留出法模式**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T02:52:21.415360Z",
     "start_time": "2019-10-08T02:52:16.852Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------------+--------------------+\n",
      "|            features|              label|          prediction|\n",
      "+--------------------+-------------------+--------------------+\n",
      "|(10,[0,1,2,3,4,5,...| -23.51088409032297| -1.6659388625179559|\n",
      "|(10,[0,1,2,3,4,5,...|-21.432387764165806|  0.3400877302576284|\n",
      "|(10,[0,1,2,3,4,5,...|-12.977848725392104|-0.02335359093652395|\n",
      "+--------------------+-------------------+--------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = [label: double, features: vector]\n",
       "training = [label: double, features: vector]\n",
       "test = [label: double, features: vector]\n",
       "lr = linReg_fcc9f1e4c374\n",
       "paramGrid = \n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Array({\n",
       "\tlinReg_fcc9f1e4c374-elasticNetParam: 0.0,\n",
       "\tlinReg_fcc9f1e4c374-fitIntercept: true,\n",
       "\tlinReg_fcc9f1e4c374-regParam: 0.1\n",
       "}, {\n",
       "\tlinReg_fcc9f1e4c374-elasticNetParam: 0.0,\n",
       "\tlinReg_fcc9f1e4c374-fitIntercept:...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import org.apache.spark.ml.evaluation.RegressionEvaluator\n",
    "import org.apache.spark.ml.regression.LinearRegression\n",
    "import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit}\n",
    "\n",
    "val data = spark.read.format(\"libsvm\").load(\"data/mllib/sample_linear_regression_data.txt\")\n",
    "val Array(training, test) = data.randomSplit(Array(0.9, 0.1), seed = 12345)\n",
    "\n",
    "val lr = new LinearRegression()\n",
    "    .setMaxIter(10)\n",
    "\n",
    "val paramGrid = new ParamGridBuilder()\n",
    "  .addGrid(lr.regParam, Array(0.1, 0.01))\n",
    "  .addGrid(lr.fitIntercept)\n",
    "  .addGrid(lr.elasticNetParam, Array(0.0, 0.5, 1.0))\n",
    "  .build()\n",
    "\n",
    "val trainValidationSplit = new TrainValidationSplit()\n",
    "  .setEstimator(lr)\n",
    "  .setEvaluator(new RegressionEvaluator)\n",
    "  .setEstimatorParamMaps(paramGrid)\n",
    "  .setTrainRatio(0.8)\n",
    "  .setParallelism(2)\n",
    "\n",
    "val model = trainValidationSplit.fit(training)\n",
    "\n",
    "model.transform(test)\n",
    "  .select(\"features\", \"label\", \"prediction\")\n",
    "  .show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 十，实用工具"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "spark.ml.linalg模块提供了线性代数向量和矩阵对象。\n",
    "\n",
    "spark.ml.stat模块提供了数理统计诸如卡方检验，相关性分析等功能。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**1，向量和矩阵**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "spark.ml.linalg 支持 DenseVector，SparseVector，DenseMatrix，SparseMatrix类。\n",
    "\n",
    "并可以使用Matrices和Vectors提供的工厂方法创建向量和矩阵。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T03:48:59.887300Z",
     "start_time": "2019-10-08T03:48:59.442Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "dense_vec: [1.0,0.0,0.0,2.0,0.0]\n",
      "dense_vec.numActives: 5\n",
      "dense_vec.toSparse: (5,[0,3],[1.0,2.0])\n",
      "sparse_vec: (5,[0,3],[1.0,2.0])\n",
      "sparse_vec.toDense: [1.0,0.0,0.0,2.0,0.0]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dense_vec = [1.0,0.0,0.0,2.0,0.0]\n",
       "sparse_vec = (5,[0,3],[1.0,2.0])\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "(5,[0,3],[1.0,2.0])"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.linalg.{DenseVector, SparseVector}\n",
    "\n",
    "\n",
    "//稠密向量\n",
    "val dense_vec = new DenseVector(Array[Double](1, 0, 0, 2.0, 0))\n",
    "\n",
    "println(\"dense_vec: \" + dense_vec)\n",
    "println(\"dense_vec.numActives: \" + dense_vec.numActives)\n",
    "println(\"dense_vec.toSparse: \" + dense_vec.toSparse)\n",
    "\n",
    "//稀疏向量\n",
    "//参数分别是维度，非零索引，非零元素值\n",
    "val sparse_vec = new SparseVector(5, Array(0,3),Array(1.0,2.0))  \n",
    "println(\"sparse_vec: \" + sparse_vec)\n",
    "println(\"sparse_vec.toDense: \" + sparse_vec.toDense)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T03:58:30.571367Z",
     "start_time": "2019-10-08T03:58:30.129Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sparse_matrix.toDense: \n",
      "1.0  0.0  4.0  \n",
      "0.0  3.0  5.0  \n",
      "2.0  0.0  6.0  \n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "dense_matrix = \n",
       "sparse_matrix = \n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "1.0  2.0\n",
       "3.0  4.0\n",
       "5.0  6.0\n",
       "3 x 3 CSCMatrix\n",
       "(0,0) 1.0\n",
       "(2,0) 2.0\n",
       "(1,1) 3.0\n",
       "(0,2) 4.0\n",
       "(1,2) 5.0\n",
       "(2,2) 6.0\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import org.apache.spark.ml.linalg.{DenseMatrix, SparseMatrix}\n",
    "\n",
    "//稠密矩阵\n",
    "//参数分别是 行数，列数，元素值，是否转置(默认false)\n",
    "val dense_matrix = new DenseMatrix(3, 2, Array[Double](1, 3, 5, 2, 4, 6))\n",
    "\n",
    "\n",
    "//稀疏矩阵\n",
    "//参数分别是 行数，列数，在第几个元素列索引加1，行索引，非零元素值\n",
    "val sparse_matrix = new SparseMatrix(3, 3, Array[Int](0, 2, 3, 6),\n",
    "    Array[Int](0, 2, 1, 0, 1, 2), Array[Double](1.0, 2.0, 3.0, 4.0, 5.0, 6.0))\n",
    "\n",
    "println(\"sparse_matrix.toDense: \\n\" +sparse_matrix.toDense)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T04:06:11.584802Z",
     "start_time": "2019-10-08T04:06:11.167Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "vec = [0.0,0.0,0.0]\n",
       "matrix = \n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "0.5318793063722395   0.7928825443333104   0.7984188008281995    0.41387367086711535\n",
       "0.9027126524050961   0.03557310080289733  0.4874390385328311    0.8542742227220153\n",
       "0.12270343464829603  0.7590682837455923   0.5515158568532805    0.023954956368488545\n",
       "0.5282288284504489   0.6363292026370957   0.002645551687648906  0.26078775055502257\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import org.apache.spark.ml.linalg.{Vectors,Matrices}\n",
    "\n",
    "//工厂方法\n",
    "val vec = Vectors.zeros(3)\n",
    "val matrix = Matrices.rand(4,4,new java.util.Random(2123456))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**2,数理统计**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T06:20:46.349681Z",
     "start_time": "2019-10-08T06:20:45.049Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Pearson correlation matrix:\n",
      " 1.0                   0.055641488407465814  NaN  0.4004714203168137  \n",
      "0.055641488407465814  1.0                   NaN  0.9135958615342522  \n",
      "NaN                   NaN                   1.0  NaN                 \n",
      "0.4004714203168137    0.9135958615342522    NaN  1.0                 \n",
      "Spearman correlation matrix:\n",
      " 1.0                  0.10540925533894532  NaN  0.40000000000000174  \n",
      "0.10540925533894532  1.0                  NaN  0.9486832980505141   \n",
      "NaN                  NaN                  1.0  NaN                  \n",
      "0.40000000000000174  0.9486832980505141   NaN  1.0                  \n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = List((4,[0,3],[1.0,-2.0]), [4.0,5.0,0.0,3.0], [6.0,7.0,0.0,8.0], (4,[0,3],[9.0,1.0]))\n",
       "df = [features: vector]\n",
       "coeff1 = \n",
       "coeff2 = \n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "1.0                   0.055641488407465814  NaN  0.4004714203168137\n",
       "0.055641488407465814  1.0                   NaN  0.9135958615342522\n",
       "NaN                   NaN                   1.0  NaN\n",
       "0.4004714203168137    0.9135958615342522    NaN  1.0\n",
       "1.0                  0.10540925533894532  NaN  0.40000000000000174\n",
       "0.10540925533894532  1.0                  NaN  0.9486832980505141\n",
       "NaN                  NaN   ...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "//相关性分析\n",
    "import org.apache.spark.ml.linalg.Vectors\n",
    "import org.apache.spark.ml.stat.Correlation\n",
    "import org.apache.spark.sql.Row\n",
    "\n",
    "val data = Seq(\n",
    "  Vectors.sparse(4, Seq((0, 1.0), (3, -2.0))),\n",
    "  Vectors.dense(4.0, 5.0, 0.0, 3.0),\n",
    "  Vectors.dense(6.0, 7.0, 0.0, 8.0),\n",
    "  Vectors.sparse(4, Seq((0, 9.0), (3, 1.0)))\n",
    ")\n",
    "\n",
    "val df = data.map(Tuple1.apply).toDF(\"features\")\n",
    "val Row(coeff1: Matrix) = Correlation.corr(df, \"features\").head\n",
    "println(s\"Pearson correlation matrix:\\n $coeff1\")\n",
    "\n",
    "val Row(coeff2: Matrix) = Correlation.corr(df, \"features\", \"spearman\").head\n",
    "println(s\"Spearman correlation matrix:\\n $coeff2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-10-08T06:21:06.038768Z",
     "start_time": "2019-10-08T06:21:05.029Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "pValues = [0.6872892787909721,0.6822703303362126]\n",
      "degreesOfFreedom [2,3]\n",
      "statistics [0.75,1.5]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "data = List((0.0,[0.5,10.0]), (0.0,[1.5,20.0]), (1.0,[1.5,30.0]), (0.0,[3.5,30.0]), (0.0,[3.5,40.0]), (1.0,[3.5,40.0]))\n",
       "df = [label: double, features: vector]\n",
       "chi = [[0.6872892787909721,0.6822703303362126],WrappedArray(2, 3),[0.75,1.5]]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[[0.6872892787909721,0.6822703303362126],WrappedArray(2, 3),[0.75,1.5]]"
      ]
     },
     "execution_count": 78,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//卡方检验\n",
    "import org.apache.spark.ml.linalg.{Vector, Vectors}\n",
    "import org.apache.spark.ml.stat.ChiSquareTest\n",
    "\n",
    "val data = Seq(\n",
    "  (0.0, Vectors.dense(0.5, 10.0)),\n",
    "  (0.0, Vectors.dense(1.5, 20.0)),\n",
    "  (1.0, Vectors.dense(1.5, 30.0)),\n",
    "  (0.0, Vectors.dense(3.5, 30.0)),\n",
    "  (0.0, Vectors.dense(3.5, 40.0)),\n",
    "  (1.0, Vectors.dense(3.5, 40.0))\n",
    ")\n",
    "\n",
    "val df = data.toDF(\"label\", \"features\")\n",
    "val chi = ChiSquareTest.test(df, \"features\", \"label\").head\n",
    "println(s\"pValues = ${chi.getAs[Vector](0)}\")\n",
    "println(s\"degreesOfFreedom ${chi.getSeq[Int](1).mkString(\"[\", \",\", \"]\")}\")\n",
    "println(s\"statistics ${chi.getAs[Vector](2)}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Spark - Scala",
   "language": "scala",
   "name": "spark_scala"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "2.11.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}