问题标签 [spark-streaming]

尝试编写使用来自 Kafka 的消息的 Spark Streaming 作业。这是我到目前为止所做的：

1. 启动 Zookeeper
2. 启动 Kafka 服务器

3. 向服务器发送了一些消息。当我运行以下命令时，我可以看到它们：

   /li>

4. 现在尝试编写一个程序来计算 5 分钟内收到的消息数。

代码看起来像这样：

不确定第三个参数（消费者组）使用什么值。当我运行它时，我得到Unable to connect to zookeeper server. 但是 Zookeeper 在端口上运行2181；否则，第 3 步将无法正常工作。

好像我没有KafkaUtils.createStream正确使用。有任何想法吗？

Kafka 和 Spark-Streaming 之间存在问题，我在生产中有一个低级流量（大约 12000 - 15000 条记录/每秒）服务，起初，消耗流量似乎正常，但在 10 - 15 分钟后，突然消耗的速度几乎剩下1/10。可能是网络流量问题？

Kafka的配置：
num.network.threads=2
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.flush.interval .messages=10000
log.flush.interval.ms=1000
log.retention.hours=12
log.segment.bytes=536870912
log.retention.check.interval.ms=60000
log.cleaner.enable=false
log.cleanup.interval .mins=1

火花流的配置（消费者）：

这是我的源代码，我从服务器端获取一些数据，它不断生成数据流。然后对于每个 RDD ，我正在应用 SQL 模式，一旦创建了这个表，我就试图从这个 DStream 中选择一些东西。

但是它抛出了这个可序列化的异常，尽管我使用的类确实实现了序列化。

The application throws a java.lang.NoSuchMethodException

Stacktrace

Seems issue has already been fixed in spark 1.1.0 as per this Link

Spark : 1.1.0
Kafka : 0.8.1.1

如何有效地将 a 拆分RDD[T]为带有元素的Seq[RDD[T]]/并保留原始顺序？Iterable[RDD[T]]n

我希望能够写出这样的东西

这应该会导致类似

spark提供这样的功能吗？如果不是，那么实现这一目标的有效方法是什么？

看起来不是很快。。

当我检查工作日志并发现以下错误时，Spark Streaming Application 每隔 2 小时就会被杀死一次，没有任何日志：

我还将 spark.cleaner.ttl 值设置为 10 分钟，但没有运气。

我有一个带有用户请求 URL 的日志流。 每分钟我都想获得一直请求的前 100 个页面并将其保存到 HDFS。

我了解如何为每个 url 维护一些请求：

但是接下来我该怎么办？显然，我需要将所有更新传递给主机以维护一个优先队列，然后每 1 分钟获取它的前 K 个。

我怎样才能做到这一点？

UPD：我已经看到了 spark 示例和 algebird 的 MapMonoid 在那里使用。但是由于我不明白它是如何工作的（奇怪的是没有在网上找到任何信息），我不想使用它。我一定有办法，对吧？

我正在尝试从我尝试的 hdfs 文件夹（sparkstreaming）流式传输：

为了检查它是否是我做的好文件夹

返回 JavaDStream 不包含任何数据，我没有收到任何错误。该文件夹已包含一些文件..

有没有其他方法可以检查文件夹路径？还有其他需要检查的吗？

注意：我还尝试从本地文件中读取，然后将其放入流中，但我仍然有一个空输入（这是本次试用的代码）。我检查了我电脑上文件的位置spark-shell .. input.print() 行给出了 java.io.Exception ...

此外，第二个版本的日志显示输入存在问题：

谢谢！

我想知道在 apache spark 流中调用 reduceByKey 时是否保证了流中记录的顺序。基本上我所做的计算的一部分必须得到最后一个值。

这是一个例子：

我想将 DStream 中的每个 RDD 与非流式、不变的参考文件一起加入。这是我的代码：

我收到很多很多错误，最常见的是：

我认为我应该广播 geoData 而不是在每个任务中读取它（它是一个 100MB 的文件），但我不确定将第一次初始化 geoData 的代码放在哪里。

另外我不确定 geoData 是否定义正确（也许它应该使用 ssc 而不是 sc？）。我看到的文档只列出了转换和连接，但没有显示静态文件是如何创建的。

关于如何广播 geoData 然后将其加入每个流式 RDD 的任何想法？