readStream kafka没有得到任何值
readStream kafka是一个用于从Kafka消息队列中读取数据的操作。它是一种流式读取数据的方式，可以实时获取Kafka中的消息。
Kafka是一个分布式的流处理平台，具有高吞吐量、可扩展性和容错性。它主要用于处理实时数据流，支持发布-订阅模式，可以将数据流分发到多个消费者进行处理。
在使用readStream kafka时，如果没有得到任何值，可能有以下几个原因：
- 消息队列中没有数据：首先需要确认Kafka消息队列中是否有待读取的数据。可以通过查看Kafka的管理界面或使用Kafka命令行工具来确认。
- 读取的topic或分区不存在：readStream kafka需要指定要读取的topic和分区。如果指定的topic或分区不存在，将无法获取到任何值。需要确认topic和分区的名称是否正确。
- 读取偏移量不正确：Kafka使用偏移量（offset）来标识消息在分区中的位置。如果读取的偏移量不正确，可能导致没有获取到任何值。可以尝试重新设置读取的偏移量，或者查看是否有其他消费者已经读取了该偏移量之后的消息。
- 网络连接或配置问题：readStream kafka需要与Kafka集群建立网络连接，并正确配置相关参数。如果网络连接不稳定或者配置有误，可能导致无法获取到消息。需要检查网络连接是否正常，并确认配置参数是否正确。
推荐的腾讯云相关产品：腾讯云消息队列 CMQ、腾讯云流数据分析 Kafka。
腾讯云消息队列 CMQ：提供高可靠、高可用的消息队列服务，支持消息的发布和订阅，适用于构建分布式系统、微服务架构等场景。产品介绍链接：https://cloud.tencent.com/product/cmq
腾讯云流数据分析 Kafka：提供高吞吐量、低延迟的分布式流处理平台，支持实时数据处理和分发。适用于大数据分析、实时监控、日志处理等场景。产品介绍链接：https://cloud.tencent.com/product/ckafka
相关·内容
我试图通过spark-shell中的Spark structured streaming来阅读Kafka的主题，但似乎我没有从Kafka那里得到任何线条。Kafka单独运行良好(使用控制台-消费者和控制台-生产者进行测试)：bladas ist
浏览 4提问于2017-12-12得票数 1
回答已采纳
1回答
当Spark作业在没有Docker的情况下通过spark-submit在本地运行时，一切工作正常。但是，在停靠容器上运行会导致不生成任何输出。为了查看Kafka本身是否正常工作，我将Kafka提取到Spark worker容器上，并让控制台使用者监听相同的主机、端口和主题(kafka:9092，crypto_topic)，它们工作正常并显示输出: container_nam
浏览 21提问于2020-09-11得票数 0
我有一个Spark结构化流媒体应用(v2.3.2)，它需要从一些Kafka主题读取，做一些相对简单的处理(主要是聚合和一些连接)，并将结果发布到其他一些Kafka主题。如果我只设置一个直接readStream，它订阅多个主题，然后使用selects拆分流，而不是每个主题1个readStream。就像这样 df = spark.readStream.format("kafka").option("subscribe", "t1,t2,t3&
浏览 28提问于2019-05-01得票数 11
回答已采纳
1回答
在星火流中，当它们到达时，我得到了日志。但是我想在一次传递中得到至少N个日志。如何才能做到这一点？
从的回答来看，卡夫卡似乎有这样一个实用工具，但却让它成为可能。
浏览 2提问于2021-01-22得票数 2
回答已采纳
我有一个火花结构流作业，它被配置为从Kafka读取数据。请查看代码，带参数查看readStream()，读取Kafka的最新数据。我知道当一个新的查询开始而不是恢复时，readStream()会从第一个偏移量开始读取。 .readStream .option("kafka
浏览 0提问于2018-05-12得票数 3
1回答
我正在使用Spark3.3.0和Kafka3.2.0。我希望在readStream中使用组id，但是当spark没有运行时，数据丢失了。当火花停止时，我如何停止数据流？我的代码； .format("kafka") .option("subscribe", "topic-
浏览 5提问于2022-08-31得票数 0
// read from kafka .readStream .load()
writeStream.aw
浏览 2提问于2021-03-14得票数 3
考虑到来自两个主题的数据在一点连接并最终发送到Kafka接收器，这是从多个主题读取的最佳方式 val df = spark .format("kafka") .option("subscribe", "t1,t2") vs val df1 = spark
.readStream.format(&q
浏览 12提问于2020-04-13得票数 0
回答已采纳
我通过waterMark从kafka topic (startOffset: latest)获取readStream，按事件时间和窗口时长分组，并写入kafka topic。我的问题是，在spark结构化流媒体作业之前，我如何处理写入kafka主题的数据？如果我简单地创建一个批处理作业并按特定的数据
浏览 87提问于2018-11-26得票数 2
回答已采纳
我需要使用Kafka主题，它为每一行生成动态Json字符串，我无法在没有模式的情况下解析Json字符串。在我的例子中，Schema可以是动态的。有没有办法把卡夫卡主题(价值)转换成数据集？这样我就可以使用接受DATASET作为输入的spark.read.json，来解析json的模式。 format("kafka").
option("kafka.bootstrap.servers", &quo
浏览 1提问于2020-06-04得票数 0
回答已采纳
怀疑他们可能没有正确的值。如何打印回以验证spark structured Streaming选项？val df = sparkSession .format("kafka")
.option("kafka.bootstrap.servers", "host1:port1
浏览 30提问于2021-03-02得票数 0
回答已采纳
3回答
我从卡夫卡的来源得到日志，然后把它放进火花里。Jin\",\"Age\":\"26\"}"}{\"Name\":\"Amy\",\"Age\":\"22\"}任何一种解决办法都会很好(使用纯Java代码、Spark或Kafka)
浏览 1提问于2019-07-22得票数 0
回答已采纳
1回答
我使用spark structured来读取来自Kafka主题的传入消息，并根据传入消息写入多个镶嵌表，因此我创建了一个readStream，因为Kafka源很常见，并且为每个镶嵌表在循环中创建了单独的写入流这工作得很好，但是readstream正在造成一个瓶颈，因为它为每个writeStream创建了一个readStream，并且没有办法缓存已经读取的数据帧。val kafkaDf=spark.readStream
.format("ka
浏览 1提问于2018-10-10得票数 0
1回答
我的end goal是write out和read，将聚合的数据按批处理到新的Kafka topic。我跟踪了official documentation和其他几个帖子，但没有运气。SparkSession.builder().appName("SparkKafka").master("local[*]").getOrCreate() .readStream.option("checkpointLocation", "&#
浏览 4提问于2020-06-07得票数 0
回答已采纳
我正在编写一个Spark应用程序，它从Kafka主题读取消息，在DB中查找记录，构造新消息并将它们发布到另一个Kafka主题。下面是我的代码： val inputMessagesDataSet: DataSet[InputMessage] = spark.readStream .option("checkpointLocation", loc) .awaitTermination
} 但是我得到<
浏览 20提问于2020-09-09得票数 0
回答已采纳
我有一个简单的结构化流媒体应用程序，它只是从一个Kafka主题读取数据，然后写入另一个主题。.builder() .getOrCreate();
.readStream() .option("kafka.bootstrap.servers", "localhost
浏览 38提问于2018-06-27得票数 3
回答已采纳
3回答
我之前已经成功地使用了pyspark for Spark Streaming (Spark 2.0.2)和Kafka (0.10.1.0)，但我的目的更适合于结构化流媒体。我尝试在线使用示例：ds1 = spark .format("kafka")
.option("kafka.bootstrap.servers文档中可以找到的任何其他值(如“舍入”)也无法阻止错误。我也尝试了"assig
浏览 2提问于2017-07-07得票数 1
1回答
我正在使用SparkReadStream阅读所有20+主题(逗号分隔的主题值)。然后过滤结果DataFrame中的每个消息行，使用正确的Avro模式应用每条消息，并将得到的Dataset[T]写入S3和Cassandra。spark是否使用不同的执行器并行读取这些主题/分区，还是需要为每个分区实现任何多线程？
v
浏览 2提问于2017-12-18得票数 4
1回答
我正在尝试使用星火流/python从Kafka主题读取流，我可以读取该消息并将其转储到带有默认kafka消息模式的青铜表中，但我无法将键和值从二进制转换为字符串，我尝试了以下方法，但它们都没有工作：raw_kafka_events = (spark.readStream .option("subscribe", TOPIC)
浏览 11提问于2022-08-29得票数 1
回答已采纳
卡夫卡的Spark readStream失败了，有以下错误：
代码： .readStream .option("kafka.bootstrap.servers", "earliest&qu
浏览 0提问于2017-08-09得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
