apache-spark - 如何打印 Python TransformedDStream

Question

我正在尝试运行集成 AWS Kinesis Stream 和 Apache Spark 的字数统计示例。随机线会定期放入 Kinesis 中。

lines = KinesisUtils.createStream(...)

当我提交申请时，lines.pprint()我没有看到任何打印的值。

尝试打印lines对象，我看到 <pyspark.streaming.dstream.TransformedDStream object at 0x7fa235724950>

如何打印PythonTransformedDStream目的？并检查是否收到数据。

我确定不存在凭据问题，如果我使用虚假凭据，则会出现访问异常。

添加了引用代码

import sys

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kinesis import KinesisUtils, InitialPositionInStream

if __name__ == "__main__":
    sc = SparkContext(appName="SparkKinesisApp")
    ssc = StreamingContext(sc, 1)

    lines = KinesisUtils.createStream(ssc, "SparkKinesisApp", "myStream", "https://kinesis.us-east-1.amazonaws.com","us-east-1", InitialPositionInStream.LATEST, 2)

    # lines.saveAsTextFiles('/home/ubuntu/logs/out.txt')
    lines.pprint()

    counts = lines.flatMap(lambda line: line.split(" "))
                           .map(lambda word: (word, 1))
                           .reduceByKey(lambda a, b: a + b)

    counts.pprint()

    ssc.start()
    ssc.awaitTermination()

Answer 1

终于我成功了。

我引用的示例代码 https://github.com/apache/spark/blob/master/external/kinesis-asl/src/main/python/examples/streaming/kinesis_wordcount_asl.py提交申请的命令错误。

我让它工作的正确命令是

$ bin/spark-submit --jars external/spark-streaming-kinesis-asl_2.11-2.1.0.jar --packages org.apache.spark:spark-streaming-kinesis-asl_2.11:2.1.0 /home/ubuntu/my_pyspark/spark_kinesis.py