问题标签 [spark-streaming-kafka]

我现在在 Python 中使用 kafka。想知道是否需要 Spark Kafka，或者我们可以通过 pyKafka 使用 kafka。

我担心 Spark 在此过程中会产生开销（pyspark），如果我们不使用任何 spark 函数，则只需要 Kafka 流。

使用 Pyspark 和 kafka spark 有哪些不便之处？

我是火花流的新手，并试图实现 kafka、MongoDB 集成，其中我的代码从 Kafka 主题中提取 JSON 数据并插入到 MongoDB 中。下面是我的代码

我正在循环到每个 RDD 并使用辅助方法 saveToMongoDB() 将其保存到 MongoDB。我收到以下错误。任何输入将不胜感激。

我正在尝试构建 Kafka 并触发流式传输用例。其中，Spark Streaming 正在使用来自 Kafka 的流。我们正在增强流并将增强的流存储到一些目标系统中。

我的问题是在纱线集群或纱线客户端模式下运行火花流作业是否有意义？（这里不涉及Hadoop）

我认为 Spark 流式作业应该只运行本地模式，但另一个问题是如何提高 Spark 流式作业的性能。

谢谢，

我一直在做通过卡夫卡消费和生产数据的火花流工作。我用的是directDstream，所以我必须自己管理offset，我们用redis来读写offset。现在有个问题，当我启动我的客户端时，我的客户端需要从redis中获取offset，而不是kafka中存在的offset本身。如何显示我编写我的代码？现在我已经在下面编写了我的代码：

但我认为 fromOffsets 是火花流客户端启动时的值（来自 redis），而不是在其运行期间。谢谢您的帮助。

我们需要使用commitAsyncKafka消费者的API。它适用于 Kafka 0.9 和 0.10 版本。

可以将 Spark 流式传输 1.6 版本与 Kafka 0.10 一起使用吗？

根据我读过的博客和网站，Kafka 0.10 仅由 spark 流 (Spark 2.1.X) 版本使用/支持

Kafka 本身指向以下链接： https ://blog.cloudera.com/blog/2017/06/offset-management-for-apache-kafka-with-apache-spark-streaming/

我正在尝试在一个简单的应用程序中将 Spark Streaming 连接到 Kafka。我通过 Spark 文档中的示例创建了这个应用程序。当我尝试运行它时，我得到了这样的异常：

这是我的代码：

我不确定问题出在配置还是代码本身。这就是我的 build.sbt 文件的样子：

我将不胜感激任何帮助，因为我无法弄清楚出了什么问题！

我正在尝试集成 kafka 消息代理和 spark 并面临一个问题 Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/spark/streaming/kafka010/LocationStrategies

下面是java spark代码

下面是我的 pom.xml 我试过很多 jar 文件版本都找不到合适的。

我正在运行我的火花工作，如下所示：

我觉得由于错误的jar文件使用，上述问题正在上升。有人可以帮我解决这个问题。我想知道这里应该使用哪些正确的 jar 文件。如果有人分享一些关于这些程序的宝贵资源，例如 Spark 和 Kafka 的集成，也将不胜感激。

自 2 天以来，我一直在尝试解决此问题，但无法解决

提前致谢。

我遇到了以下情况

有关可能的 kafkaParams，请参阅 Kafka 消费者配置文档。如果您的 Spark 批处理持续时间大于默认的 Kafka 心跳会话超时（30 秒），请适当增加 heartbeat.interval.ms 和 session.timeout.ms。对于大于 5 分钟的批次，这将需要更改代理上的 group.max.session.timeout.ms

在此链接上 https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

如果我在 spark conf 上设置了以下属性，这是否适用

另外，将 heartbeat.interval.ms 和 session.timeout.ms 设置为大于 kafka 流批处理持续时间的原因是什么？kafka 的心跳不会响应消费者投票请求吗？

我还在本地机器上运行 spark 流应用程序和 kafka。我的batch size是1分钟，我的kafka配置如下

但是，当批处理持续时间为 1 分钟及以上的心跳间隔和会话超时值运行时，我并没有真正看到任何问题。我在这里错过了什么吗？

我正在使用 Spark 结构化流从 Kafka 主题中读取记录；我打算计算 Spark 中每个“微批次”中收到的记录数readstream

这是一个片段：

我从文档中了解到 kafka_df 将在 astreamingQuery启动（接下来）时被懒惰地评估，并且在评估时，它拥有一个微批次。所以，我想先做一个主题，然后做一个应该工作groupBy。count

像这样：

现在要评估所有这些，我们需要一个 streaminQuery，比如说，一个控制台接收器查询以在控制台上打印它。这就是我看到问题的地方。DataFrames上的流式aggregate查询，例如kafka_df仅适用于outputMode complete/update而不适用于append。

这实际上意味着，streamingQuery 报告的计数是累积的。

像这样：

在受控设置中，其中：
实际发布记录：1500
实际接收微批次：3
实际接收记录：1500

每个微批次的计数应该是500，所以我希望（希望）查询打印到控制台：

主题：测试
计数：500
主题：测试
计数：500
主题：测试
计数：500

但事实并非如此。它实际上打印：

主题：测试
计数：500
主题：测试
计数：1000
主题：测试
计数：1500

我理解这是因为'outputMode'完成/更新（累积）

我的问题：是否有可能准确地得到每个微批次的计数是 Spark-Kafka 结构化流式传输？

从文档中，我发现了水印方法（支持追加）：

但是这样的结果console_query是不准确的，而且看起来很离谱。

TL;DR - 任何关于准确计算 Spark-Kafka 微批处理中记录的想法都将不胜感激。

我尝试使用pyspark将 spark 和 kafka 集成到 Jupyter notebook中。这是我的工作环境。

Spark 版本：Spark 2.2.1 Kafka 版本：Kafka_2.11-0.8.2.2 Spark 流式 kafka jar：spark-streaming-kafka-0-8-assembly_2.11-2.2.1.jar

我在 spark-defaults.conf 文件中添加了一个 Spark 流式 kafka 程序集 jar 文件。

当我启动 pyspark 流的 streamingContext 时，此错误显示为无法从 MANIFEST.MF 读取 kafka 版本。

这是我的代码。