特征提取与转换 - 基于RDD的API

from pyspark.mllib.feature import HashingTF, IDF
# 加载文档（每行一个）。
documents = sc.textFile("data/mllib/kmeans_data.txt").map(lambda line: line.split(" "))
hashingTF = HashingTF()
tf = hashingTF.transform(documents)
# 在应用 HashingTF 时，只需要对数据进行一次遍历，而应用 IDF 需要两次遍历：
# 第一次计算 IDF 向量，第二次通过 IDF 缩放词频。
tf.cache()
idf = IDF().fit(tf)
tfidf = idf.transform(tf)
# spark.mllib 的 IDF 实现提供了一个忽略术语的选项
# 该术语在少于最小文档数的情况下出现。
# 在这种情况下，这些术语的 IDF 设置为 0。
# 可以通过将 minDocFreq 值传递给 IDF 构造函数来使用此功能。
idfIgnore = IDF(minDocFreq=2).fit(tf)
tfidfIgnore = idfIgnore.transform(tf)

import org.apache.spark.mllib.feature.{HashingTF, IDF}
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.rdd.RDD
// 加载文档（每行一个）。
val documents: RDD[Seq[String]] = sc.textFile("data/mllib/kmeans_data.txt")
.map(_.split(" ").toSeq)
val hashingTF = new HashingTF()
val tf: RDD[Vector] = hashingTF.transform(documents)
// 在应用 HashingTF 时只需要对数据进行一次传递，而应用 IDF 需要两次传递：
// 首先计算 IDF 向量，其次通过 IDF 缩放词频。
tf.cache()
val idf = new IDF().fit(tf)
val tfidf: RDD[Vector] = idf.transform(tf)
// spark.mllib IDF 实现提供了一个选项，可以忽略出现在少于
// 最小文档数的术语。在这种情况下，这些术语的 IDF 设置为 0。
// 通过将 minDocFreq 值传递给 IDF 构造函数可以使用此功能。
val idfIgnore = new IDF(minDocFreq = 2).fit(tf)
val tfidfIgnore: RDD[Vector] = idfIgnore.transform(tf)

from pyspark.mllib.feature import Word2Vec
inp = sc.textFile("data/mllib/sample_lda_data.txt").map(lambda row: row.split(" "))
word2vec = Word2Vec()
model = word2vec.fit(inp)
synonyms = model.findSynonyms('1', 5)
for word, cosine_distance in synonyms:
print("{}: {}".format(word, cosine_distance))

import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}
val input = sc.textFile("data/mllib/sample_lda_data.txt").map(line => line.split(" ").toSeq)
val word2vec = new Word2Vec()
val model = word2vec.fit(input)
val synonyms = model.findSynonyms("1", 5)
for((synonym, cosineSimilarity) <- synonyms) {
println(s"$synonym $cosineSimilarity")
}
// 保存和加载模型
model.save(sc, "myModelPath")
val sameModel = Word2VecModel.load(sc, "myModelPath")

from pyspark.mllib.feature import StandardScaler
from pyspark.mllib.linalg import Vectors
from pyspark.mllib.util import MLUtils
data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
label = data.map(lambda x: x.label)
features = data.map(lambda x: x.features)
scaler1 = StandardScaler().fit(features)
scaler2 = StandardScaler(withMean=True, withStd=True).fit(features)
# data1 将具有单位方差。
data1 = label.zip(scaler1.transform(features))
# data2 将具有单位方差和零均值。
data2 = label.zip(scaler2.transform(features.map(lambda x: Vectors.dense(x.toArray()))))

import org.apache.spark.mllib.feature.{StandardScaler, StandardScalerModel}
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.util.MLUtils
val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
val scaler1 = new StandardScaler().fit(data.map(x => x.features))
val scaler2 = new StandardScaler(withMean = true, withStd = true).fit(data.map(x => x.features))
// scaler3 是与 scaler2 相同的模型，将产生相同的转换
val scaler3 = new StandardScalerModel(scaler2.std, scaler2.mean)
// data1 将具有单位方差。
val data1 = data.map(x => (x.label, scaler1.transform(x.features)))
// data2 将具有单位方差和零均值。
val data2 = data.map(x => (x.label, scaler2.transform(Vectors.dense(x.features.toArray))))

from pyspark.mllib.feature import Normalizer
from pyspark.mllib.util import MLUtils
data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
labels = data.map(lambda x: x.label)
features = data.map(lambda x: x.features)
normalizer1 = Normalizer()
normalizer2 = Normalizer(p=float("inf"))
# data1 中的每个样本将使用 $L^2$ 范数进行标准化。
data1 = labels.zip(normalizer1.transform(features))
# data2 中的每个样本将使用 $L^\infty$ 范数进行标准化。
data2 = labels.zip(normalizer2.transform(features))

import org.apache.spark.mllib.feature.Normalizer
import org.apache.spark.mllib.util.MLUtils
val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
val normalizer1 = new Normalizer()
val normalizer2 = new Normalizer(p = Double.PositiveInfinity)
// data1中的每个样本将使用 $L^2$ 范数进行标准化。
val data1 = data.map(x => (x.label, normalizer1.transform(x.features)))
// data2中的每个样本将使用 $L^\infty$ 范数进行标准化。
val data2 = data.map(x => (x.label, normalizer2.transform(x.features)))

import org.apache.spark.mllib.feature.ChiSqSelector
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.util.MLUtils
// 加载一些libsvm格式的数据
val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")
// 将数据离散化为16个相等的区间，因为ChiSqSelector需要分类特征
// 尽管特征是双精度数，ChiSqSelector将每个唯一值视为一个类别
val discretizedData = data.map { lp =>
LabeledPoint(lp.label, Vectors.dense(lp.features.toArray.map { x => (x / 16).floor }))
}
// 创建ChiSqSelector，该选择器将选择692个特征中的前50个
val selector = new ChiSqSelector(50)
// 创建ChiSqSelector模型（选择特征）
val transformer = selector.fit(discretizedData)
// 从每个特征向量中过滤前50个特征
val filteredData = discretizedData.map { lp =>
LabeledPoint(lp.label, transformer.transform(lp.features))
}

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.feature.ChiSqSelector;
import org.apache.spark.mllib.feature.ChiSqSelectorModel;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
JavaRDD<LabeledPoint> points = MLUtils.loadLibSVMFile(jsc.sc(),
"data/mllib/sample_libsvm_data.txt").toJavaRDD().cache();
// 将数据分为16个相等的区间，因为ChiSqSelector需要分类特征
// 尽管特征是双精度数，但ChiSqSelector将每个唯一值视为一个类别
JavaRDD<LabeledPoint> discretizedData = points.map(lp -> {
double[] discretizedFeatures = new double[lp.features().size()];
for (int i = 0; i < lp.features().size(); ++i) {
discretizedFeatures[i] = Math.floor(lp.features().apply(i) / 16);
}
return new LabeledPoint(lp.label(), Vectors.dense(discretizedFeatures));
});
// 创建ChiSqSelector，选择692个特征中的前50个
ChiSqSelector selector = new ChiSqSelector(50);
// 创建ChiSqSelector模型（选择特征）
ChiSqSelectorModel transformer = selector.fit(discretizedData.rdd());
// 从每个特征向量中过滤出前50个特征
JavaRDD<LabeledPoint> filteredData = discretizedData.map(lp ->
new LabeledPoint(lp.label(), transformer.transform(lp.features())));

from pyspark.mllib.feature import ElementwiseProduct
from pyspark.mllib.linalg import Vectors
data = sc.textFile("data/mllib/kmeans_data.txt")
parsedData = data.map(lambda x: [float(t) for t in x.split(" ")])
# 创建权重向量
transformingVector = Vectors.dense([0.0, 1.0, 2.0])
transformer = ElementwiseProduct(transformingVector)
# 批量转换
transformedData = transformer.transform(parsedData)
# 单行转换
transformedData2 = transformer.transform(parsedData.first())

import org.apache.spark.mllib.feature.ElementwiseProduct
import org.apache.spark.mllib.linalg.Vectors
// 创建一些向量数据；对稀疏向量也适用
val data = sc.parallelize(Seq(Vectors.dense(1.0, 2.0, 3.0), Vectors.dense(4.0, 5.0, 6.0)))
val transformingVector = Vectors.dense(0.0, 1.0, 2.0)
val transformer = new ElementwiseProduct(transformingVector)
// 批量转换和逐行转换给出相同的结果：
val transformedData = transformer.transform(data)
val transformedData2 = data.map(x => transformer.transform(x))

import java.util.Arrays;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.feature.ElementwiseProduct;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
// 创建一些向量数据；也适用于稀疏向量
JavaRDD<Vector> data = jsc.parallelize(Arrays.asList(
Vectors.dense(1.0, 2.0, 3.0), Vectors.dense(4.0, 5.0, 6.0)));
Vector transformingVector = Vectors.dense(0.0, 1.0, 2.0);
ElementwiseProduct transformer = new ElementwiseProduct(transformingVector);
// 批处理转换和逐行转换给出相同的结果：
JavaRDD<Vector> transformedData = transformer.transform(data);
JavaRDD<Vector> transformedData2 = data.map(transformer::transform);