Spark stream为1.6.2的Kafka 0.10.0无法接收任何消息
Spark Stream是Apache Spark的一个组件，用于实时处理和分析数据流。它提供了高级别的API，使开发人员能够轻松地处理实时数据，并将其集成到Spark的批处理和机器学习任务中。
针对您提到的问题，Spark Stream无法接收任何消息的原因可能有以下几个方面：
- 版本兼容性问题：Spark Stream 1.6.2与Kafka 0.10.0的版本可能存在不兼容的情况。建议您检查Spark Stream和Kafka的版本兼容性，并确保它们能够正常协同工作。您可以参考Spark官方文档或Kafka官方文档来获取相关版本兼容性信息。
- 配置问题：Spark Stream与Kafka之间的连接需要正确的配置参数。您需要确保在Spark Stream的配置中正确设置了Kafka的相关参数，如Kafka的地址、主题名称等。您可以参考Spark官方文档中关于Kafka集成的配置指南来进行配置。
- 网络或权限问题：如果Spark Stream无法接收任何消息，可能是由于网络连接问题或者Kafka的权限设置导致的。您可以检查网络连接是否正常，并确保Spark Stream能够访问到Kafka集群。此外，还需要确保Spark Stream具有足够的权限来读取Kafka中的消息。
针对上述问题，腾讯云提供了一系列与云计算和大数据相关的产品，可以帮助您解决实时数据处理的需求。例如，腾讯云的云数据仓库CDW产品可以提供实时数据处理和分析的能力，您可以通过CDW来处理和分析Kafka中的数据流。您可以访问腾讯云官方网站，了解更多关于CDW产品的详细信息和使用案例。
请注意，以上答案仅供参考，具体解决方案可能因实际情况而异。建议您根据具体问题进行调试和排查，并参考相关文档和资源来解决您的问题。
相关·内容
2回答
我希望使用Spark (1.6.2)流从Kafka (broker v 0.10.2.1)中的主题接收消息。
我正在使用Receiver方法。代码如下：
public static void main(String[] args) throws Exception
{
SparkConf sparkConf = new SparkConf().setAppName("SimpleStreamingApp");
JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkCon
浏览 4提问于2018-01-16得票数 10
最近我们转到了HDP 2.5，它有Kafka 0.10.0和Spark 1.6.2。所以我修改了我的pom和一些APIs来使用新的Kafka。我可以运行代码，但我看不到任何消息。我在下面添加了一个代码片段。我也发布了我的pom。我不确定这里出了什么问题。有人能帮帮忙吗。
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName(
"SparkApp");
JavaStreamingContext jssc = new JavaStreamingContext(con
浏览 5提问于2017-02-01得票数 0
回答已采纳
我正在使用Apache Flink，并尝试使用Apache Kafka协议连接到Azure eventhub以接收来自它的消息。我设法连接到Azure eventhub并接收消息，但我不能使用flink功能"setStartFromTimestamp(...)“如下所述()。当我试图从timestamp获取一些消息时，Kafka说broker端的消息格式是0.10.0之前的版本。有没有人遇到过这种情况？Apache Kafka客户端版本为2.0.1 Apache Flink版本为1.7.2
更新:尝试在消费包中使用Azure-Event-Hub quickstart examples
浏览 8提问于2019-03-21得票数 0
如果我使用Spark 1.6.2，Scala2.10和spark-streaming-kafka_2.10，我能从Scala代码中消费来自Kafka 0.10.1.0的消息吗？(有一台远程机器，其Kafka为0.10.1.0，代理间协议版本为0.10.1.0，Scala为2.11)。或者这意味着我应该使用Spark 2。*
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
浏览 5提问于2017-01-02得票数 1
回答已采纳
3回答
我正在使用kafka-node ConsumerGroup来阅读卡夫卡的信息。我想从主题中读取新消息，即不是所有以前未使用的消息，而是在创建ConsumerGroup之后到达的消息。
只有当给定的fromOffset: 'latest'没有存储偏移量时，选项groupId才会出现。我可以通过在每次创建groupId时创建一个新的ConsumerGroup来使用这个方法，但是我希望避免这种方法，因为这会在ConsumerGroup服务器上创建大量的groupIds和存储偏移量。
我还尝试手动将偏移量设置为-1，如下所示：
offset = new kafka.Offset(con
浏览 2提问于2017-08-25得票数 3
回答已采纳
1回答
我有一个Spark streaming (Scala)应用程序运行在CDH 5.13中，使用client 0.10.0使用Kafka的消息。我的Kafka集群包含3个代理。Kafka topic被划分为12个分区，均匀分布在这3个代理之间。我的Spark流媒体客户有12个执行器，每个执行器都有一个核心。Spark streaming开始从Kafka读取每批数百万条消息，但由于Spark无法应对负载和创建未处理批量的队列，因此将消息数减少到数千。这很好，我的期望是Spark处理小批处理非常快，并恢复正常，但是我看到有时只处理数百条消息的一个执行器在从Kafka读取最后一个偏移量后会出现“请求超时
浏览 0提问于2018-10-09得票数 1
我正在尝试使用IntelliJ在Scala中构建一个Kafka消费者，以便从Kafka主题读取消息并将它们保存在hdfs上。我使用的是spark 1.6.2，kafka_2.10-0.10，scala 2.10.5和hdp 2.5.3。我得到以下错误：
Exception in thread "main" java.lang.NoSuchMethodError: kafka.consumer.SimpleConsumer.<init>(Ljava/lang/String;IIILjava/lang/String;Lorg/apache/kafka/common/pr
浏览 0提问于2018-04-06得票数 0
我试着把数据从卡夫卡流出来
我用的是火花1.6.2和kafka 0.9.0.1和scala 2.11.8
当我使用基于接收器的方法(KafkaUtils.createStream())时，一切都很好，但是当我尝试这种没有接收器的直接方法时，一切都很好。
val kafkaStreams = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc,
Map("group.id" -> "blah",
"auto.offset.re
浏览 2提问于2016-07-09得票数 1
回答已采纳
我正在使用CDH (hadoop的cloudera分发版)中的kafka 1.0.1-kafka-3.1.0-SNAPSHOT
在我的批处理-1边缘服务器上，可以通过以下方式生成消息：
kafka-console-producer --broker-list batch-1:9092 --topic MyTopic
我可以使用消息，这要感谢在我的第一个节点上的动物园管理员：
kafka-console-consumer --zookeeper data1:2181 --topic MyTopic --from-beginning
但是我得到了nothing和引导服务器选项：
kafka-con
浏览 0提问于2018-10-30得票数 1
运行时在java代码下面运行时会出现错误，
对于像log4js之类的日志记录，是否需要包含任何依赖项？
为什么这个错误在编译时没有出现，这样就更容易了。
这是我的Java代码，
SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount11").setMaster("local[*]");
sparkConf.set("spark.streaming.concurrentJobs", "3");
// Create th
浏览 4提问于2016-11-23得票数 0
回答已采纳
我们在Google云平台上使用Bitnami Kafka 0.8.2 + spark 1.5.2。我们的spark streaming作业(消费者)没有接收到发送到特定主题的所有消息。它会收到50条消息中的1条(已在作业流中添加日志并已标识)。我们在kafka日志中没有看到任何错误。无法从kafka层进一步调试。控制台使用者显示在控制台中接收到输入主题。它没有达到spark-kafka集成流。任何关于如何调试这个问题的想法。另一个主题是在相同的设置下工作得很好。再次尝试使用spark 1.3.0，kafka 0.8.1.1也有同样的问题。所有这些工作在我们的本地实验室服务器上都运行良好
浏览 1提问于2015-12-24得票数 0
1回答
我正在尝试将数据写入一个卡夫卡主题后，阅读了一个蜂巢表如下。
write_kafka_data.py:
read_df = spark.sql("select * from db.table where some_column in ('ASIA', 'Europe')")
final_df = read_df.select(F.to_json(F.struct(F.col("*"))).alias("value"))
final_df.write.format("kafka")\
浏览 0提问于2021-09-13得票数 4
回答已采纳
我正在尝试来自的示例代码。代码可以在没有任何错误的情况下运行，但我无法接收任何记录。如果我从一开始就运行kafka-控制台- can . If，我就能得到记录。有人知道原因吗？我的代码如下：
val broker = "221.181.73.44:19092"
val topics = Array("connect-test")
val groupid = "SparkStreamingLoad3"
val kafkaParams = Map[String, Object](
"bootstrap.servers" ->
浏览 8提问于2017-06-22得票数 1
我想使用Apache Spark Structured Streaming和Kafka，Spark Structured Streaming支持Kafka 0.10及更高版本，我的Kafka集群使用kafka版本0.8.2.1。我想将当前kafka 0.8.2.1集群中的一些主题复制到基于2.2.0的新Kafka集群中。 为此，我尝试在Kafka 2.2.0集群上使用kafka-console-consumer来监听来自kafka集群0.8.2.1的消息，并将kafka-console-consumer的结果通过管道传输到kafka 2.2.0集群上的kafka-console-produc
浏览 17提问于2019-05-23得票数 1
1回答
我试图让多个用户使用相同的groupId来处理卡夫卡主题的多个分区，这将帮助我扩展消息的消耗。
根据Kafka的文件，它说：
如果所有使用者实例都有相同的使用者组，那么记录将有效地在使用者实例上实现负载平衡。
将消费者作为同一个消费者组的一部分意味着提供来自主题分区的消息分布在组成员之间的“竞争消费者”模式。每个使用者从一个或多个分区(“自动”分配给它)接收消息，而其他使用者(分配给不同分区)不会接收相同的消息。通过这种方式，我们可以将消费者的数量扩展到分区的数量(让一个消费者只读取一个分区)；
但是，当我使用相同的groupId部署多个spark应用程序时，它给了我以下例外：
java.l
浏览 1提问于2019-10-14得票数 2
2回答
创建props对象后设置使用者的代码
val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(util.Arrays.asList(topic))
代码导入如下所示
package main.scala
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.strea
浏览 2提问于2017-08-11得票数 0
1回答
当我尝试使用Kerberized集群中的Spark流应用程序使用Kafka的消息时，我遇到了一些问题。我尝试了两种方法-- ：
基于接收机的方法：KafkaUtils.createStream
直接进场(无接收器)：KafkaUtils.createDirectStream
基于接收器的方法(KafkaUtils.createStream)引发两种类型的异常(不同的异常，无论我处于本地模式(--master local[*])还是纱线模式(--master yarn --deploy-mode client) )：
星火本地应用程序中奇怪的kafka.common.Brok
浏览 9提问于2017-12-26得票数 4
回答已采纳
我们有一个在Spark2.3.3上运行的星火流应用程序
基本上，它打开了一个卡夫卡流：
kafka_stream = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "mykafka:9092") \
.option("subscribe", "mytopic") \
.load()
kafka主题有两个分区。之后，在列上有一些基本的过滤操作、一些Python和basic ()，例如：
浏览 1提问于2019-04-02得票数 10
回答已采纳
使用Spark 2.4.0
合流模式-注册表接收模式
消息键是在字符串中序列化的，在Avro中是值，因此我尝试使用io.confluent.kafka.serializers.KafkaAvroDeserializer，反序列化这个值，但是它不起作用。有人能检查一下我的代码，看看哪里出了问题吗？
进口图书馆：
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient
import io.confluent.kafka.serializers.KafkaAvroDeserializer
import or
浏览 4提问于2020-03-20得票数 1
回答已采纳
1回答
我有两个用Java编写的微服务，使用Spring。，通过，在它们之间发送消息。我需要发送一个自定义头，但到目前为止没有成功。我在互联网和Spring 上已经阅读和尝试了大多数东西。
..。不过，我还是无法使它发挥作用。
这意味着我从未在接收方中接收到消息，因为无法找到标头，也不能为空。我怀疑标题从来没有写在消息中。现在我正试图用Kafkacat来验证这一点。
任何帮助都将是wellcome
提前谢谢。
--信息
这里是发件人代码：
@SendTo("notifications")
public void send(NotificationPayload payload, Str
浏览 5提问于2020-10-29得票数 0
1回答
火花版本： 1.6.2
火花配置：
遗嘱执行人: 18人
每个执行器的内存: 30g
核心x执行器:2
卡夫卡配置：
分区: 18
上下文：
我有一个火花流管道，是使用新的直接方法从卡夫卡，而不是接收器。这个管道从kafka接收数据，转换数据，然后插入到Hive。我们发布用于插入数据的操作的触发器具有相当于RACK_LOCAL的本地级别，而不是PROCESS_LOCAL或NODE_LOCAL。此外，输入大小列显示网络信息而不是内存。我已将配置变量spark.locality.wait=30s设置为测试spark是否足以选择最佳本地模式，但没有显示任何改进。
考虑
浏览 1提问于2017-06-30得票数 2
回答已采纳
向Kafka topic写入json，并从kafka Topic读取json。实际上，我是逐行订阅主题和编写控制台的。但我必须接收/写入文件csv。但是我不能。我只写了一次csv，但没有附加。 你可以在下面看到我的代码。 谢谢! import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as func
spark = SparkSession.builder\
.config('
浏览 23提问于2020-01-11得票数 1
回答已采纳
我基本上是想使用Kafka中的数据，并将其写入HDFS。但是发生的情况是，它没有在hdfs中写入任何文件。它会创建空文件。
也请指导我，如果我想在hdfs中写avro格式，我可以如何修改代码。
为了简单起见，我写入本地C驱动器。
import org.apache.spark.SparkConf
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds
import o
浏览 4提问于2017-10-08得票数 0
我使用的是Java。
我正在接收Kafka消息上的文件路径。我需要将这个文件加载到spark RDD中，对其进行处理，然后将其转储到HDFS。
我能够从Kafka消息中检索到文件路径。我希望在此文件上创建一个dataset / RDD。
我不能在Kafka消息数据集上运行map函数。由于sparkContext在worker上不可用，因此出现NPE错误。
我不能在Kafka消息数据集上运行foreach。它会出错，并显示以下消息：
Queries with streaming sources must be executed with writeStream.start();"
我不
浏览 0提问于2017-09-26得票数 4
我已经在Scala中设置了Spark-Kafka Consumer，它接收来自多个主题的消息：
val properties = readProperties()
val streamConf = new SparkConf().setMaster("local[*]").setAppName("Kafka-Stream")
val ssc = new StreamingContext(streamConf, Seconds(10))
val kafkaParams = Map("metadata.broker.list" -> pro
浏览 0提问于2017-03-17得票数 0
我正在尝试读取星火流应用程序中的kafka流数据；而在读取数据的过程中，我得到了以下例外：
16/12/24 11:09:05信息storage.BlockManagerMaster:注册BlockManager
Exception in thread "main" java.lang.NoClassDefFoundError: kafka/serializer/StringDecoder
at com.inndata.RSVPSDataStreaming.KafkaToSparkStreaming.main(KafkaToSparkStreaming.java:69)
浏览 0提问于2016-12-24得票数 0
我正在使用全新的(标记为"alpha")的Spark2.0.2结构化流来读取kafka主题中的消息，并从其中更新两个cassandra表：
val readStream = sparkSession.readStream
.format("kafka")
.option("subscribe", "maxwell")
.option("kafka.bootstrap.servers", "localhost:9092")
.load
.as[KafkaMessage]
.ma
浏览 1提问于2016-12-23得票数 1
回答已采纳
我有一个每秒产生60,000条消息的应用程序。我向Kafka发送消息，我希望在其他应用程序中通过spark streaming接收这些消息，但Spark接收的消息速率约为40,000。我想增加Spark每个间隔接收的消息数，我该怎么做呢？
浏览 0提问于2017-03-11得票数 1
我已经使用Apache和Apache结构化流构建了一个应用程序。我面临以下问题。
场景：
我建立了一个星火结构流与一个来源的卡夫卡主题和接收器作为卡夫卡主题。
我们运行流并产生许多关于卡夫卡主题的信息。
通过清除流的检查点位置，我们停止了流并重新启动流。跑了5到6个小时后，流正在随机消耗旧卡夫卡的信息。
在清除检查点位置之后，我期待的只是流上的新消息。
星火版本: 2.4.0，Kafka -客户端版本: 2.0.0，Kafka版本: 2.0.0，集群管理器: Kubernetes。
我通过更改检查点位置尝试了这种情况，但问题仍然存在。
{
SparkConf sparkCo
浏览 0提问于2019-01-17得票数 1
回答已采纳
当我试图使用Intellij点击Spark流应用程序时
Env
火花核心版本2.2.0 Intellij IDEA 2017.3.5版
附加信息:星火是运行在纱模式。
获取错误:
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Exception in thread "main" java.lang.ExceptionInInitializerError
at kafka_stream.kafka_stream.main(kafka_stream.sc
浏览 0提问于2018-04-04得票数 0
我正在运行一个程序，它使用Apache从Apache集群获取数据，并将数据放在Hadoop文件中。我的计划如下：
public final class SparkKafkaConsumer {
public static void main(String[] args) {
SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount");
JavaStreamingContext jssc = new JavaStreamingContext(sparkConf
浏览 3提问于2016-07-21得票数 1
回答已采纳
Clickhouse版本21.12.3.32。我m following this PR([https://github.com/ClickHouse/ClickHouse/pull/21850](https://github.com/ClickHouse/ClickHouse/pull/21850)) to handle incorrect messages from kafka topic, but after some investigation I发现，如果单个消息包含损坏的数据，则无法解析整批接收到的消息，并可能导致数据丢失。
卡夫卡发动机表：
CREATE TABLE default
浏览 43提问于2022-03-16得票数 0
回答已采纳
2回答
为了捕捉Server中的数据更改，我尝试执行从Server到Pyspark的管道，我已经准备好了一切：
在Server中启用CDC
从SQL Server到Kafka进行生产，并使用Pyspark结构化流中的Kafka主题。
问题是:当我试图用控制台使用者检查数据更改是否经过Kafka时，它会向我显示JSON格式的消息，分为两个记录: Schema和Payload，以及Payload内部，在此之前和之后，分别给出更改之前的数据和更改后的数据。
我只在有效负载中被处理->在这部分JSON消息之后
因为当我像这样流它时，在木星命令行中，在我需要的字段上显示n
浏览 0提问于2021-06-30得票数 0
回答已采纳
1回答
我正在做一个使用Spring Boot，Kafka和Spark的小项目。到目前为止，我已经能够在一个项目中创建一个Kafka生产者，并作为消费者创建一个Spark-Kafka直接流。
我能够看到消息通过，一切似乎都在按预期进行。但是，我在运行使用者的项目上有一个rest端点。每当我禁用Direct Stream时，端点都工作得很好。然而，当我让流运行时，Postman说没有响应。我在服务器日志中也看不到任何指示曾经收到过请求的信息。
Spark使用者在项目启动时由一个bean启动。这是否会阻止localhost:8080上的正常服务器启动？
浏览 17提问于2018-12-20得票数 1
回答已采纳
我需要使用Scala和Spark使用来自远程Kafka队列主题的消息。默认情况下，远程计算机上的Kafka端口设置为7072，而不是9092。此外，在远程计算机上安装了以下版本：
Kafka 0.10.1.0
Scala 2.11
这意味着我应该将代理列表(带有端口7072)从Scala传递到远程Kafka，否则它将尝试使用默认端口。
问题是，根据日志，远程机器无法识别参数bootstrap.servers。我还尝试将此参数重命名为metadata.broker.list、broker.list和listeners，但在日志中始终出现相同的错误，然后默认使用端口9092 (显然
浏览 2提问于2017-01-02得票数 1
我可以使用星火流1.5.1与kafka 0.10.0？该网站spark.apache.org建议火花流应该与卡夫卡0.8.2；
我只想知道，如果我用卡夫卡0.10.0和火花1.5.1？
有人能帮忙吗。
浏览 3提问于2017-07-14得票数 1
回答已采纳
当kafka代理启动并运行时，在集群模式下运行的spark进程能够读取来自kafka主题的消息。但是当代理被故意关闭时，spark消费者仍然处于运行状态。
是否有任何参数来控制spark消费者和zookeeper进程之间的握手间隔，以便如果无法访问代理，spark进程可能会失败。或者有没有让消费者失望的替代方法。请提个建议。
浏览 0提问于2021-02-18得票数 0
2回答
SBT无法导入Kafka编解码类
、、、、
项目设置：
1生产者-序列化对象&向Kafka发送字节。
1火花使用者-应该在DefaultDecoder包中使用kafka.serializer来消耗字节。
发行：
SBT导入正确的库(kafka-客户端+ kafka_2.10)，但无法在kafka_2.10 jar中找到任何类。
它似乎是在错误的路径下搜索(org.apache.spark.streaming.kafka而不是org.apache.kafka)。
错误消息：
object serializer is not a member of package org.apache.spa
浏览 4提问于2016-04-04得票数 7
回答已采纳
我创建了一个有5个分区的Kafka主题。我正在使用createStream接收器API，如下所示。但不知何故，只有一个接收者正在获取输入数据。其余的接收者不进行任何处理。你能帮忙吗？
JavaPairDStream<String, String> messages = null;
if(sparkStreamCount > 0){
// We create an input DStream for each partition of the topic, unify those streams, and then re
浏览 1提问于2016-09-14得票数 1
2回答
我是新的火花和卡夫卡和我有一个稍微不同的使用模式火花流与卡夫卡。我在用
spark-core_2.10 - 2.1.1
spark-streaming_2.10 - 2.1.1
spark-streaming-kafka-0-10_2.10 - 2.0.0
kafka_2.10 - 0.10.1.1
连续事件数据正在被流到一个kafka主题中，我需要从多个火花流应用程序中处理这个主题。但是当我运行星火流应用程序时，只有其中一个接收数据。
Map<String, Object> kafkaParams = new HashMap<String, Object>(
浏览 3提问于2017-08-29得票数 4
回答已采纳
我已经使用KSQL为已经包含一些消息的Kafka主题创建了一个来自Kafka的流。但不会将来自该主题的任何消息接收到创建的流中。消息采用Avro格式，并在一定时间间隔后生成。
我想从头开始阅读，即从最早的邮件开始阅读。我还尝试将偏移量属性设置为最早，但没有收到任何消息。
create stream sample_transition with(topic_name='transition',value_format='avro');
浏览 1提问于2018-10-15得票数 1
2回答
我已经创建并部署了一个由4个正在运行的容器组成的星火集群。
火花主
火花工
火花-提交
数据挂载容器:从本地目录访问脚本
我在所有这些容器中添加了必需的依赖jar。
并且还在主机上部署了kafka，在那里它通过生产者产生流。
我按照以下文件中的具体步骤启动了kafka
我验证了kafka生产商和消费者在9092端口上交换的消息，该端口运行良好。
下面是我想要作为结构化流处理的简单pyspark脚本
from pyspark import SparkContext
from pyspark.sql import SparkSession
pri
浏览 1提问于2020-09-09得票数 0
回答已采纳
对于一个类项目，我需要一个Spark Java程序作为Kafka消费者来监听，并将Kafka主题的所有接收到的消息写入一个文件(例如"/user/zaydh/my_text_file.txt")。
我可以将消息作为JavaPairReceiverInputDStream对象接收；我还可以将其转换为JavaDStream<String> (来自Spark Kafka示例)。
但是，我找不到一种好的Java语法来将这些数据写到本质上是单个日志文件的地方。我尝试在JavaDStream对象上使用foreachRDD，但我找不到一种干净、并行安全的方法将其存储到单个日志文件
浏览 0提问于2015-10-22得票数 0
我们已经实现了一个应用程序来生成消息和标题，并将其发送到kafka主题。消费者端应用程序正在聆听卡夫卡的主题。消费者端应用程序能够接收有效载荷，但是头不是received.Earlier，我们的头是在有效载荷属性中接收的。为了禁用将报头(包括contentType)嵌入到有效负载中，我们使用了
spring.cloud.stream.bindings.output.producer.headerMode=raw
spring.cloud.stream.bindings.input.consumer.headerMode=raw
但是，我们无法在使用者端应用程序中接收报头。我们如何接收这些标头？
浏览 2提问于2019-05-08得票数 0
2回答
火花流卡夫卡消费者
、、、
我正在尝试建立一个星火流简单的应用程序，它将读取来自卡夫卡主题的信息。
经过大量的工作，我现在在这个阶段，但得到的例外情况如下所示。
代码：
public static void main(String[] args) throws Exception {
String brokers = "my.kafka.broker" + ":" + "6667";
String topics = "MyKafkaTopic";
// Create context with a 2 seconds batch i
浏览 7提问于2016-10-14得票数 4
回答已采纳
1回答
我正在尝试创建一个星火来流来自卡夫卡主题的消息。因此，我在构建中添加了以下依赖项：
"org.apache.spark:spark-streaming-kafka_2.10:1.6.2"
然后我创建了以下类：
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import
浏览 0提问于2016-10-10得票数 0
1回答
我已经在kafka中创建了一个示例主题，并尝试使用以下脚本使用spark中的内容：
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import
浏览 0提问于2018-09-15得票数 0
我正在尝试将Kafka集成到我的Spark应用程序中，这是我的POM文件所需的条目：
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>${spark.stream.kafka.version}</version>
</dependency>
<dependency&g
浏览 0提问于2017-09-01得票数 4
1回答
我能够在PySpark中通过两个不同的步骤从Kafka主题读取流，并将(转换后的)数据写回另一个Kafka主题。执行此操作的代码如下所示：
# Define Stream:
df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "localhost:9092") \
.option("subscribe", "instream") \
.load()
# T
浏览 0提问于2018-10-07得票数 0
我使用下面的python生成器将一些msg发布到我的kafka主题(我也可以使用jupyter中的python消费者接收我发布的数据)。
from kafka import KafkaProducer
import json,time
userdata={
"ipaddress": "172.16.0.57",
"logtype": "",
"mid": "",
"name":"TJ"
}
produce
浏览 9提问于2017-08-28得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
