如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？我从Kafka获得消息流，我想对它应用Map操作，对于每个键，我想查询像Cassandra这样的数据存储，并获得该键的更多信息，并在流上应用进一步的操作。我如何使用Spark Structured Streaming 2.2.0来做到这一点？
浏览 2提问于2017-09-07得票数 1
我正在尝试使用结构化流从Kafka读取数据。从kafka收到的数据采用json格式。我的代码如下所示:在代码中，我使用from_json函数将json转换为dataframe以供进一步处理。",new StructType() . )
va
浏览 3提问于2019-10-15得票数 4
回答已采纳
我正在尝试使用Java从kafka中读取复杂的嵌套JSON数据，但在形成数据集时遇到了问题 发送给kafka的实际JSON文件 {"sample_title": {"txn_date": "2019-().format("kafka")
.option("spark</em
浏览 19提问于2019-04-01得票数 1
回答已采纳
5回答
我正在尝试使用Spark结构化流从Kafka主题读取XML数据。df = spark \ .forma
浏览 19提问于2017-09-01得票数 6
回答已采纳
我正在将一个Java项目从Cloudera 5.10升级到Cloudera 6.2。我们有Spark Streaming从Kafka读取数据来处理它并将结果写到其他地方。在升级过程中，Spark从v1.6升级到v2.1，Kafka从v0.8升级到v2.1。为了执行流处理，我们使用KafkaUtils.createStream(...)连接到Kafka</e
浏览 11提问于2021-09-27得票数 0
我正在编写一个应用程序，将消息从Kafka读取到Spark使用结构化流。SparkSession spark = createSparkSession();
Dataset<Row> df =spark
浏览 2提问于2017-08-22得票数 1
1回答
使用Kafka Spark-Streaming。能够读取和处理生产者发来的数据。我在这里有一个场景，让我们假设生产者正在产生消息，而消费者被关闭了一段时间，然后打开了。现在，Conumser只读取实时数据。相反，它还应该保留它停止读取的数据。这是我一直在使用的pom.xml。行为与预期一致(消费者从数据离开的地方读取数据)。因此，在这个版本中，偏移量被正确地拾取。我也尝试过在上面的pom.xml中使用</em
浏览 3提问于2017-02-06得票数 0
我想了解apache如何处理具有AVRO模式的Kafka主题(模式包括嵌套对象、对象数组等)因为我没有找到任何资源或例子，说明我们如何使用带有avro模式的Kafka注入数据。根据我的理解，apache必须为嵌套的Json对象提供平面模式或其他选项，我们可以使用transform函数。有什么卡夫卡连接的皮诺做数据注入吗？] }} }
浏览 1提问于2021-05-21得票数 0
1回答
我正在尝试使用结构化流媒体从Kafka中读取数据。从kafka收到的数据为json格式。我使用样例json创建模式，稍后在代码中使用from_json函数将json转换为数据帧以进行进一步处理。我面临的问题是嵌套模式和多值。示例模式将标记(比如a)定义为结构。从kafka读取的json
浏览 6提问于2019-06-11得票数 0
1回答
带有火花流的Kafka抛出了一个错误：from pyspark.streaming.kafka import KafkaUtils
ssc = StreamingCon
浏览 2提问于2020-05-19得票数 4
回答已采纳
3回答
我没有太多使用Kafka/ Spark-Streaming的经验，但我读过很多关于组合在为分析/仪表板构建实时系统方面有多棒的文章。有人能给我解释一下为什么spark-streaming不能单独完成吗？换句话说，为什么Kafka在数据源和spark-streaming之间？
谢谢
浏览 1提问于2016-01-11得票数 1
我是spark的新手，正在尝试探索Spark结构化流媒体。我将使用来自Kafka(嵌套的JSON)的消息，根据JSON属性上的特定条件过滤这些消息。的文档 Dataset<Row> df = spark.format("kafka")
.option("kafka.bootstrap.servers", "host1:
浏览 17提问于2019-04-12得票数 0
我使用的是Spark流媒体应用程序。应用程序使用直接流从Kafka topic (具有200个分区)中读取消息。:1431)at org.apache.spark.streaming.kafka010Spark
浏览 1提问于2017-12-03得票数 2
回答已采纳
1回答
我在我的ElasticSearch集群中有一些与销售相关的JSON数据，我想使用Spark (使用Spark1.4.1)动态地通过Kafka从我的eCommerce网站聚合即将到来的销售事件，以便对用户的总销售额从我所读到的文档中，我不太清楚的是，如何在Spark应用程序启动时从ElasticSearch加载历史数据，并计算每个用户的总收入(基于历史记录，以及从<
浏览 2提问于2015-07-27得票数 3
回答已采纳
我有3个kafka流，每个都有600k+记录，火花流需要超过10分钟来处理流之间的简单连接。Spark群集配置：这就是我在spark(scala)中读取kafka streams to tempviews的方法.format("kafka").option("endingOffsets", &
浏览 0提问于2020-03-04得票数 1
我正在尝试读取星火流应用程序中的kafka流数据；而在读取数据的过程中，我得到了以下例外： at java.net.URLClassLoader.findClass(URLClassLoader.java:
浏览 0提问于2016-12-24得票数 0
在spark批处理作业中，我通常会将JSON数据源写入到一个文件中，并且可以使用DataFrame阅读器的损坏列功能将损坏的数据写出到不同的位置，并使用另一个阅读器从同一作业中写入有效数据。(数据写为拼接) 但在Spark Structred Streaming中，我首先通过kafka将流作为字符串读取，然后使用from_json</em
浏览 34提问于2018-12-27得票数 1
回答已采纳
1回答
Hadoop可以做流媒体吗？
、、、、
有人建议Hadoop做流媒体，并引用了Flume和Kafka作为例子。
虽然我知道它们可能具有流媒体功能，但我想知道它们是否可以与Storm/Spark/Flink等流处理技术并驾齐驱。Kafka是一个“发布-订阅模型消息传递系统”，而Flume是一个数据摄取工具。即使它们与hadoop交互/集成，从技术上讲，它们是“hadoop”本身的一部分吗？
浏览 4提问于2018-04-07得票数 0
我正在Scala中实现Spark Streaming，我从Kafka主题中提取JSON字符串，并希望将它们加载到dataframe中。有没有一种方法可以让Spark自己从RDDString中推断出模式
浏览 0提问于2015-06-26得票数 9
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
