我正在从spark (结构化数据流)中读取Kafka中的数据，但在spark中从kafka中获取的数据不是字符串格式。},"Gender":"female"}# spark-submit --jars kafka-client
浏览 1提问于2019-12-27得票数 0
下面的代码从Kafka读取消息，消息在Avro中，所以我如何解析该消息并将其放入Spark2.2.0中的数据文件中？Dataset<Row> df = sparkSession.readStream() .option("kafka.boot
浏览 2提问于2017-12-16得票数 3
回答已采纳
在这个项目中，我要从Kafka流接收解析数据(Proto Buf消息)我正在尝试理解下面的代码来开始解析protobuf消息。master("local") .getOrCreate()
import spark</e
浏览 0提问于2017-06-07得票数 2
我有一个需要在spark ( java )中解析的事件流(格式如下)。我能够读取流，但是还找不到将消息转换为java bean的示例。session_id; private String event;} 将消息作为字符串读取的代码Dataset<String> lines = <e
浏览 26提问于2020-01-06得票数 0
回答已采纳
1回答
我正在尝试用spark做一个结构化的流媒体应用程序，主要思想是从kafka源读取，处理输入，写回另一个主题。我已经成功地让spark读写了kafka，但是我的问题出在处理部分。我尝试过foreach函数来捕获每一行并在写回kafka之前对其进行处理，但是它总是只做foreach部分，而不会写回kafka。但是，如果我从writestream中删除foreach部分，它将继续写入，但现在我丢失了处理。
浏览 37提问于2019-07-17得票数 0
回答已采纳
在spark批处理作业中，我通常会将JSON数据源写入到一个文件中，并且可以使用DataFrame阅读器的损坏列功能将损坏的数据写出到不同的位置，并使用另一个阅读器从同一作业中写入有效数据。(数据写为拼接) 但在Spark Structred Streaming中，我首先通过kafka将流作为字符串读取，然后使用from_json获取我的
浏览 34提问于2018-12-27得票数 1
回答已采纳
我正在尝试使用结构化流从Kafka读取数据。从kafka收到的数据采用json格式。我的代码如下所示:在代码中，我使用from_json函数将json转换为dataframe以供进一步处理。",new StructType() . )
val df: Dat
浏览 3提问于2019-10-15得票数 4
回答已采纳
我正在尝试通过kafka将json文件解析为spark流。found : org.apache.spark.rdd.RDD[Array[Byte]] required: Array[Byte] parseAVROToString(rdd
浏览 3提问于2016-07-13得票数 1
回答已采纳
我正在使用星火结构化流处理来自卡夫卡的数据。我将每条消息转换为JSON。但是，spark需要一个显式的模式才能从JSON获得列。使用DStreams的火花流允许执行以下操作其中jsons是RDD[String]。在星火结构流的情况下类似的方法
df.spar
浏览 1提问于2018-02-05得票数 2
我创建了一个带有JSON格式记录的Kafka主题。./kafka-console-consumer.sh --new-consumer \ --from-beginning \ --consumer.config /root/client.propert
浏览 5提问于2017-05-24得票数 1
回答已采纳
我需要使用Kafka主题，它为每一行生成动态Json字符串，我无法在没有模式的情况下解析Json字符串。在我的例子中，Schema可以是动态的。这样我就可以使用接受DATASET作为输入的spark.read.json，来解析json的模式。val klines = spark.
readStrea
浏览 1提问于2020-06-04得票数 0
回答已采纳
3回答
我使用的是Spark 2.1.0和Kafka 0.9.0。有没有人知道这样的事情是否可行？我用的是spark shell：
spark</em
浏览 3提问于2018-04-06得票数 6
2回答
为了捕捉Server中的数据更改，我尝试执行从Server到Pyspark的管道，我已经准备好了一切：
从SQL Server到Kafka进行生产，并使用Pyspark结构化流中的Kafka主题。问题是:当我试图用控制台使用者检查数据更改是否经过Kafka时，它会向我显示JSON格式的消息，分为
浏览 0提问于2021-06-30得票数 0
回答已采纳
我正在尝试写一个重要的星火结构的流媒体作业，读从卡夫卡。作为需求的一部分，我需要将'event_time‘作为一个额外的列添加到流中。我正在尝试这样的方法：val myStream = sparkSession.readStream
浏览 2提问于2020-03-03得票数 0
1回答
您能提供一些关于使用Python使用Azure Service Bus流消息的建议吗？因为我发现Azure Service Bus没有spark结构化流源，所以在这种情况下，我可以使用提供的Python客户端读取Azure Service Bus消息，然后从Python客户端读取每个消息并将其写入Kafka主题，在这个Kafka主题上，我将应用spark结构化流编
浏览 0提问于2021-03-09得票数 0
1回答
火花版本: 2.3.0卡夫卡版本: 2.2.0
我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。
浏览 3提问于2019-05-02得票数 3
回答已采纳
1回答
我使用Spark3.2从Kafka 2.12-3.0.0获得JSON流。解析JSON后，我收到查询中的错误。Kafka主题流JSON：
b'{"pmu_id": 2, "time": 1642771653.06, "stream_id": 2,"analog": [], "digit
浏览 5提问于2022-01-21得票数 1
回答已采纳
1回答
我正在设计一个火花流应用与卡夫卡。我有以下几个问题:将RDBMS表中的数据流到kafka，并使用Spark使用者使用Spark处理消息和处理。问题: 1.我将数据从表流到kafka (键作为tablename，值作为表数据形式的JSON记录) --这是正确的体系结构吗？在spark使用者中，我试图使用DStream.foreachRDD(x =&
浏览 0提问于2017-01-31得票数 0
我正在Scala中实现Spark Streaming，我从Kafka主题中提取JSON字符串，并希望将它们加载到dataframe中。有没有一种方法可以让Spark自己从RDDString中推断出模式
浏览 0提问于2015-06-26得票数 9
我正在学习apache spark与kafka的集成，这样当新消息到达Kafka的主题时，我的代码就可以自动运行。我也读过官方文档 https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html 但我仍然对它的工作原理感到困惑我有我的字数统计写的程序在j
浏览 15提问于2019-12-10得票数 1
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
