Python 中 Apache Spark 的输出 Dstream

Question

我正在尝试将用于构建实时数据管道的技术，在将我的内容导出到文件时我遇到了一些问题运行。

我已经设置了一个本地 kafka 集群和一个 node.js 生产者，它发送一条简单的文本消息只是为了测试功能并粗略估计实施的复杂性。

这是从 kafka 读取的 spark 流作业，我正试图让它写入文件。

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

# Create a local StreamingContext with two working thread and batch interval of 1 second
sc = SparkContext("local[2]", "KafkaStreamingConsumer")
ssc = StreamingContext(sc, 10)

kafkaStream = KafkaUtils.createStream(ssc, "localhost:2181", "consumer-group", {"test": 1})

kafkaStream.saveAsTextFile('out.txt')

print 'Event recieved in window: ', kafkaStream.pprint()

ssc.start()
ssc.awaitTermination()

我在提交 spark 作业时看到的错误是：

kafkaStream.saveAsTextFile('out.txt')
AttributeError: 'TransformedDStream' object has no attribute 'saveAsTextFile'

没有对数据执行任何计算或转换，我只是想构建流程。 change/add 我需要什么才能导出文件中的数据？

Answer 1

http://spark.apache.org/docs/latest/api/python/pyspark.streaming.html

saveAsTextFiles（注意复数）

saveAsTextFile（单数）是 RDD 上的一种方法，而不是 DStream。