我正在编写一个使用结构化流媒体的Spark应用程序。该应用程序从卡夫卡主题topic1读取消息，构造新消息，将其序列化到Array[Byte]，并将其发布到另一个卡夫卡主题topic2。序列化到字节数组很重要，因为我使用了一个特定的序列化程序/反序列化程序，topic2的下游使用者也使用该序列化程序/反序列化程序。 不过，我在制作Kafka时遇到了麻烦。我甚至不确定如何做到这一点……网上只有很多关于JSON数据排队的例子。val outputMess
浏览 38提问于2020-09-11得票数 0
回答已采纳
1回答
我正在学习Udemy上关于Kafka和Spark的课程，我正在学习apache spark与Kafka的集成 下面是apache spark的代码 SparkSession session = SparkSession.builder-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming -->
<!--
浏览 26提问于2020-09-25得票数 4
回答已采纳
我想使用Apache Spark Structured Streaming和Kafka，Spark Structured Streaming支持Kafka 0.10及更高版本，我的Kafka集群使用kafka我想将当前kafka 0.8.2.1集群中的一些主题复制到基于2.2.0的新Kafka集群中。为此，我尝试在Kafka 2.2.0集群上使用
浏览 17提问于2019-05-23得票数 1
我经常使用HDFS和Kafka，我注意到Kafka比HDFS更可靠。所以现在使用Spark-structured-streaming，我很惊讶检查点只有HDFS。用Kafka来勾点会更快更可靠。那么，有没有可能在没有HDFS的情况下使用spark structured？奇怪的是，我们只能在Kafka中将HDFS用于流数据。Spark 2.4.7 谢谢
浏览 39提问于2021-05-07得票数 2
我正在尝试编写一个使用结构化流媒体的Kafka消费者(一个protobuf)。让我们调用protobuf as，它应该在Scala中反序列化为字节数组(ArrayByte)。我尝试了所有我能在网上找到的方法，但仍然不能正确地解析消息A 方法1:在集成指南(https://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html)中，我应该将值</
浏览 6提问于2019-09-24得票数 0
回答已采纳
我使用spark(3.0.0)结构化流从kafka读取主题。因此，我的问题是:在我的情况下，如何将流数据写入数据库或文件？我是否必须将数据写入kafka，然后使用kafka连接将数据读回文件/db？
附注：我跟踪了这些文章以获得aggregated流查询。stackoverflow.com/questions/62738727/how-to-deduplicate-and-keep-latest-based-on-timestamp-field-in-spa
浏览 5提问于2020-09-16得票数 0
我正在学习apache spark与kafka的集成，这样当新消息到达Kafka的主题时，我的代码就可以自动运行。我也读过官方文档 https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html 但我仍然对它的工作原理感到困惑结构化流媒体是kafka和spark java代码之间的桥梁吗？它是否一直在监听<
浏览 15提问于2019-12-10得票数 1
1回答
Kaka主题定义为配置文件中的列表，每个Kafka主题都有一个目标表。val query1 = df.writeStream.start()
spark.streams.awaitAnyTermination
浏览 2提问于2020-11-06得票数 1
回答已采纳
怀疑他们可能没有正确的值。如何打印回以验证spark structured Streaming选项？val df = sparkSession .format("kafka")
.option("kafka.bootstrap.servers", "host1:port1
浏览 30提问于2021-03-02得票数 0
回答已采纳
我在用我配置了两个JARS(spark-sql-kafka-0-10_2.12-3.0.0-preview2.jar和kafka-clients-2.2.0.jar)，并将其保存在$SPARK_HOME/jars文件夹中。当我试图查看来自Kafka服务器的数据的键，值(因为Kafka的数据以JSON格式以Key对表示)时，我面临
浏览 9提问于2020-04-08得票数 2
我想向我的应用程序spark & kafka添加一些参数，以便将Dataframe写入主题Kafka。我在spark-kafka文档中没有找到ack和compression.codec .write .option("kafka.sasl.mechanism", Config.KAFKA_SASL_MECHANISM)
.option("
浏览 39提问于2019-06-21得票数 1
1回答
屏蔽来自Kafka流的数据
、、、、
我使用spark Structured来流式传输来自kafka的数据，它为我提供了以下模式的数据帧 Column Typevalue binarytopic stringoffset longtimestampType int 值Colum是以二进制格式出现的
浏览 23提问于2020-10-16得票数 0
/kafka_project'
print("PySpark Structured Streaming with Kafka Demo Applicationspark = SparkSession \ .appName("PySpark Structured Streaming with <e
浏览 10提问于2020-07-27得票数 1
2回答
Streaming's Kafka libraries not found in class path.--packages org.apache.spark:spark-streaming-kafka-0-8:2.3.0 ...
2., Artifact Id = spark-streaming-kafka-0-8-assembly, Version = 2.3.0.Then, include the jar in th
浏览 0提问于2018-06-29得票数 1
回答已采纳
.appName("PySpark Structured Streaming with Kafka Demo") \
.config("spark.jars", "/home/bupry_dev/development/spark_home/spark-2.4.7-bin-hadoop2.7/jars/spark-streaming</
浏览 0提问于2021-01-25得票数 1
我正在尝试从Kafka读取JSON消息，并使用spark structured将它们存储在HDFS中。我遵循了示例，当我的代码如下所示时： .read \ .option("kafka.bootstrap.servers", "df.writeStream.format("json").option("checkpointLocati
浏览 43提问于2018-07-25得票数 1
我正在尝试使用火花结构化流(Sparkv2.2.0)来使用kafka提供的json数据。但是，我遇到了以下错误。pyspark.sql.utils.StreamingQueryException：‘缺少必要的配置“"partition.assignment.strategy”，它没有默认值。spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 sparksstream.py
这是整个py
浏览 4提问于2017-10-10得票数 7
2回答
使用HDFS存储的Spark作业
、、、、
我有一个长期运行的Spark Structured Streaming Job，它运行在Google Cloud Dataproc上，使用Kafka作为源和接收器。运行一周后，我注意到它正在稳定地消耗所有100 GB的磁盘存储空间，将文件保存到/hadoop/dfs/data/current/BP-315396706-10.128.0.26-1568586969675=org.apache.spark:spark-<em
浏览 0提问于2019-10-01得票数 4
1回答
我是第一次接触kafka和spark structured streaming。我想知道spark in batch模式是如何知道从哪个偏移量读取的？df = spark \ .format("kafka") \
.option("kafka.bootstrap.servers", broker) \startingOffsets", "earli
浏览 0提问于2020-10-24得票数 0
我正在尝试将Kafka与Spark-Structured Streaming集成到MongoDB Sink。如果我出错了，我需要帮助来纠正我的代码。集成了Kafka-Spark和Spark-Mongo。现在尝试集成来自Kafka-Spark-Mongo的管道 import org.apache.spark.sql.streaming.Trigger
import com.mongo
浏览 48提问于2019-04-15得票数 1
回答已采纳
云服务器
ICP备案
云直播
腾讯会议
实时音视频
腾讯云开发者
