我需要使用Kafka主题，它为每一行生成动态Json字符串，我无法在没有模式的情况下解析Json字符串。在我的例子中，Schema可以是动态的。spark.read.json可以推断json schema.But，它需要"DATASET“或"JSON文件”。
有没有办法把卡夫卡主题(价值)转换成数据集？这样我就可以使用接受DATASET作为输入
浏览 1提问于2020-06-04得票数 0
回答已采纳
1回答
因为我发现Azure Service Bus没有spark结构化流源，所以在这种情况下，我可以使用提供的Python客户端读取Azure Service Bus消息，然后从Python客户端读取每个消息并将其写入Kafka主题，在这个Kafka主题上，我将应用spark结构化流编程。我的用例是使用Azure Service Bus流消息，并通过将其转换为时间流数据库InfluxDb
浏览 0提问于2021-03-09得票数 0
我正在尝试使用结构化流从Kafka读取数据。从kafka收到的数据采用json格式。我的代码如下所示:在代码中，我使用from_json函数将json转换为dataframe以供进一步处理。",new StructType() . )
va
浏览 3提问于2019-10-15得票数 4
回答已采纳
我有一个spark 2.0应用程序，它使用spark streaming从kafka读取消息(使用spark-streaming-kafka-0-10_2.11)。结构化流媒体看起来真的很酷，所以我想试着迁移代码，但我不知道如何使用它。
在常规的流中，我对createDstrean使用kafkaUtils，并且在我传递的参数中，它是值反序列化器。在结构化流
浏览 3提问于2016-11-20得票数 9
3回答
当我从Kafka主题创建一个流并打印它的内容时 os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell' from pyspark.streamingimport Stream
浏览 2提问于2019-12-08得票数 1
回答已采纳
1回答
我不知道星火‘结构化API’是什么意思。如果你能更具体地告诉我，我将不胜感激。
对不起，我没有完美的英语。(我得到了翻译的帮助。)
浏览 1提问于2019-06-06得票数 2
回答已采纳
3回答
我使用的是Spark 2.1.0和Kafka 0.9.0。有没有人知道这样的事情是否可行？我用的是spark shell：
spark-shell --packages org.apache.spark:spark
浏览 3提问于2018-04-06得票数 6
1回答
我有一个连接到Apache Kafka数据流作为源的Spark 2.0.2结构化流作业。该工作从Kafka获取Twitter数据(JSON)，并使用CoreNLP为数据添加情感、词性标注等。然而，当我设置一个独立的Spark集群时，只有一个worker被用来处理数据。我有两个具有相同能力的工人。
提交作业时是否需要设置缺少的内容。我尝试在我的spark</
浏览 0提问于2016-12-06得票数 1
我正在读取来自Kafka源的流数据，但所有来自kafka的数据都是在单个微批次中读取的。spark.readStream.format("kafka").option("kafka.bootstrap.servers",bootstrap_servers).option("subscribe", topics).option("startingOf
浏览 2提问于2020-06-01得票数 1
1回答
我正在尝试使用结构化流媒体从Kafka中读取数据。从kafka收到的数据为json格式。我使用样例json创建模式，稍后在代码中使用from_json函数将json转换为数据帧以进行进一步处理。从kafka读取的json数据对于同一标签可以有一个或多个值(两个不同的值)。val df0= spark</
浏览 6提问于2019-06-11得票数 0
()我已经做了: 1.从twitter 2获取流；2.筛选并映射它以获取每个twitt你们知道如何将Dstream转换成流数据集/DataFrame吗？
编辑: ForeachRDD函数确实将Dstream更改为正常的Data
浏览 2提问于2019-11-13得票数 1
回答已采纳
我有一个DataSet[Row]，其中每一行都是JSON字符串。我只想打印JSON流，或者对每批JSON流进行计数。到目前为止，以下是我的代码 .format("kafka")
.optionHowever when I run my Kafka consumer separately (independent ofSpark I can
浏览 1提问于2017-05-14得票数 1
1回答
我对结构化流很陌生，我想知道是否有一种方法可以指定Kafka值的schema，就像我们在普通结构化流作业中所做的那样。Kafka值中的格式是50+字段，类似syslog的csv，手动拆分非常慢。下面是我代码的简短部分() .option("kafka.bootstrap.servers",
浏览 0提问于2018-08-07得票数 2
回答已采纳
下面的代码从Kafka读取消息，消息在Avro中，所以我如何解析该消息并将其放入Spark2.2.0中的数据文件中？Dataset<Row> df = sparkSession.readStream() .option("kafka.bootstrap.servers", "localhost
浏览 2提问于2017-12-16得票数 3
回答已采纳
我使用Spark 2.4.3和Kafka 2.3.0。我想用从Kafka到Spark的数据做Spark结构化流媒体。一般来说，它可以在测试模式下工作，但由于我必须对数据进行一些处理(并且不知道另一种方法)，Spark数据帧不再具有流式传输功能。 #!the stream .format("kafka</em
浏览 22提问于2019-09-10得票数 0
回答已采纳
我是数据科学专业的新手，正在寻求帮助。: org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
<e
浏览 0提问于2018-08-08得票数 0
我想使用星火流，并连接它与卡夫卡。然而，我仍然得到了NoSuchMethodError: NoSuchMethodError错误，现在我只是不知道下一步该做什么。.load()
.format("console")\我在星火上运行它spark-submit --master local[2] --jars /home/some_path/spar
浏览 1提问于2018-02-09得票数 3
我是新的火花的结构化流和工作的poc，需要实现的结构化流。我试图在预定义结构的火花数据中格式化传入的json。到目前为止，我能够获取json事件，并能够在控制台中获得结果(而不是预期的格式)。如果你能推动我朝着正确的方向前进，或者提出一个解决方案，那将是非常有帮助的。
下面是我目前的代码。bin
浏览 5提问于2022-03-11得票数 1
回答已采纳
从读取数据流到Apache的标准方法是： .format("json") \ # or parquet, kafka, orc# format specific options .load("path/to/data")
但是在应用架构之前，我需要清理一些数据重新排
浏览 4提问于2017-11-10得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
