火花流答案 - 爱码网

【问题标题】：Spark Streaming火花流
【发布时间】：2017-07-27 06:52:22
【问题描述】：

我在 Pyspark 中有 Spark Streaming，“批处理间隔”= 30 秒

ssc = StreamingContext(sc, 30)

然后我想使用 window() 函数来获取最后一小时的数据，并每隔 30 秒对这些数据进行切片。

kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})

counts = kvs.map(lambda (k, v): json.loads(v))\
            .map(TransformInData).window(108000)

我有一个错误

16/02/18 10:23:01 INFO JobScheduler: Added jobs for time 1455790980000 ms
16/02/18 10:23:30 INFO PythonTransformedDStream: Slicing from 1455683040000 ms to 1455791010000 ms (aligned to 1455683040000 ms and 1455791010000 ms)
16/02/18 10:23:30 INFO PythonTransformedDStream: Time 1455790650000 ms is invalid as zeroTime is 1455790650000 ms and slideDuration is 30000 ms and difference is 0 ms
16/02/18 10:23:31 INFO JobScheduler: Added jobs for time 1455791010000 ms

我读过这个https://groups.google.com/forum/#!topic/spark-users/GQoxJHAAtX4 但我不明白为什么它不起作用

【问题讨论】：

您使用的是哪个版本的 Spark？
1.6.0版本火花
同样的问题：我猜它正在抛出“时间无效..”，因为isTimeValid() 方法的time - ZeroTIme(difference) 是Dstream 类的0，这意味着time = zeroTime 和@987654329 @ 使用上述消息返回 true，导致 isTimeValid 方法到 return false。
我在打开 N dstream 之一并将它们与其他连接时遇到了同样的问题。（Spark 1.6.0、Scala 2.10.5、Hadoop 2.6）
你能确定你的时区是同步的吗？ abdussamad.com/archives/…

标签： apache-spark streaming pyspark

【解决方案1】：

我也遇到了同样的问题。升级到 Spark 2.0.1 修复了它。

【讨论】：

【解决方案2】：

是的，请升级到 Spark 2.1 然后在ms中添加kafka frequency = 1000，在你的Dstream()中添加offset

【讨论】：