1

我想用来自 kafka 主题的数据训练一个流式 kmeans 模型。

我的问题是如何呈现 kmeans streamig 模型的数据

sc = SparkContext(appName="PythonStreamingKafka")
ssc = StreamingContext(sc, 30)

zkQuorum, topic = sys.argv[1:]
kvs = KafkaUtils.createStream(ssc, zkQuorum, "spark-streaming-consumer", {topic: 1})
lines = kvs.map(lambda x: x[1])

lines.pprint()

此输出（这些是我的功能，用“|”分隔）：

1.0|2.0|0.0|21.0|2.0

1.0|2.0|0.0|21.0|2.0

然后我想这样做

model = StreamingKMeans(k=2, decayFactor=1.0).setRandomCenters(3, 1.0, 0)
model.trainOn(lines)

如果我结合两段代码，我会得到错误：

TypeError: Cannot convert type <type 'unicode'> into Vector
4

1 回答 1

1

第一个问题是格式化从 kafka 中提取的流。这是用于管道分隔数据的方法 

sc = SparkContext(appName="PythonStreamingKafka")
ssc = StreamingContext(sc, 30)

zkQuorum, topic = sys.argv[1:]
kvs = KafkaUtils.createStream(ssc, zkQuorum, "spark-streaming-consumer", {topic: 1})

raw = kvs.flatMap(lambda kafkaS: [kafkaS])
lines = raw.map(lambda xs: xs[1].split("|"))

lines = lines.map(lambda x: DenseVector(x))

第二个问题是数据的维度：第一个参数setRandomCenters（它应该与特征数量相同）

于 2017-07-21T09:29:46.077 回答