4.Spark特征提取、转换和选择 - 简书
- -在实际机器学习项目中，我们获取的数据往往是不规范、不一致、有很多缺失数据，甚至不少错误数据，这些数据有时又称为脏数据或噪音，在模型训练前，务必对这些脏数据进行处理，否则，再好的模型，也只能脏数据进，脏数据出. 这章我们主要介绍对数据处理涉及的一些操作，主要包括：. 特征提取一般指从原始数据中抽取特征.
4.Spark特征提取、转换和选择 - 简书
标签：
| 发表时间：2018-08-13 15:08 | 作者：
出处：https://www.jianshu.com
在实际机器学习项目中，我们获取的数据往往是不规范、不一致、有很多缺失数据，甚至不少错误数据，这些数据有时又称为脏数据或噪音，在模型训练前，务必对这些脏数据进行处理，否则，再好的模型，也只能脏数据进，脏数据出。
这章我们主要介绍对数据处理涉及的一些操作，主要包括：
- 特征提取
- 特征转换
- 特征选择
4.1 特征提取
特征提取一般指从原始数据中抽取特征。
4.1.1 词频－逆向文件频率（TF-IDF）
词频－逆向文件频率（TF-IDF）是一种在文本挖掘中广泛使用的特征向量化方法，它可以体现一个文档中词语在语料库中的重要程度。
在下面的代码段中，我们以一组句子开始。首先使用分解器Tokenizer把句子划分为单个词语。对每一个句子（词袋），我们使用HashingTF将句子转换为特征向量，最后使用IDF重新调整特征向量。这种转换通常可以提高使用文本特征的性能。
import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
val sentenceData = spark.createDataFrame(Seq(
(0, "Hi I heard about Spark"),
(0, "I wish Java could use case classes"),
(1, "Logistic regression models are neat")
)).toDF("label", "sentence")
val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words")
val wordsData = tokenizer.transform(sentenceData)
val hashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(20)
val featurizedData = hashingTF.transform(wordsData)
// CountVectorizer也可获取词频向量
val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
val idfModel = idf.fit(featurizedData)
val rescaledData = idfModel.transform(featurizedData)
rescaledData.select("features", "label").take(3).foreach(println) 4.1.2 Word2Vec
Word2vec是一个Estimator，它采用一系列代表文档的词语来训练word2vecmodel。 在下面的代码段中，我们首先用一组文档，其中每一个文档代表一个词语序列。对于每一个文档，我们将其转换为一个特征向量。此特征向量可以被传递到一个学习算法。
import org.apache.spark.ml.feature.Word2Vec
// 输入数据，每行为一个词袋，可来自语句或文档。
val documentDF = spark.createDataFrame(Seq(
"Hi I heard about Spark".split(" "),
"I wish Java could use case classes".split(" "),
"Logistic regression models are neat".split(" ")
).map(Tuple1.apply)).toDF("text")
//训练从词到向量的映射
val word2Vec = new Word2Vec()
.setInputCol("text")
.setOutputCol("result")
.setVectorSize(3)
.setMinCount(0)
val model = word2Vec.fit(documentDF)
val result = model.transform(documentDF)
result.select("result").take(3).foreach(println) 4.1.3 计数向量器
计数向量器（Countvectorizer）和计数向量器模型（Countvectorizermodel）旨在通过计数来将一个文档转换为向量。
以下用实例来说明计数向量器的使用。
假设有以下列id和texts构成的DataFrame：
| id | texts |
|---|---|
| 0 | Array("a", "b", "c") |
| 1 | Array("a", "b", "b", "c", "a") |
每行text都是Array [String]类型的文档。调用fit，CountVectorizer产生CountVectorizerModel含词汇（a，b，c）。转换后的输出列“向量”包含：
调用的CountVectorizer产生词汇(a,b,c)的CountVectorizerModel，转换后的输出向量如下：
| id | texts | vector |
|---|---|---|
| 0 | Array("a", "b", "c") | (3,[0,1,2],[1.0,1.0,1.0]) |
| 1 | Array("a", "b", "b", "c", "a") | (3,[0,1,2],[2.0,2.0,1.0]) |
每个向量代表文档的词汇表中每个词语出现的次数。
import org.apache.spark.ml.feature.{CountVectorizer,CountVectorizerModel}
val df = spark.createDataFrame(Seq(
(0,Array("a","b","c")),
(1,Array("a","b","b","c","a"))
)).toDF("id","words")
//从语料库中拟合CountVectorizerModel
val cvModel:CountVectorizerModel=newCountVectorizer()
.setInputCol("words")
.setOutputCol("features")
.setVocabSize(3)
.setMinDF(2)
.fit(df)
//或者，用先验词汇表定义CountVectorizerModel
val cvm =newCountVectorizerModel(Array("a","b","c"))
.setInputCol("words")
.setOutputCol("features")
cvModel.transform(df).show(false)
+---+---------------+-------------------------+
|id |words |features |
+---+---------------+-------------------------+
|0 |[a, b, c] |(3,[0,1,2],[1.0,1.0,1.0])|
|1 |[a, b, b, c, a]|(3,[0,1,2],[2.0,2.0,1.0])|
+---+---------------+-------------------------+ 4.2 特征转换
在机器学习中，数据处理是一件比较繁琐的事情，需要对原有特征做多种处理，如类型转换、标准化特征、新增衍生特征等等，需要耗费大量的时间和精力编写处理程序，不过，自从Spark推出ML后，情况大有改观，Spark ML包中提供了很多现成转换器，例如：StringIndexer、IndexToString、OneHotEncoder、VectorIndexer，它们提供了十分方便的特征转换功能，这些转换器类都位org.apache.spark.ml.feature包下。
4.2.1分词器
分词器（Tokenization）将文本划分为独立个体（通常为单词）。
import org.apache.spark.ml.feature.{RegexTokenizer,Tokenizer}
import org.apache.spark.sql.functions._
val sentenceDataFrame = spark.createDataFrame(Seq(
(0,"Hi I heard about Spark"),
(1,"I wish Java could use case classes"),
(2,"Logistic,regression,models,are,neat")
)).toDF("id","sentence")
val tokenizer =newTokenizer().setInputCol("sentence").setOutputCol("words")
val regexTokenizer =newRegexTokenizer()
.setInputCol("sentence")
.setOutputCol("words")
.setPattern("\\W")//或者使用 .setPattern("\\w+").setGaps(false)
val countTokens = udf {(words:Seq[String])=> words.length }
val tokenized = tokenizer.transform(sentenceDataFrame)
tokenized.select("sentence","words")
.withColumn("tokens", countTokens(col("words"))).show(false)
+-----------------------------------+------------------------------------------+------+
|sentence |words |tokens|
+-----------------------------------+------------------------------------------+------+
|Hi I heard about Spark |[hi, i, heard, about, spark] |5 |
|I wish Java could use case classes |[i, wish, java, could, use, case, classes]|7 |
|Logistic,regression,models,are,neat|[logistic,regression,models,are,neat] |1 |
+-----------------------------------+------------------------------------------+------+
val regexTokenized = regexTokenizer.transform(sentenceDataFrame)
regexTokenized.select("sentence","words")
.withColumn("tokens", countTokens(col("words"))).show(false)
+-----------------------------------+------------------------------------------+------+
|sentence |words |tokens|
+-----------------------------------+------------------------------------------+------+
|Hi I heard about Spark |[hi, i, heard, about, spark] |5 |
|I wish Java could use case classes |[i, wish, java, could, use, case, classes]|7 |
|Logistic,regression,models,are,neat|[logistic, regression, models, are, neat] |5 |
+-----------------------------------+------------------------------------------+------+ 4.2.2 移除停用词
停用词为在文档中频繁出现，但未承载太多意义的词语，它们不应该被包含在算法输入中，所以会用到移除停用词（StopWordsRemover）。
示例：
假设我们有如下DataFrame，有id和raw两列
| id | raw |
|---|---|
| 0 | [I, saw, the, red, baloon] |
| 1 | [Mary, had, a, little, lamb] |
通过对raw列调用StopWordsRemover，我们可以得到筛选出的结果列如下
| id | raw | filtered |
|---|---|---|
| 0 | [I, saw, the, red, baloon] | [saw, red, baloon] |
| 1 | [Mary, had, a, little, lamb] | [Mary, little, lamb] |
其中，“I”, “the”, “had”以及“a”被移除。
实现以上功能的详细代码：
import org.apache.spark.ml.feature.StopWordsRemover
val remover = new StopWordsRemover()
.setInputCol("raw")
.setOutputCol("filtered")
val dataSet = spark.createDataFrame(Seq(
(0, Seq("I", "saw", "the", "red", "balloon")),
(1, Seq("Mary", "had", "a", "little", "lamb"))
)).toDF("id", "raw")
remover.transform(dataSet).show(false) 4.2.3 n-gram
一个n-gram是一个长度为整数n的字序列。NGram可以用来将输入转换为n-gram。
import org.apache.spark.ml.feature.NGram
val wordDataFrame = spark.createDataFrame(Seq(
(0, Array("Hi", "I", "heard", "about", "Spark")),
(1, Array("I", "wish", "Java", "could", "use", "case", "classes")),
(2, Array("Logistic", "regression", "models", "are", "neat"))
)).toDF("id", "words")
val ngram = new NGram().setN(2).setInputCol("words").setOutputCol("ngrams")
val ngramDataFrame = ngram.transform(wordDataFrame)
ngramDataFrame.select("ngrams").show(false)
+------------------------------------------------------------------+
|ngrams |
+------------------------------------------------------------------+
|[Hi I, I heard, heard about, about Spark] |
|[I wish, wish Java, Java could, could use, use case, case classes]|
|[Logistic regression, regression models, models are, are neat] |
+------------------------------------------------------------------+ 4.2.4 二值化
二值化，通过设置阀值，将连续型的特征转化为两个值。大于阀值为1，否则为0。
注：以下规范化操作一般是针对一个特征向量（dataFrame中的一个colum）来操作的。
import org.apache.spark.ml.feature.Binarizer
val data = Array((0, 0.1), (1, 0.8), (2, 0.2))
val dataFrame = spark.createDataFrame(data).toDF("id", "feature")
+---+-------+
| id|feature|
+---+-------+
| 0| 0.1|
| 1| 0.8|
| 2| 0.2|
+---+-------+
val binarizer: Binarizer = new Binarizer()
.setInputCol("feature")
.setOutputCol("binarized_feature")
.setThreshold(0.5)
val binarizedDataFrame = binarizer.transform(dataFrame)
println(s"Binarizer output with Threshold = ${binarizer.getThreshold}")
binarizedDataFrame.show()
+---+-------+-----------------+
| id|feature|binarized_feature|
+---+-------+-----------------+
| 0| 0.1| 0.0|
| 1| 0.8| 1.0|
| 2| 0.2| 0.0|
+---+-------+-----------------+ 4.2.5 主成分分析
主成分分析被广泛应用在各种统计学、机器学习问题中，是最常见的降维方法之一。
PCA在Spark2.0用法比较简单，只需要设置：
.setInputCol(“features”)//保证输入是特征值向量
.setOutputCol(“pcaFeatures”)//输出
.setK(3)//主成分个数
注意：PCA前一定要对特征向量进行规范化（标准化）！！！ import org.apache.spark.ml.feature.PCA
import org.apache.spark.ml.feature.PCAModel//不是mllib
import org.apache.spark.ml.feature.StandardScaler
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
val rawDataFrame=spark.read.format("libsvm").load("file:///home/hadoop/bigdata/spark/data/mllib/sample_libsvm_data.txt")
val scaledDataFrame=new StandardScaler()
.setInputCol("features")
.setOutputCol("scaledFeatures")
.setWithMean(false)//对于稀疏数据（如本次使用的数据），不要使用平均值
.setWithStd(true)
.fit(rawDataFrame)
.transform(rawDataFrame)
//PCA Model
val pcaModel=new PCA().setInputCol("scaledFeatures")
.setOutputCol("pcaFeatures")
.setK(3)//
.fit(scaledDataFrame)
//进行PCA降维
pcaModel.transform(scaledDataFrame).select("label","pcaFeatures").show(10,false)
//没有标准化特征向量，直接进行PCA主成分：各主成分之间值变化太大，有数量级的差别。
//标准化特征向量后PCA主成分，各主成分之间值基本上在同一水平上，结果更合理
+-----+-------------------------------------------------------------+
|label|pcaFeatures |
+-----+-------------------------------------------------------------+
|0.0 |[-14.998868464839624,-10.137788261664621,-3.042873539670117] |
|1.0 |[2.1965800525589754,-4.139257418439533,-11.386135042845101] |
|1.0 |[1.0254645688925883,-0.8905813756164163,7.168759904518129] |
|1.0 |[1.5069317554093433,-0.7289177578028571,5.23152743564543] |
|1.0 |[1.6938250375084654,-0.4350617717494331,4.770263568537382] |
|0.0 |[-15.870371979062549,-9.999445137658528,-6.521920373215663]
//如何选择k值？
val pcaModel=new PCA().setInputCol("scaledFeatures")
.setOutputCol("pcaFeatures")
.setK(50)//
.fit(scaledDataFrame)
var i=1
for( x<-pcaModel.explainedVariance.toArray){
println(i+"\t"+x+" ")
i +=1
}
//运行结果（前10行），随着k的增加，精度趋于平稳。
1 0.25934799275530857
2 0.12355355301486977
3 0.07447670060988294
4 0.0554545717486928
5 0.04207050513264405
6 0.03715986573644129
7 0.031350566055423544
8 0.027797304129489515
9 0.023825873477496748
10 0.02268054946233242 4.2.6 多项式展开
多项式展开(PolynomialExpansion)即通过产生n维组合将原始特征将特征扩展到多项式空间。下面的示例会介绍如何将你的特征集拓展到3维多项式空间。
import org.apache.spark.ml.feature.PolynomialExpansion
import org.apache.spark.ml.linalg.Vectors
val data = Array(
Vectors.dense(2.0, 1.0),
Vectors.dense(0.0, 0.0),
Vectors.dense(3.0, -1.0)
)
val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val polyExpansion = new PolynomialExpansion()
.setInputCol("features")
.setOutputCol("polyFeatures")
.setDegree(3)
val polyDF = polyExpansion.transform(df)
polyDF.show(false)
+----------+------------------------------------------+
|features |polyFeatures |
+----------+------------------------------------------+
|[2.0,1.0] |[2.0,4.0,8.0,1.0,2.0,4.0,1.0,2.0,1.0] |
|[0.0,0.0] |[0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0] |
|[3.0,-1.0]|[3.0,9.0,27.0,-1.0,-3.0,-9.0,1.0,3.0,-1.0]|
+----------+------------------------------------------+ 4.2.7 离散余弦变换
离散余弦变换（DCT）是与傅里叶变换相关的一种变换，它类似于离散傅立叶变换，但是只使用实数。
import org.apache.spark.ml.feature.DCT
import org.apache.spark.ml.linalg.Vectors
val data = Seq(
Vectors.dense(0.0, 1.0, -2.0, 3.0),
Vectors.dense(-1.0, 2.0, 4.0, -7.0),
Vectors.dense(14.0, -2.0, -5.0, 1.0))
val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val dct = new DCT()
.setInputCol("features")
.setOutputCol("featuresDCT")
.setInverse(false)
val dctDf = dct.transform(df)
dctDf.select("featuresDCT").show(false)
+----------------------------------------------------------------+
|featuresDCT |
+----------------------------------------------------------------+
|[1.0,-1.1480502970952693,2.0000000000000004,-2.7716385975338604]|
|[-1.0,3.378492794482933,-7.000000000000001,2.9301512653149677] |
|[4.0,9.304453421915744,11.000000000000002,1.5579302036357163] |
+----------------------------------------------------------------+ 4.2.8 字符串－索引变换
| id | category |
|---|---|
| 0 | a |
| 1 | b |
| 2 | c |
| 3 | a |
| 4 | a |
| 5 | c |
字符串—索引变换（StringIndexer）是将字符串列编码为标签索引列。示例数据为一个含有id和category两列的DataFrame
| id | category |
|---|---|
| 0 | a |
| 1 | b |
| 2 | c |
| 3 | a |
| 4 | a |
| 5 | c |
| id | category | categoryIndex |
|---|---|---|
| 0 | a | 0.0 |
| 1 | b | 2.0 |
| 2 | c | 1.0 |
| 3 | a | 0.0 |
| 4 | a | 0.0 |
| 5 | c | 1.0 |
category是有3种取值的字符串列（a、b、c），使用StringIndexer进行转换后我们可以得到如下输出，其中category作为输入列，categoryIndex作为输出列：
| id | category | categoryIndex |
|---|---|---|
| 0 | a | 0.0 |
| 1 | b | 2.0 |
| 2 | c | 1.0 |
| 3 | a | 0.0 |
| 4 | a | 0.0 |
| 5 | c | 1.0 |
| id | category |
|---|---|
| 0 | a |
| 1 | b |
| 2 | c |
| 3 | d |
a获得索引0，因为它是最频繁的，随后是具有索引1的c和具有索引2的b。
如果测试数据集中比训练数据集多了一个d类：
| id | category |
|---|---|
| 0 | a |
| 1 | b |
| 2 | c |
| 3 | d |
| id | category | categoryIndex |
|---|---|---|
| 0 | a | 0.0 |
| 1 | b | 2.0 |
| 2 | c | 1.0 |
如果您没有设置StringIndexer如何处理未看见的标签（默认值）或将其设置为“错误”，则会抛出异常。 但是，如果您调用了setHandleInvalid（“skip”），d类将不出现，结果为以下数据集：
| id | category | categoryIndex |
|---|---|---|
| 0 | a | 0.0 |
| 1 | b | 2.0 |
| 2 | c | 1.0 |
以下是使用StringIndexer的一个示例：
import org.apache.spark.ml.feature.StringIndexer
val df = spark.createDataFrame(
Seq((0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c"))
).toDF("id", "category")
val indexer = new StringIndexer()
.setInputCol("category")
.setOutputCol("categoryIndex")
val indexed = indexer.fit(df).transform(df)
indexed.show()
+---+--------+-------------+
| id|category|categoryIndex|
+---+--------+-------------+
| 0| a| 0.0|
| 1| b| 2.0|
| 2| c| 1.0|
| 3| a| 0.0|
| 4| a| 0.0|
| 5| c| 1.0|
+---+--------+-------------+ 4.2.9 索引－字符串变换
| id | categoryIndex |
|---|---|
| 0 | 0.0 |
| 1 | 2.0 |
| 2 | 1.0 |
| 3 | 0.0 |
| 4 | 0.0 |
| 5 | 1.0 |
与StringIndexer对应，索引—字符串变换（IndexToString）是将指标标签映射回原始字符串标签。
| id | categoryIndex |
|---|---|
| 0 | 0.0 |
| 1 | 2.0 |
| 2 | 1.0 |
| 3 | 0.0 |
| 4 | 0.0 |
| 5 | 1.0 |
| id | categoryIndex | originalCategory |
|---|---|---|
| 0 | 0.0 | a |
| 1 | 2.0 | b |
| 2 | 1.0 | c |
| 3 | 0.0 | a |
| 4 | 0.0 | a |
| 5 | 1.0 | c |
应用IndexToString，将categoryIndex作为输入列，将originalCategory作为输出列，我们可以检索我们的原始标签（它们将从列的元数据中推断）：
| id | categoryIndex | originalCategory |
|---|---|---|
| 0 | 0.0 | a |
| 1 | 2.0 | b |
| 2 | 1.0 | c |
| 3 | 0.0 | a |
| 4 | 0.0 | a |
| 5 | 1.0 | c |
以下是以上整个过程的一个实例：
import org.apache.spark.ml.attribute.Attribute
import org.apache.spark.ml.feature.{IndexToString, StringIndexer}
val df = spark.createDataFrame(Seq(
(0, "a"),
(1, "b"),
(2, "c"),
(3, "a"),
(4, "a"),
(5, "c")
)).toDF("id", "category")
val indexer = new StringIndexer()
.setInputCol("category")
.setOutputCol("categoryIndex")
.fit(df)
val indexed = indexer.transform(df)
println(s"Transformed string column '${indexer.getInputCol}' " +
s"to indexed column '${indexer.getOutputCol}'")
indexed.show()
val inputColSchema = indexed.schema(indexer.getOutputCol)
println(s"StringIndexer will store labels in output column metadata: " +
s"${Attribute.fromStructField(inputColSchema).toString}\n")
val converter = new IndexToString()
.setInputCol("categoryIndex")
.setOutputCol("originalCategory")
val converted = converter.transform(indexed)
println(s"Transformed indexed column '${converter.getInputCol}' back to original string " +
s"column '${converter.getOutputCol}' using labels in metadata")
converted.select("id", "categoryIndex", "originalCategory").show() 4.2.10 独热编码
独热编码（OneHotEncoder）将标签指标映射为二值向量，其中最多一个单值。
import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer}
val df = spark.createDataFrame(Seq(
(0, "a"),
(1, "b"),
(2, "c"),
(3, "a"),
(4, "a"),
(5, "c")
)).toDF("id", "category")
val indexer = new StringIndexer()
.setInputCol("category")
.setOutputCol("categoryIndex")
.fit(df)
val indexed = indexer.transform(df)
val encoder = new OneHotEncoder()
.setInputCol("categoryIndex")
.setOutputCol("categoryVec")
.setDropLast(false)
val encoded = encoder.transform(indexed)
encoded.show() 【说明】
1、OneHotEncoder缺省状态下将删除最后一个分类或把最后一个分类作为0.
//示例
import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer}
val fd = spark.createDataFrame( Seq((1.0, "a"), (1.5, "a"), (10.0, "b"), (3.2, "c"),(3.8,"c"))).toDF("x","c")
val ss =new StringIndexer().setInputCol("c").setOutputCol("c_idx")
val ff = ss.fit(fd).transform(fd)
ff.show() 显示结果如下：
+----+---+-----+
| x| c|c_idx|
+----+---+-----+
| 1.0| a| 0.0|
| 1.5| a| 0.0|
|10.0| b| 2.0|
| 3.2| c| 1.0|
| 3.8| c| 1.0|
+----+---+-----+ 最后一个分类为b，通过OneHotEncoder变为向量后，已被删除。
val oe = new OneHotEncoder().setInputCol("c_idx").setOutputCol("c_idx_vec")
val fe = oe.transform(ff)
fe.show() 显示结果如下：
+----+---+-----+-------------+
| x| c|c_idx| c_idx_vec|
+----+---+-----+-------------+
| 1.0| a| 0.0|(2,[0],[1.0])|
| 1.5| a| 0.0|(2,[0],[1.0])|
|10.0| b| 2.0| (2,[],[])|
| 3.2| c| 1.0|(2,[1],[1.0])|
| 3.8| c| 1.0|(2,[1],[1.0])|
+----+---+-----+-------------+ 与其他特征组合为特征向量后，将置为0，请看下例
val assembler = new VectorAssembler().setInputCols(Array("x", "c_idx", "c_idx_vec")).setOutputCol("features")
val vecDF: DataFrame = assembler.transform(fe)
vecDF.show(false) 显示结果如下：
+----+---+-----+-------------+------------------+
|x |c |c_idx|c_idx_vec |features |
+----+---+-----+-------------+------------------+
|1.0 |a |0.0 |(2,[0],[1.0])|[1.0,0.0,1.0,0.0] |
|1.5 |a |0.0 |(2,[0],[1.0])|[1.5,0.0,1.0,0.0] |
|10.0|b |2.0 |(2,[],[]) |[10.0,2.0,0.0,0.0]|
|3.2 |c |1.0 |(2,[1],[1.0])|[3.2,1.0,0.0,1.0] |
|3.8 |c |1.0 |(2,[1],[1.0])|[3.8,1.0,0.0,1.0] |
+----+---+-----+-------------+------------------+ 如果想不删除最后一个分类，可添加setDropLast(False)。
oe.setDropLast(false)
val fl = oe.transform(ff)
fl.show() 显示结果如下：
+----+---+-----+-------------+
| x| c|c_idx| c_idx_vec|
+----+---+-----+-------------+
| 1.0| a| 0.0|(3,[0],[1.0])|
| 1.5| a| 0.0|(3,[0],[1.0])|
|10.0| b| 2.0|(3,[2],[1.0])|
| 3.2| c| 1.0|(3,[1],[1.0])|
| 3.8| c| 1.0|(3,[1],[1.0])|
+----+---+-----+-------------+ 与其他特征向量结合后，情况如下：
val vecDFl: DataFrame = assembler.transform(fl)
vecDFl.show(false) 显示结果如下：
+----+---+-----+-------------+----------------------+
|x |c |c_idx|c_idx_vec |features |
+----+---+-----+-------------+----------------------+
|1.0 |a |0.0 |(3,[0],[1.0])|(5,[0,2],[1.0,1.0]) |
|1.5 |a |0.0 |(3,[0],[1.0])|(5,[0,2],[1.5,1.0]) |
|10.0|b |2.0 |(3,[2],[1.0])|[10.0,2.0,0.0,0.0,1.0]|
|3.2 |c |1.0 |(3,[1],[1.0])|[3.2,1.0,0.0,1.0,0.0] |
|3.8 |c |1.0 |(3,[1],[1.0])|[3.8,1.0,0.0,1.0,0.0] |
+----+---+-----+-------------+----------------------+ 2、如果分类中出现空字符，需要进行处理，如设置为"None",否则会报错。
4.2.11 向量－索引变换
在下面的例子中，我们读取一个标记点的数据集，然后使用VectorIndexer来决定哪些特征应该被视为分类。我们将分类特征值转换为它们的索引。这个变换的数据然后可以被传递到诸如DecisionTreeRegressor的处理分类特征的算法。
import org.apache.spark.ml.feature.VectorIndexer
val data = spark.read.format("libsvm").load("file:///home/hadoop/bigdata/spark/data/mllib/sample_libsvm_data.txt")
val indexer = new VectorIndexer()
.setInputCol("features")
.setOutputCol("indexed")
.setMaxCategories(10)
val indexerModel = indexer.fit(data)
val categoricalFeatures: Set[Int] = indexerModel.categoryMaps.keys.toSet
println(s"Chose ${categoricalFeatures.size} categorical features: " +
categoricalFeatures.mkString(", "))
// Create new column "indexed" with categorical values transformed to indices
val indexedData = indexerModel.transform(data)
indexedData.show()
+-----+--------------------+--------------------+
|label| features| indexed|
+-----+--------------------+--------------------+
| 0.0|(692,[127,128,129...|(692,[127,128,129...|
| 1.0|(692,[158,159,160...|(692,[158,159,160...|
| 1.0|(692,[124,125,126...|(692,[124,125,126...|
| 1.0|(692,[152,153,154...|(692,[152,153,154...|
| 1.0|(692,[151,152,153...|(692,[151,152,153...|
| 0.0|(692,[129,130,131...|(692,[129,130,131...| 4.2.12交互式
例子,假设我们有以下DataFrame的列“id1”，“vec1”和“vec2”
| id1 | vec1 | vec2 |
|---|---|---|
| 1 | [1.0,2.0,3.0] | [8.0,4.0,5.0] |
| 2 | [4.0,3.0,8.0] | [7.0,9.0,8.0] |
| 3 | [6.0,1.0,9.0] | [2.0,3.0,6.0] |
| 4 | [10.0,8.0,6.0] | [9.0,4.0,5.0] |
| 5 | [9.0,2.0,7.0] | [10.0,7.0,3.0] |
| 6 | [1.0,1.0,4.0] | [2.0,8.0,4.0] |
应用与这些输入列的交互，然后interactionedCol作为输出列包含：
| id1 | vec1 | vec2 | interactedCol |
|---|---|---|---|
| 1 | [1.0,2.0,3.0] | [8.0,4.0,5.0] | [8.0,4.0,5.0,16.0,8.0,10.0,24.0,12.0,15.0] |
| 2 | [4.0,3.0,8.0] | [7.0,9.0,8.0] | [56.0,72.0,64.0,42.0,54.0,48.0,112.0,144.0,128.0] |
| 3 | [6.0,1.0,9.0] | [2.0,3.0,6.0] | [36.0,54.0,108.0,6.0,9.0,18.0,54.0,81.0,162.0] |
| 4 | [10.0,8.0,6.0] | [9.0,4.0,5.0] | [360.0,160.0,200.0,288.0,128.0,160.0,216.0,96.0,120.0] |
| 5 | [9.0,2.0,7.0] | [10.0,7.0,3.0] | [450.0,315.0,135.0,100.0,70.0,30.0,350.0,245.0,105.0] |
| 6 | [1.0,1.0,4.0] | [2.0,8.0,4.0] | [12.0,48.0,24.0,12.0,48.0,24.0,48.0,192.0,96.0] |
以下是实现以上转换的具体代码：
import org.apache.spark.ml.feature.Interaction
import org.apache.spark.ml.feature.VectorAssembler
val df = spark.createDataFrame(Seq(
(1, 1, 2, 3, 8, 4, 5),
(2, 4, 3, 8, 7, 9, 8),
(3, 6, 1, 9, 2, 3, 6),
(4, 10, 8, 6, 9, 4, 5),
(5, 9, 2, 7, 10, 7, 3),
(6, 1, 1, 4, 2, 8, 4)
)).toDF("id1", "id2", "id3", "id4", "id5", "id6", "id7")
val assembler1 = new VectorAssembler().
setInputCols(Array("id2", "id3", "id4")).
setOutputCol("vec1")
val assembled1 = assembler1.transform(df)
val assembler2 = new VectorAssembler().
setInputCols(Array("id5", "id6", "id7")).
setOutputCol("vec2")
val assembled2 = assembler2.transform(assembled1).select("id1", "vec1", "vec2")
val interaction = new Interaction()
.setInputCols(Array("id1", "vec1", "vec2"))
.setOutputCol("interactedCol")
val interacted = interaction.transform(assembled2)
interacted.show(truncate = false) 4.2.13正则化
以下示例演示如何加载libsvm格式的数据集，然后将每行标准化为具有单位L1范数和单位L∞范数。
import org.apache.spark.ml.feature.Normalizer
import org.apache.spark.ml.linalg.Vectors
val dataFrame = spark.createDataFrame(Seq(
(0, Vectors.dense(1.0, 0.5, -1.0)),
(1, Vectors.dense(2.0, 1.0, 1.0)),
(2, Vectors.dense(4.0, 10.0, 2.0))
)).toDF("id", "features")
// 使用L^1正规化向量
val normalizer = new Normalizer()
.setInputCol("features")
.setOutputCol("normFeatures")
.setP(1.0)
val l1NormData = normalizer.transform(dataFrame)
println("Normalized using L^1 norm")
l1NormData.show()
//使用L^∞正规化向量..
val lInfNormData = normalizer.transform(dataFrame, normalizer.p -> Double.PositiveInfinity)
println("Normalized using L^inf norm")
lInfNormData.show() 4.2.14规范化(StandardScaler)
以下示例演示如何以libsvm格式加载数据集，然后规范化每个要素的单位标准偏差。
import org.apache.spark.ml.feature.StandardScaler
val dataFrame = spark.read.format("libsvm").load("file:///u01/bigdata/spark/data/mllib/sample_libsvm_data.txt")
val scaler = new StandardScaler()
.setInputCol("features")
.setOutputCol("scaledFeatures")
.setWithStd(true)
.setWithMean(false)
//通过拟合StandardScaler来计算汇总统计
val scalerModel = scaler.fit(dataFrame)
//标准化每个特征使其有单位标准偏差
val scaledData = scalerModel.transform(dataFrame)
scaledData.show() 4.2.15最大值－最小值缩放
下面的示例展示如果读入一个libsvm形式的数据以及调整其特征值到[0,1]之间。
调用示例：
import org.apache.spark.ml.feature.MinMaxScaler
import org.apache.spark.ml.linalg.Vectors
val dataFrame = spark.createDataFrame(Seq(
(0, Vectors.dense(1.0, 0.1, -1.0)),
(1, Vectors.dense(2.0, 1.1, 1.0)),
(2, Vectors.dense(3.0, 10.1, 3.0))
)).toDF("id", "features")
val scaler = new MinMaxScaler()
.setInputCol("features")
.setOutputCol("scaledFeatures")
//进行汇总统计并生成MinMaxScalerModel
val scalerModel = scaler.fit(dataFrame)
//将每个特征重新缩放至[min，max]范围
val scaledData = scalerModel.transform(dataFrame)
println(s"Features scaled to range: [${scaler.getMin}, ${scaler.getMax}]")
scaledData.select("features", "scaledFeatures").show() 显示结果如下：
Features scaled to range: [0.0, 1.0]
+--------------+--------------+
| features|scaledFeatures|
+--------------+--------------+
|[1.0,0.1,-1.0]| [0.0,0.0,0.0]|
| [2.0,1.1,1.0]| [0.5,0.1,0.5]|
|[3.0,10.1,3.0]| [1.0,1.0,1.0]|
+--------------+--------------+ 4.2.16最大值－绝对值缩放
以下示例演示如何加载libsvm格式的数据集，然后将每个特征重新缩放到[-1，1]。
import org.apache.spark.ml.feature.MaxAbsScaler
import org.apache.spark.ml.linalg.Vectors
val dataFrame = spark.createDataFrame(Seq(
(0, Vectors.dense(1.0, 0.1, -8.0)),
(1, Vectors.dense(2.0, 1.0, -4.0)),
(2, Vectors.dense(4.0, 10.0, 8.0))
)).toDF("id", "features")
val scaler = new MaxAbsScaler()
.setInputCol("features")
.setOutputCol("scaledFeatures")
// 进行汇总统计并生成MaxAbsScalerModel
val scalerModel = scaler.fit(dataFrame)
// rescale each feature to range [-1, 1]
val scaledData = scalerModel.transform(dataFrame)
scaledData.select("features", "scaledFeatures").show() 运行结果如下：
+--------------+----------------+
| features| scaledFeatures|
+--------------+----------------+
|[1.0,0.1,-8.0]|[0.25,0.01,-1.0]|
|[2.0,1.0,-4.0]| [0.5,0.1,-0.5]|
|[4.0,10.0,8.0]| [1.0,1.0,1.0]|
+--------------+----------------+ 4.2.17离散化重组
以下示例演示如何将双列列存储到另一个索引列的列中。
import org.apache.spark.ml.feature.Bucketizer
val splits = Array(Double.NegativeInfinity, -0.5, 0.0, 0.5, Double.PositiveInfinity)
val data = Array(-999.9, -0.5, -0.3, 0.0, 0.2, 999.9)
val dataFrame = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val bucketizer = new Bucketizer()
.setInputCol("features")
.setOutputCol("bucketedFeatures")
.setSplits(splits)
// 把原来的数据转换为箱式索引
val bucketedData = bucketizer.transform(dataFrame)
println(s"Bucketizer output with ${bucketizer.getSplits.length-1} buckets")
bucketedData.show() 运行结果如下：
+--------+----------------+
|features|bucketedFeatures|
+--------+----------------+
| -999.9| 0.0|
| -0.5| 1.0|
| -0.3| 1.0|
| 0.0| 2.0|
| 0.2| 2.0|
| 999.9| 3.0|
+--------+----------------+ 4.2.18元素乘积
下面的示例演示了如何使用变换向量值来变换向量
import org.apache.spark.ml.feature.ElementwiseProduct
import org.apache.spark.ml.linalg.Vectors
//创建一些向量数据; 也适用于稀疏向量。
val dataFrame = spark.createDataFrame(Seq(
("a", Vectors.dense(1.0, 2.0, 3.0)),
("b", Vectors.dense(4.0, 5.0, 6.0)))).toDF("id", "vector")
val transformingVector = Vectors.dense(0.0, 1.0, 2.0)
val transformer = new ElementwiseProduct()
.setScalingVec(transformingVector)
.setInputCol("vector")
.setOutputCol("transformedVector")
//创建新列，可批量转换向量。
transformer.transform(dataFrame).show() 运行结果如下：
+---+-------------+-----------------+
| id| vector|transformedVector|
+---+-------------+-----------------+
| a|[1.0,2.0,3.0]| [0.0,2.0,6.0]|
| b|[4.0,5.0,6.0]| [0.0,5.0,12.0]|
+---+-------------+-----------------+ 4.2.19 SQL转换器
假设我们有以下DataFrame和列id，v1和v2
| id | v1 | v2 |
|---|---|---|
| 0 | 1.0 | 3.0 |
| 2 | 2.0 | 5.0 |
这是SQLTransformer "SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM THIS":语句的输出。
| id | v1 | v2 | v3 | v4 |
|---|---|---|---|---|
| 0 | 1.0 | 3.0 | 4.0 | 3.0 |
| 2 | 2.0 | 5.0 | 7.0 | 10.0 |
以下是实现以上结果的具体代码：
import org.apache.spark.ml.feature.SQLTransformer
val df = spark.createDataFrame(
Seq((0, 1.0, 3.0), (2, 2.0, 5.0))).toDF("id", "v1", "v2")
val sqlTrans = new SQLTransformer().setStatement(
"SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM __THIS__")
sqlTrans.transform(df).show() 4.2.20向量汇编
例子
假设我们有一个带有id，hour，mobile，userFeatures和clicked列的DataFrame：
| id | hour | mobile | userFeatures | clicked |
|---|---|---|---|---|
| 0 | 18 | 1.0 | [0.0, 10.0, 0.5] | 1.0 |
userFeatures是一个包含三个用户特征的向量列。我们希望将hour，mobile和userFeatures合并成一个称为特征的单一特征向量，并使用它来预测是否被点击。如果我们将VectorAssembler的输入列设置为hour，mobile和userFeatures，并将列输出到特征，则在转换后，我们应该得到以下DataFrame：
| id | hour | mobile | userFeatures | clicked | features |
|---|---|---|---|---|---|
| 0 | 18 | 1.0 | [0.0, 10.0, 0.5] | 1.0 | [18.0, 1.0, 0.0, 10.0, 0.5] |
以下是实现上述功能的代码：
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.linalg.Vectors
val dataset = spark.createDataFrame(
Seq((0, 18, 1.0, Vectors.dense(0.0, 10.0, 0.5), 1.0))
).toDF("id", "hour", "mobile", "userFeatures", "clicked")
val assembler = new VectorAssembler()
.setInputCols(Array("hour", "mobile", "userFeatures"))
.setOutputCol("features")
val output = assembler.transform(dataset)
println("Assembled columns 'hour', 'mobile', 'userFeatures' to vector column 'features'")
output.select("features", "clicked").show(false) 4.2.21分位数离散化
示例：
假设我们有如下DataFrame包含id,hour：
| id | hour |
|---|---|
| 0 | 18.0 |
| ---- | ------ |
| 1 | 19.0 |
| ---- | ------ |
| 2 | 8.0 |
| ---- | ------ |
| 3 | 5.0 |
| ---- | ------ |
| 4 | 2.2 |
hour是Double类型的连续特征。我们希望将连续特征变成一个分级特征。给定numBuckets = 3，我们可得到以下DataFrame：
| id | hour | result |
|---|---|---|
| 0 | 18.0 | 2.0 |
| ---- | ------ | ------ |
| 1 | 19.0 | 2.0 |
| ---- | ------ | ------ |
| 2 | 8.0 | 1.0 |
| ---- | ------ | ------ |
| 3 | 5.0 | 1.0 |
| ---- | ------ | ------ |
| 4 | 2.2 | 0.0 |
实现以上功能的scala代码如下：
import org.apache.spark.ml.feature.QuantileDiscretizer
val data = Array((0, 18.0), (1, 19.0), (2, 8.0), (3, 5.0), (4, 2.2))
val df = spark.createDataFrame(data).toDF("id", "hour")
val discretizer = new QuantileDiscretizer()
.setInputCol("hour")
.setOutputCol("result")
.setNumBuckets(3)
val result = discretizer.fit(df).transform(df)
result.show() 4.3 特征选择
特征选择（Feature Selection）是从特征向量中选择那些更有效的特征，组成新的、更简单有效的特征向量的过程。它在数据分析中常用使用，尤其在高维数据分析中，可以剔除冗余或影响不大的特征，提升模型的性能。
4.3.1 向量机
假设我们有一个DataFrame与列userFeatures：
userFeatures
[0.0, 10.0, 0.5]
userFeatures是一个包含三个用户特征的向量列。假设userFeature的第一列全部为零，因此我们要删除它并仅选择最后两列。 VectorSlicer使用setIndices（1,2）选择最后两个元素，然后生成一个名为features的新向量列：
| userFeatures | features |
|---|---|
| [0.0, 10.0, 0.5] | [10.0, 0.5] |
假设userFeatures有输入属性，如[“f1”，“f2”，“f3”]，那么我们可以使用setNames（“f2”，“f3”）来选择它们。
| userFeatures | features |
|---|---|
| [0.0, 10.0, 0.5] | [10.0, 0.5] |
| ["f1", "f2", "f3"] | ["f2", "f3"] |
以下是实现向量选择的一个scala代码示例
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
val data = Arrays.asList(
Row(Vectors.sparse(3, Seq((0, -2.0), (1, 2.3)))),
Row(Vectors.dense(-2.0, 2.3, 0.0))
)
val defaultAttr = NumericAttribute.defaultAttr
val attrs = Array("f1", "f2", "f3").map(defaultAttr.withName)
val attrGroup = new AttributeGroup("userFeatures", attrs.asInstanceOf[Array[Attribute]])
val dataset = spark.createDataFrame(data, StructType(Array(attrGroup.toStructField())))
val slicer = new VectorSlicer().setInputCol("userFeatures").setOutputCol("features")
slicer.setIndices(Array(1)).setNames(Array("f3"))
// or slicer.setIndices(Array(1, 2)), or slicer.setNames(Array("f2", "f3"))
val output = slicer.transform(dataset)
output.show(false) 运行结果：
+--------------------+-------------+
|userFeatures |features |
+--------------------+-------------+
|(3,[0,1],[-2.0,2.3])|(2,[0],[2.3])|
|[-2.0,2.3,0.0] |[2.3,0.0] |
+--------------------+-------------+ 4.3.2 R公式
示例：
假设我们有一个DataFrame含有id,country, hour和clicked四列：
| id | country | hour | clicked |
|---|---|---|---|
| 7 | "US" | 18 | 1.0 |
| 8 | "CA" | 12 | 0.0 |
| 9 | "NZ" | 15 | 0.0 |
如果我们使用RFormula公式clicked ~ country+ hour，则表明我们希望基于country和hour预测clicked，通过转换我们可以得到如下DataFrame：
| id | country | hour | clicked | features | label |
|---|---|---|---|---|---|
| 7 | "US" | 18 | 1.0 | [0.0, 0.0, 18.0] | 1.0 |
| 8 | "CA" | 12 | 0.0 | [0.0, 1.0, 12.0] | 0.0 |
| 9 | "NZ" | 15 | 0.0 | [1.0, 0.0, 15.0] | 0.0 |
以下是实现上述功能的scala代码：
import org.apache.spark.ml.feature.RFormula
val dataset = spark.createDataFrame(Seq(
(7, "US", 18, 1.0),
(8, "CA", 12, 0.0),
(9, "NZ", 15, 0.0)
)).toDF("id", "country", "hour", "clicked")
val formula = new RFormula()
.setFormula("clicked ~ country + hour")
.setFeaturesCol("features")
.setLabelCol("label")
val output = formula.fit(dataset).transform(dataset)
output.select("features", "label").show() 4.3.3 卡方特征选择
示例：
假设我们有一个DataFrame含有id,features和clicked三列，其中clicked为需要预测的目标：
| id | features | clicked |
|---|---|---|
| 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 |
如果我们使用ChiSqSelector并设置numTopFeatures为1，根据标签clicked，features中最后一列将会是最有用特征：
| id | features | clicked | selectedFeatures |
|---|---|---|---|
| 7 | [0.0, 0.0, 18.0, 1.0] | 1.0 | [1.0] |
| 8 | [0.0, 1.0, 12.0, 0.0] | 0.0 | [0.0] |
| 9 | [1.0, 0.0, 15.0, 0.1] | 0.0 | [0.1] |
使用ChiSqSelector的scala代码示例：
import org.apache.spark.ml.feature.ChiSqSelector
import org.apache.spark.ml.linalg.Vectors
val data = Seq(
(7, Vectors.dense(0.0, 0.0, 18.0, 1.0), 1.0),
(8, Vectors.dense(0.0, 1.0, 12.0, 0.0), 0.0),
(9, Vectors.dense(1.0, 0.0, 15.0, 0.1), 0.0)
)
val df = spark.createDataset(data).toDF("id", "features", "clicked")
val selector = new ChiSqSelector()
.setNumTopFeatures(1)
.setFeaturesCol("features")
.setLabelCol("clicked")
.setOutputCol("selectedFeatures")
val result = selector.fit(df).transform(df)
println(s"ChiSqSelector output with top ${selector.getNumTopFeatures} features selected")
result.show() 结果显示：
+---+------------------+-------+----------------+
| id| features|clicked|selectedFeatures|
+---+------------------+-------+----------------+
| 7|[0.0,0.0,18.0,1.0]| 1.0| [18.0]|
| 8|[0.0,1.0,12.0,0.0]| 0.0| [12.0]|
| 9|[1.0,0.0,15.0,0.1]| 0.0| [15.0]|
+---+------------------+-------+----------------+ 4.4 小结
本章主要介绍了对数据特征或变量的一些常用操作，包括特征提取，特征转换以及特征选择等方法，这些任务在实际项目中往往花费大量时间和精力，尤其要自己编写这方面的代码或函数，更是如此，Spark ML目前提供了很多现成函数，有效使用这些函数将有助于提供我们开发效率，同时使我们有更多时间优化或提升模型性能。下一章我们将介绍优化或提升模型性能一些方法。
相关 [spark 特征提取 选择] 推荐：
Spark-mllib 文本特征提取算法 - CSDN博客
- -Spark MLlib 提供三种文本特征提取方法，分别为TF-IDF、Word2Vec以及CountVectorizer，. 词频－逆向文件频率（TF-IDF）是一种在文本挖掘中广泛使用的特征向量化方法，它可以体现一个文档中词语在语料库中的重要程度. 词语由t表示，文档由d表示，语料库由D表示. 词频TF(t,,d)是词语t在文档d中出现的次数.
scikit-learn中的文本特征提取
- - 标点符文本分析是机器学习算法的主要应用领域. 由于大部分机器学习算法只能接收固定长度的数值型矩阵特征，导致文本字符串等并不能直接被使用，针对此问题scikit-learn提供了将文本转化为数值型特征的方法，今天就一起来学习下. scikit-learn中的sklearn.feature_extraction.text提供了将文本转化为特征向量的工具：.
【图像分析】低层次特征提取
- - CSDN博客推荐文章本节主要介绍提取低层次特征信息的方要方法. 在有些情形下，这些信息对理解图是足够的. 但是，低层次特征提取函数的目的通常是为更高级分析提供信息. 用于边界，角点和运动提取的方法，非常具有Robustness以及性能特久所必备的特性. 低层次特征是不需要任何形状/空间关系的信息就可以从图像中自动提取的基本特征，如：常用地阈值方法就是作为点处理的一种低层次特征提方式.
浅谈模式识别中的特征提取
- - CSDN博客综合推荐文章 这两天一直在看深度学习的东西，看的头晕脑胀，不过晕乎归晕乎，感觉对模式识别中的特征提取有了更深一点的小理解，暂时记载下来. 突然觉得，模式识别的所有问题都绕不过两个关键门槛，第一是分类器，第二便是特征提取. 而且几乎所有模式识别方面的研究都是在优化这两个问题，要么是造一个更牛的分类器，要么是找出一些表现力更高的特征形式.
图像处理之特征提取 - 简书
- -目前火热的 Deep Learning 会灭绝传统的 SIFT / SURF 特征提取方法吗. 由于之前研究过SIFT和HOG这两种传统的特征提取方法，故本篇文章先对SIFT和HOG作一综述，并比较二者优缺点. 之后，将SIFT和HOG同神经网络特征提取做一对比，浅谈对上述问题的看法. 如果能写得快一些，再简单介绍其他几种传统的特征提取的方法——SURF、ORB、LBP、HAAR等等.
Spark概览
- - 简单文本Spark具有先进的DAG执行引擎，支持cyclic data flow和内存计算. 因此，它的运行速度，在内存中是Hadoop MapReduce的100倍，在磁盘中是10倍. 这样的性能指标，真的让人心动啊. Spark的API更为简单，提供了80个High Level的操作，可以很好地支持并行应用.
Spark与Mapreduce？
- - 崔永键的博客我本人是类似Hive平台的系统工程师，我对MapReduce的熟悉程度是一般，它是我的底层框架. 我隔壁组在实验Spark，想将一部分计算迁移到Spark上. 年初的时候，看Spark的评价，几乎一致表示，Spark是小数据集上处理复杂迭代的交互系统，并不擅长大数据集，也没有稳定性. 但是最近的风评已经变化，尤其是14年10月他们完成了Peta sort的实验，这标志着Spark越来越接近替代Hadoop MapReduce了.
Spark迷思
- - ITeye博客目前在媒体上有很大的关于Apache Spark框架的声音，渐渐的它成为了大数据领域的下一个大的东西. 证明这件事的最简单的方式就是看google的趋势图：. 上图展示的过去两年Hadoop和Spark的趋势. Spark在终端用户之间变得越来越受欢迎，而且这些用户经常在网上找Spark相关资料. 这给了Spark起了很大的宣传作用；同时围绕着它的也有误区和思维错误，而且很多人还把这些误区作为银弹，认为它可以解决他们的问题并提供比Hadoop好100倍的性能.
Spark 优化
- - CSDN博客推荐文章提到Spark与Hadoop的区别，基本最常说的就是Spark采用基于内存的计算方式，尽管这种方式对数据处理的效率很高，但也会往往引发各种各样的问题，Spark中常见的OOM等等. 效率高的特点，注定了Spark对性能的严苛要求，那Spark不同程序的性能会碰到不同的资源瓶颈，比如：CPU，带宽、内存.