如何在Spark中为Kmeans映射MongoDB数据？

Question

sc = SparkContext(appName="KMeansExample")  # SparkContext
data = sc.textFile("/home/mhoeller/kmeans_data.txt")
parsedData = data.map(lambda line: array([int(x) for x in line.split(' ')]))
clusters = KMeans.train(parsedData, 2, maxIterations=10, initializationMode="random")

spark = SparkSession \
.builder \
.appName("myApp") \
.config("spark.mongodb.input.uri", "mongodb://127.0.0.1/ycsb.usertable") \
.config("spark.mongodb.output.uri", "mongodb:/127.0.0.1/ycsb.usertable") \
.getOrCreate()

df = spark.read.format("com.mongodb.spark.sql.DefaultSource").option("uri","mongodb://127.0.0.1/ycsb.usertable").load()

# <<<< Here I am missing the parsing >>>>>

clusters = KMeans.train(parsedData, 2, maxIterations=10, initializationMode="random")

我喜欢理解如何从df映射数据，以便它可以用作Kmeans的输入。

数据库的“布局”是D8:二进制(nullable=true)
--field9:二进制(nullable=true)

Answer 1

我喜欢理解如何从df映射数据，以便它可以用作Kmeans的输入。

根据您的代码片段，我假设您使用的是PySpark。

如果您查看clustering.kmeans Python API文档，可以看到第一个参数需要是向量或可转换序列类型的RDD

df = spark.read.format("com.mongodb.spark.sql.DefaultSource")
               .option("uri","mongodb://127.0.0.1/ycsb.usertable")
               .load()

有了上面的信息，让我们进入其中：

# Drop _id column and get RDD representation of the DataFrame
rowRDD = df.drop("_id").rdd

# Convert RDD of Row into RDD of numpy.array
parsedRdd = rowRDD.map(lambda row: array([int(x) for x in row]))

# Feed into KMeans 
clusters = KMeans.train(parsedRdd, 2, maxIterations=10, initializationMode="random")

如果希望保留布尔值（true/false）而不是整数(1/0)，那么可以删除int部分。如下：

parsedRdd = rowRDD.map(lambda row: array([x for x in row]))

把它们放在一起：

from numpy import array 
from pyspark.mllib.clustering import KMeans
import org.apache.spark.sql.SparkSession

spark = SparkSession \
.builder \
.appName("myApp") \
.config("spark.mongodb.input.uri", "mongodb://127.0.0.1/ycsb.usertable") \
.config("spark.mongodb.output.uri", "mongodb:/127.0.0.1/ycsb.usertable") \
.getOrCreate()

df = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource").load()

rowRDD = df.drop("_id").rdd
parsedRdd = rowRDD.map(lambda row: array([int(x) for x in row]))

clusters = KMeans.train(parsedRdd, 2, maxIterations=10, initializationMode="random")
clusters.clusterCenters