问题标签 [spark-streaming-kafka]

我正在尝试从 spark-streaming-kafka-0.8 移动到 spark-streaming-kafka-0.10 并且遇到以下错误： KafkaConsumer is not safe for multi-threaded access

我们在不同的 DC 中有多个 kafka 集群，我想在 Scala Spark Streaming 应用程序中同时使用它们。在 0.8 版中，它可以正常工作——我们刚刚多次调用 createDirectStream，每个集群调用一次。但升级到 0.10 后它停止工作。

我找到的唯一相关答案是：KafkaConsumer 对于来自 SparkStreaming 的多线程访问不安全 ，但它与使用来自同一集群的多个主题有关。目前不可能在一次调用 createDirectStream 中指定多个集群，同时多次调用它会导致错误。

我的问题是：有什么方法可以使用来自多个集群的 spark-streaming-kafka-0.10 数据？

为什么以及何时选择将 Spark 流与 Kafka 一起使用？

假设我有一个系统通过 Kafka 每秒接收数千条消息。我需要对这些消息应用一些实时分析并将结果存储在数据库中。

我有两个选择：

1. 创建我自己的从 Kafka 读取消息的工作程序，运行分析算法并将结果存储在数据库中。在 Docker 时代，只需使用 scale 命令就可以轻松地在我的整个集群中扩展这个 worker。我只需要确保我的分区数量与我的工作人员相同或更多，并且一切都很好，并且我有真正的并发性。

2. 使用 Kafka 流输入创建 Spark 集群。让 Spark 集群进行分析计算，然后存储结果。

是否存在第二种选择更好的情况？在我看来，这只是额外的开销。

我的 kafka 集群版本是 0.10.0.0，我想使用 pyspark 流来读取 kafka 数据。但在 Spark Streaming + Kafka 集成指南http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html 中没有 python 代码示例。那么pyspark可以使用spark-streaming-kafka-0-10来集成kafka吗？

预先感谢您的帮助 ！

我正在将火花流与 Kafka 集成一起使用，当我在本地模式下从我的 IDE 运行流应用程序时，一切都像一个魅力。但是，一旦我将其提交到集群，我就会不断出现以下错误：

java.lang.ClassNotFoundException：org.apache.kafka.common.serialization.StringDeserializer

我正在使用 sbt 程序集来构建我的项目。

我的 sbt 是这样的：

我也尝试使用 --package 选项

尝试 1

尝试 2

一切都没有成功。有没有人有什么建议

我正在使用以下 pom.xml 来构建我的项目。当我使用 spark submit 运行生成的 jar 时，会出现以下错误。

java.lang.NoClassDefFoundError: org/apache/spark/streaming/kafka/KafkaUtils

但是，如果我包含 --packages "org.apache.spark:spark-streaming-kafka-0-8-assembly_2.11:2.0.2" 选项，它会照常运行。当我的依赖项作为 pom.xml 的一部分包含时，我不明白为什么我必须提供此选项。

http://maven.apache.org/maven-v4_0_0.xsd"> 4.0.0 test.project ede-products-uis-spark-streaming jar 1.0-SNAPSHOT ede-products-uis-spark-streaming

我是 spark 和 kafka 的新手，我对使用 kafka 的 spark 流的使用模式略有不同。我在用

连续事件数据正在流式传输到我需要从多个火花流应用程序处理的 kafka 主题。但是当我运行 spark 流应用程序时，只有其中一个接收到数据。

我有两个火花流应用程序，通常我提交的第一个使用 kafka 消息。第二个应用程序只是等待消息并且永远不会继续。正如我所读到的，可以从多个消费者订阅 kafka 主题，火花流不是这样吗？或者我在 kafka 主题及其配置方面缺少什么？

提前致谢 。

我想使用 Kafka 集成来进行 Spark 流式传输。我使用 Spark 2.0.0 版。

但我得到一个未解决的依赖错误（“未解决的依赖：org.apache.spark#spark-sql-kafka-0-10_2.11;2.0.0: not found”）。

我怎样才能访问这个包？还是我做错了什么/错过了什么？

我的 build.sbt 文件：

谢谢你的帮助。

我正在使用版本 2.0.2 的 spark-streaming-kafka-0-10_2.10 进行火花流式传输作业。我收到这样的警告： 17/10/10 16:42:25 WARN KafkaUtils: overriding enable.auto.commit to false for executor 17/10/10 16:42:25 WARN KafkaUtils: overriding auto.offset.reset to none for executor 17/10/10 16:42:25 WARN KafkaUtils: overriding executor group.id to spark-executor-dump_user_profile 17/10/10 16:42:25 WARN KafkaUtils: overriding receive.buffer.bytes to 65536 see KAFKA-3135

并且当我查看源代码时，有一段代码修复了 KafkaUtils 中名为 fixKafkaParams(...) 的参数，如下所示：

```

} ``` 我怎样才能度过这个难关？非常感谢

在我们的集群中，我们有 Kafka 0.10.1 和 Spark 2.1.0。火花流应用程序与检查点机制（HDFS 上的检查点）一起工作得很好。但是，我们注意到，如果代码发生更改，使用检查点流应用程序不会重新启动。

探索 Spark Streaming 文档 - 在 Kafka 上存储偏移量：

http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html#kafka-itself，上面写着：

在此之后，我修改了我们的代码，如下所示：

现在，当我尝试启动 Streaming 应用程序时，它没有启动并查看日志，这是我们看到的：

如果我们遗漏了什么，有人可以建议吗？

我正在使用 Spark 流应用程序。应用程序使用直接流从 Kafka 主题（具有 200 个分区）中读取消息。有时应用程序会抛出 ConcurrentModificationException->

我的火花集群有两个节点。Spark 版本是 2.1。该应用程序运行两个执行程序。从我从异常和 kafka 消费者代码中可以看出，似乎两个线程正在使用同一个 kakfa 消费者。我不知道两个线程是如何访问同一个接收器的。理想情况下，每个执行程序都应该有一个由单个线程提供的专有 kafka 接收器服务，该线程必须读取所有分配的分区的消息。从 kafka-> 读取的代码片段