python - python : bugs in countByValue and countByValueAndWindow? 中的 Spark 流式传输

Question

我在 Ubuntu 环境中使用 Spark 1.5.2 和 Python 2.7。

根据有关 countByValue 和 countByValueAndWindow 的文档: Transformations on dstreams
Window operations

countByValue: 当在类型为 K 的元素的 DStream 上调用时，返回一个新的(K，Long)对的 DStream，其中每个键的值是它在源 DStream 的每个 RDD 中的频率.

countByValueAndWindow: 当在 (K, V) 对的 DStream 上调用时，返回一个新的 (K, Long) 对的 DStream，其中每个键的值是它在滑动窗口中的频率。与 reduceByKeyAndWindow 一样，reduce 任务的数量可通过可选参数进行配置。

所以基本上这两个函数的返回值应该是(K, Long) 对列表，对吧？

然而，当我在做一些实验时，返回值竟然是一个整数列表，而不是对!

另外，pySpark在Github上的官方测试代码中: Link1 Link2

您可以看到“预期结果”是整数列表!在我看来，它正在计算不同元素的数量并将它们组合在一起。

我以为我以某种方式误解了文档，直到我在 Github 上看到 scala 的测试代码:Link1 Link2

类似的测试用例，但此时的结果是成对的序列!

所以总而言之，scala 的文档和测试用例告诉我们结果是成对的。但是 python 测试用例和我自己的实验表明结果是整数。

我是 PySpark 和 Spark Streaming 的新手。有人可以帮我解释一下这种不一致吗？现在我正在使用 reduceByKey 和 reduceByKeyAndWindow 作为解决方法。

引用资料:

1. PySpark streaming documentation about countByValue

2. PySpark streaming documentation about countByVauleAndWindow

3. Dpark test cases of countByVauleAndWindow

4. An example using countByValue in PySpark (not streaming)

---

更新

此错误计划在 pyspark 2.0.0 中修复

Answer 1

我同意，countByValueAndWindow 有一个错误，它应该返回按值计数，而不仅仅是没有值的计数。即使您在 Python 中运行与 Scala 版本运行相同的测试用例 (link)，您也可以看到此函数的 pyspark 版本如何仅返回计数而不返回它们的值(例如，对)

>>> input = [['a'], ['b', 'b'], ['a', 'b']]
>>> from pyspark.streaming import StreamingContext
>>> ssc = StreamingContext(sc, 1)
>>> input = [sc.parallelize(d, 1) for d in input]
>>> input_stream = ssc.queueStream(input)
>>> input_stream2 = input_stream.countByValueAndWindow(2, 1)
>>> def f(rdd):
...     rows = rdd.collect()
...     for r in rows:
...         print r
... 
>>> input_stream2.foreachRDD(f)
>>> 
>>> sc.setCheckpointDir('/home/xxxx/checkpointdir')
>>> ssc.start() 
>>> 1
2
2
2
0

您应该将其作为 Jira 中的错误 (link) 提出，这应该很容易修复。我看不出任何人如何以当前形式使用此函数，因为如果没有键，返回的那些值数字将毫无意义。