我正在尝试编写一个使用结构化流媒体的Kafka消费者(一个protobuf)。让我们调用protobuf as，它应该在Scala中反序列化为字节数组(ArrayByte)。我尝试了所有我能在网上找到的方法，但仍然不能正确地解析消息A 方法1:在集成指南(https://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html= SQL
浏览 6提问于2019-09-24得票数 0
回答已采纳
我们有Spark Streaming从Kafka读取数据来处理它并将结果写到其他地方。在升级过程中，Spark从v1.6升级到v2.1，Kafka从v0.8升级到v2.1。为了执行流处理，我们使用KafkaUtils.createStream(...)连接到Kafka，但KafkaUtils在Kafka 2.11中不再可用。然而，我似乎找不到任何Spark Streaming + Kafka在Java中不使用此方法的示例或文档。 我是不是漏掉了什么？在
浏览 11提问于2021-09-27得票数 0
Spark streaming应用程序实时接收来自大量IoT设备的数据。但它们都是少量的数据。整体流程看起来像是-> Iot -> Kafka(1个主题/所有数据) ->火花流(过滤错误日志) -> DB(保存) ->警报屏幕 有没有使用spark或python进行实时日志分析的好方法？
浏览 13提问于2020-11-19得票数 0
回答已采纳
我想使用流处理的概念从本地目录读取文件，然后发布到Apache Kafka。我考虑过使用Spark Structured Streaming。此外，如果我们在结构化流媒体中使用检查点，当代码有任何升级或任何更改时，我们是否会有任何问题。
浏览 0提问于2019-02-25得票数 2
我使用结构化数据流从Kafka流式传输数据 .readStream .option("kafka.bootstrap.servers.option("group.id", UUID.randomUUID().toString) .loa
浏览 0提问于2018-11-16得票数 1
我正在尝试连接我的结构化流式spark 2.4.5与kafka，但所有的时间，我尝试这个数据源提供程序出现错误。遵循我的scala代码和sbt构建： import org.apache.spark.sql._import org.apache.spark.sql.functionsval KAFKA_TOPIC_NAME_CONS = "test&qu
浏览 86提问于2020-05-04得票数 0
回答已采纳
1回答
我正在尝试将结构化的流式数据文件发送到我的卡夫卡主题之一，detection。这是结构化流数据格式的架构： |-- timestamp: timestamp (nullable = true)dfwriter=df \ .writeStream \
.format("
浏览 9提问于2021-11-02得票数 0
回答已采纳
我想把推特上的数据写进卡夫卡。出于教育目的，我尝试使用结构化流来实现这一点。我创建了一个基于socket-Source的Twitter-Source，它运行得很好。接下来，我想将每条推文都保留在稍微闪闪发亮的模式中，直到Kafka： .toJSON.as("value")Ex
浏览 0提问于2017-08-10得票数 2
回答已采纳
我想创建一个Storm Spout，它从Apache Kafka的主题中读取数据，并将这些数据发送到连接到MongoDB的Storm bolt，并查询我从Kafka收集的消息以丰富数据。例如:我有一个personID (我是通过Kafka发来的消息得到的)，我想使用这个personID查询MongoDB中的个人地址。在我的MongoDB集合中，每个文档都有personID和地址。一
浏览 0提问于2018-04-18得票数 0
1回答
我想从Kafka源创建Spark结构化流式作业读取消息，写入Kafka接收器，失败后将恢复只读取当前，最新的消息。出于这个原因，我不需要为我的工作设置检查站。但它看起来没有选项来禁用检查点，而写入到Kafka接收器的结构化流。
浏览 2提问于2020-06-23得票数 0
我正在对来自kafka的数据运行结构化流式处理(spark 2.1.0 with zeppelin 0.7)，并尝试使用spark.sql可视化流式处理结果%spark2 .builder() .master("yarn").getOrCreate()
浏览 4提问于2017-07-13得票数 6
1回答
这是我运行strm.py文件的终端命令
无法使用URI org.apache.spark:spark-cassand
浏览 0提问于2020-03-04得票数 0
我将使用pyspark进行流处理，并使用Kafka作为数据源。
我发现Spark Python API不支持Kafka 0.10连接器。我可以在Spark 2.3.0中使用Kafka 0.8连接器吗？
浏览 14提问于2018-03-02得票数 4
我是spark的新手。我使用结构化流媒体从kafka中读取数据。我可以在Scala中使用以下代码读取数据： .format("kafka") .option("startingOffsets"
浏览 1提问于2017-10-19得票数 2
我使用从Kafka Source到PySpark Dataframe的结构化流。Kafka提供的数据类型是JSON，结构如下： {"user_id":1,...,id |user_id|status |
---------
浏览 15提问于2020-06-19得票数 1
回答已采纳
我有一个Spark Streaming应用程序，可以从Kafka中读取记录中的配置单元表名，例如table1。table2..table3..等。我想在hive表上执行结构化查询，并将结果流式传输到另一个Kafka主题。val hqls = rdd当我执行此命令时，我在sparkSession.sql调用中得到一个NullPointerException，如下所示：
org.ap
浏览 0提问于2019-02-25得票数 1
2回答
我们想知道是否有更好的方法对数据进行批处理，并在进行批处理时避免数据丢失。initialize(topic, consGroup, STREAMSERDE.STRINGDESER, STREAMSERDE.JSONDESER, props); throwable.printStackTrace(); }克隆数据以避免数据丢失的代码listOfMessages.a
浏览 1提问于2019-01-22得票数 0
1回答
我们正在使用spark 2.4.3运行一个结构化的流式处理过程，它从kafka读取数据，转换数据(使用udf平面化并创建一些列)，然后将数据写回kafka到不同的主题。10-12小时后，我们注意到由于高内存消耗，我们的pod正在下降。正如我上面解释的那样，我们没有聚合，也没有在数据集上使用持久化。我们注意到的是堆内存在不断增长。有什么想法吗？
浏览 17提问于2019-07-08得票数 1
1回答
我有一个结构化的流式作业，从kafka读取，执行聚合和写入hdfs。该作业在yarn中以群集模式运行。我使用的是spark2.4。每隔2-3天，这项工作就会卡住。它不会失败，但会卡在某个微批处理中。当我终止流式作业并重新启动时，该作业再次开始正常运行。如何解决这个问题？
浏览 34提问于2020-10-03得票数 0
1回答
我使用的是Kafka和Spark结构化流媒体。我收到了以下格式的kafka消息。Dataset<String> data = spark .format("kafka")在我的json消息中，"data“是byte[]的字符串形式。
浏览 0提问于2017-02-24得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
