问题标签 [spark-kafka-integration]

• 火花版本：2.4.5
• 组件：Spark Streaming
• 类：DirectKafkaInputDStream

在课堂DirectKafkaInputDStream上，我有点困惑为什么要paraniodPoll在 seekToEnd 之前调用？

在 spark-batch 作业（在 EMR 上运行）中每天阅读来自 kafka 主题的最新消息的最佳选择是什么？我不想使用 spark-streaming ，因为没有 24/7 的集群。我看到了kafka-utils的选项： https ://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka_2.11 但是我看到最后一个版本是在2016年。它仍然是最好的选择吗？

谢谢！

- - - - - - - - - - - 编辑 - - - - - - -

感谢您的回复，我尝试了这个 JAR：

在 EMR 上运行它：scalaVersion = '2.12.11' sparkVersion = '2.4.4'

使用以下代码：

我想阅读每一批次，kafka 中所有可用的消息。该程序因以下错误而失败：

我做错了什么？谢谢。

我有一个结构流作业，它从 Kafka 主题读取消息，然后保存到 dbfs。代码如下：

我可以看到检查点工作正常。但是，没有数据输出。

显示空表。任何线索为什么？

正如标题所说，我需要在我的 Databricks 笔记本中使用来自 CloudKarafka（免费 Kafka 集群）中某个主题的消息。我有以下代码：

但是当我执行这段代码时：

我收到了这个错误：java.lang.IllegalArgumentException: No serviceName defined in either JAAS or Kafka config

你能帮助我吗？

提前致谢。

下面是我第一个使用 kafka 和 pyspark 的程序。代码似乎毫无例外地运行，但我的查询输出为空。

我正在启动火花和卡夫卡。后来，在 Kafka 启动中，我订阅了 topic = "quickstart-events" 并从终端为该主题生成了消息。但是当我运行这段代码时，它给了我空白的数据框。

我该如何解决？

代码：

输出：

我对startingOffsets结构化流媒体感到困惑。在这里的官方文档中，它说查询类型

1. 流式传输- 这是连续流式传输吗？
2. 批处理- 这是用于使用 forEachBatch 还是触发器进行查询？（最新的不允许）

我的工作流程也已checkpoints启用。这如何与 一起工作startingOffsets？如果我的工作流程崩溃并且我有startingOffsetsas latest，火花检查 kafka 偏移量或火花检查点偏移量或两者兼而有之？

我正在尝试为 pyspark readStream 找到兼容版本的 jar。我探索了许多版本，但找不到兼容的 jar。如果我做错了什么，请告诉我。

我的系统配置和使用的 jar

例外：我在 KafkaConfigUpdater 上遇到错误，不确定解决方案是什么。

示例代码

我是一个工作流数据的 scala + kafka。

我对 kafka 中的功能 commitAsync offsetRanges 有 1 个问题

我的代码：

当前，我运行应用程序，变量isCommit总是true 我想要从另一个函数分配的变量值，isCommit或者它从 rdd.foreachPartition 内部接收值。

====另外，我还有1个问题当我们从kafka接收100条消息时，在我们处理成功50条消息后，失败50条消息。我们如何为 50 条成功消息 commitAsync ？

谢谢大家！

我编写了一个 Spark 结构化流应用程序，它从 Kafka 主题中读取数据并将这些消息实时写入 Oracle。一切正常，除非作业被终止并且正在写入的当前消息被破坏（即它是用疯狂的值写入的）。

在这里你可以看到我的代码的快照。有人知道我该如何解决这样的问题吗？PS。这个问题也不会引发异常。

spark-streaming-kafka-0-8_2.11:2.3.0:(我的旧逻辑

如何实现相同的功能 spark-streaming-kafka-0-10_2.12：

问题：

1. 如何转换成foreachRDD spark-streaming-kafka-0-10_2.12？

2. 如何实现以下功能

   行 = rdd.map(lambda x: x[1]).collect()

我得到以下错误