我正在尝试从kafka读取一个流，其中的值是一个逗号分隔值的字符串(表示数据集中的列)，目标是读取两个这样的流并连接它们。val stearm_L: DataFrame = spark.readStream
浏览 11提问于2018-09-10得票数 1
回答已采纳
1回答
我有一个要求，从多个卡夫卡topicsAvro的流媒体，并把他们在格林梅利与小修改的有效载荷。val query1 = df.writeStream.start()或
df.writeStream.start().awaitAnyTermination(
浏览 2提问于2020-11-06得票数 1
回答已采纳
1回答
我使用星火流与Kafka集成，当我在本地模式下从IDE运行流应用程序时，一切都是有魅力的。我使用sbt程序集来构建我的项目。我的sbt就是这样： "org.apache.spark" % "spark-streaming-kafka-0-10_2.11" % "2.2.0尝试1
--pac
浏览 3提问于2017-08-22得票数 2
回答已采纳
我使用的是Spark流媒体应用程序。应用程序使用直接流从Kafka topic (具有200个分区)中读取消息。<init>(KafkaRDD.scala:211)
at org.apache.spark.streaming.kafka010.KafkaRDD.compute(KafkaRDD.scala:186Spark版本是2.1。应用程序运行两个执行器。从我<em
浏览 1提问于2017-12-03得票数 2
回答已采纳
我使用spark(3.0.0)结构化流从kafka读取主题。我使用了joins，然后使用了mapGropusWithState来获取流数据，因此，根据我对火花官方指南：的理解，我不得不使用更新模式。因此，我的问题是:在我的情况下，如何将流数据写入数据库或文件？我是否必须将数据写入kafka，然后使用kafka连接将数据读回文件&
浏览 5提问于2020-09-16得票数 0
我有火花流应用程序从Kafka读取消息使用火花直接流(非接收者)方法和处理每个分区的消息。spark.batch.duration=10 secondsspark.streami
浏览 2提问于2017-01-20得票数 3
我对spark结构化数据流和kafka的偏移管理是新的。使用spark-streaming-kafka-0-10-2.11。在消费者中，我如何从主题的特定分区中读取？comapany_df = sparkSession .format("kafka").option("<e
浏览 22提问于2019-05-29得票数 0
如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？我从Kafka获得消息流，我想对它应用Map操作，对于每个键，我想查询像Cassandra这样的数据存储，并获得该键的更多信息，并在流上应用进一步的操作。我如何使用Spark Structured Streaming 2.2.0来做到这一点？
浏览 2提问于2017-09-07得票数 1
2回答
所以我的项目流程是Kafka -> Spark Streaming ->HBase现在我不知道如何使用Spark Streaming从HB
浏览 30提问于2016-07-25得票数 1
回答已采纳
我写了一个火花流应用程序，它读取卡夫卡的数据。我已经用spark1.6.0和kafka0.8.2.2构建了jar。我正在使用kafka直接流apis：
KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams我在应用程序jar中打包了kafka类，甚至在运行时，kafka/api/Topi
浏览 0提问于2017-07-20得票数 1
我正在将一个Java项目从Cloudera 5.10升级到Cloudera 6.2。我们有Spark Streaming从Kafka读取数据来处理它并将结果写到其他地方。在升级过程中，Spark从v1.6升级到v2.1，Kafka从v0.8升级到v2.1。为了执行流处理，我们使用KafkaUtils.createStream(...)连接到Kafka，但Kafka
浏览 11提问于2021-09-27得票数 0
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.
浏览 2提问于2019-12-08得票数 1
回答已采纳
我希望使用火花结构流从卡夫卡读取数据，并处理它，并写入蜂窝表。val spark = SparkSession .appName("Kafka Test") .config("spark.streaming.backpres
浏览 5提问于2018-02-06得票数 1
1回答
我正在尝试通过Spark流读取从kafka传递的行，传入的记录有4列，其中一列有换行符，其中一列用双引号括起来a ,b , "hifine那么如何读取或者在通过spark streaming从kafka读取时需要设置哪些选项，以便正确读取记录？
浏览 4提问于2018-08-01得票数 0
尝试从kafka源文件中读取。我想从收到的消息中提取时间戳，以进行结构化火花流。kafka(0.10.0.0版) spark streaming(2.0.1版)
浏览 1提问于2016-11-14得票数 4
如何使用直播流API指定kafka spark流的消费组id。使用spark1.3
kafkaParams.put("group.id", "app1");
浏览 0提问于2016-04-09得票数 7
回答已采纳
1回答
您能提供一些关于使用Python使用Azure Service Bus流消息的建议吗？因为我发现Azure Service Bus没有spark结构化流源，所以在这种情况下，我可以使用提供的Python客户端读取Azure Service Bus消息，然后从Python客户端读取每个消息并将其写入Kafka主题，在这个Kafka主题上，我将应用spark结构化流编
浏览 0提问于2021-03-09得票数 0
1回答
我使用的是Spark 1.6和Kafka 0.8.2.1。
我正在尝试使用Spark Streaming从Kafka获取一些数据，并对这些数据进行一些操作。为此，我应该知道获取的数据的模式，有什么方法可以做到这一点，或者我们可以通过使用字段名从流中获取值吗？
浏览 0提问于2017-05-29得票数 1
我的火花流版本是2.0，kafka版本是0.10.0.1，火花流-kafka-0-10_2.11。我使用直接的方式获得卡夫卡的记录，我现在想限制我在一批中获得的信息的最大数量。那么火花流中记录的最大数量是max.poll.records*consumers？
浏览 1提问于2018-09-27得票数 2
我已经设置了星火结构化流(Spark2.3.2)来阅读Kafka (2.0.0)。如果消息在星火流作业启动前进入主题，则无法从主题开始使用。使用以下命令启动spark：spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.2.3.1.0.0-78 --repositories("10 s
浏览 1提问于2019-06-19得票数 8
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
