Spark Structured无法在卡夫卡中writeStream
Spark Structured是Apache Spark的一个模块，用于处理结构化数据。它提供了一种高级API，可以轻松地处理和分析各种数据源，包括文件、数据库和流式数据。
卡夫卡（Kafka）是一个分布式流处理平台，用于构建实时数据流应用程序和微服务。它具有高吞吐量、可扩展性和持久性的特点，被广泛应用于大规模数据处理和实时数据流处理场景。
在Spark Structured中，可以使用Kafka作为数据源进行读取和处理。但是，目前的版本中，Spark Structured无法直接将数据写入Kafka中的writeStream。这是因为Spark Structured的设计初衷是将数据写入可靠的数据存储系统，如文件系统或数据库，而不是消息队列系统。
如果需要将数据写入Kafka，可以考虑使用其他工具或库来实现。例如，可以使用Kafka的Java客户端或者Kafka Connect来将数据写入Kafka。另外，还可以使用Spark Streaming模块来处理流式数据，并将结果写入Kafka。
总结起来，Spark Structured无法直接在卡夫卡中使用writeStream写入数据，但可以通过其他工具或模块来实现将数据写入Kafka的需求。
相关·内容
1回答
我有一个要求，从多个卡夫卡topicsAvro的流媒体，并把他们在格林梅利与小修改的有效载荷。
Kaka主题定义为配置文件中的列表，每个Kafka主题都有一个目标表。
我正在寻找一个单一的星火结构应用程序和一个更新的配置文件，以听取新的主题或停止。听这个话题。
我在寻求帮助，因为我对使用单个查询还是多个查询感到困惑：
val query1 = df.writeStream.start()
val query2 = df.writeStream.start()
spark.streams.awaitAnyTermination()
或
df.writeStream.start().awaitA
浏览 2提问于2020-11-06得票数 1
回答已采纳
给定一个writeStream调用：
val outDf = (sdf.writeStream
.outputMode(outputMode)
.format("memory")
.queryName("MyInMemoryTable")
.trigger(Trigger.ProcessingTime(interval))
.start())
如何对sql运行MyInMemoryTable。
val df = spark.sql("""select Origin,Dest,Carrier,avg(DepDelay)
浏览 0提问于2019-07-16得票数 2
回答已采纳
1回答
我试图使用由kafka发布的Spark来消费数据，但我无法这样做。我用的是火花2.2。
我希望使用Spark使用Kafka发送的数据，处理它并存储在本地文件或HDFS中。
我想打印出来的数据由卡夫卡(由火花消费)在控制台运行火花作业。
对于卡夫卡，我遵循本教程：
[cloudera@quickstart kafka]$ ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>message 1
>message 2
>message
浏览 0提问于2017-08-21得票数 1
我很难解决以下异常：“必须使用writeStream.start();；kafka执行流源查询”
我的代码如下：
val spark = SparkSession
.builder()
.getOrCreate()
val bootstrapServers = "localhost:9092"
val topicName = "name"
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.
浏览 3提问于2021-05-28得票数 3
回答已采纳
1回答
如何将选定的专栏写到卡夫卡主题？
、、、、
我使用的是星星之火-SQL2.4.1v和java1.8。和卡夫卡版本火花-sql-卡夫卡-0-10_2.11_2.4.3和卡夫卡-客户_0.10.0.0
StreamingQuery queryComapanyRecords =
comapanyRecords
.writeStream()
.format("kafka")
.option("kafka.bootstrap.servers",KAFKA_BROKER)
浏览 6提问于2019-11-25得票数 1
回答已采纳
我试着阅读“火花”中卡夫卡(第10版)的信息，并试图打印出来。
import spark.implicits._
val spark = SparkSession
.builder
.appName("StructuredNetworkWordCount")
.config("spark.master", "local")
.getOrCreate()
val ds1 = sp
浏览 10提问于2016-11-15得票数 31
正在尝试测试Spark Structured Streams ...and失败...怎样才能正确地测试它们？
我遵循了的通用Spark测试问题，我最接近的尝试是[]，看起来像这样：
import simpleSparkTest.SparkSessionTestWrapper
import org.scalatest.FunSpec
import org.apache.spark.sql.types.{StringType, IntegerType, DoubleType, StructType, DateType}
import org.apache.spark.sql.streaming
浏览 1提问于2018-03-28得票数 6
1回答
df.writeStream
.foreachBatch((batchDF: DataFrame, batchId: Long) =>
batchDF.write
.format("org.apache.spark.sql.cassandra")
.cassandraFormat(cassandraTable, cassandraKeyspace, cassandraCluster)
.mode("append")
.save())
浏览 10提问于2020-08-21得票数 2
回答已采纳
我用的是火花2.3.2。
我收到了卡夫卡的数据。我必须对同一个数据进行多个聚合()。然后，所有聚合结果将转到相同的数据库(列或表可能被更改)。例如：
val kafkaSource = spark.readStream.option("kafka") ...
val agg1 = kafkaSource.groupBy().agg ...
val agg2 = kafkaSource.groupBy().mapgroupswithstate() ...
val agg3 = kafkaSource.groupBy().mapgroupswithstate() ...
但是，当我
浏览 0提问于2019-03-19得票数 2
回答已采纳
正在尝试创建spark数据流writeStream函数的测试，如下所示：
SparkSession spark = SparkSession.builder().master("local").appName("spark
session").getOrCreate()
val lakeDF = spark.createDF(List(("hi")), List(("word", StringType, true)))
lakeDF.writeStream
.trigger(Trigger.Once)
.format
浏览 0提问于2018-07-19得票数 8
在将CDC数据合并到基本增量表之前，我有以下脚本使用spark structured streaming读取CDC数据。 streamDf = spark \
.readStream \
.format('csv') \
.option("mergeSchema", "true") \
.option('header', 'true') \
.option("path", CDCLoadPath) \
.load()
streamQuery = (
浏览 35提问于2021-07-08得票数 0
回答已采纳
我正在尝试使用Spark/Python收集Azure Eventhub消息。每次，我都会得到异常"StreamingQueryException:输入字节数组有错误的4字节结束单元“ 有什么想法吗？ conf = {}
conf["eventhubs.connectionString"] = "Endpoint=sb://XXXXXXXXX.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXX=;EntityPath
浏览 20提问于2020-10-23得票数 1
回答已采纳
我需要使用Kafka主题，它为每一行生成动态Json字符串，我无法在没有模式的情况下解析Json字符串。在我的例子中，Schema可以是动态的。spark.read.json可以推断json schema.But，它需要"DATASET“或"JSON文件”。
有没有办法把卡夫卡主题(价值)转换成数据集？这样我就可以使用接受DATASET作为输入的spark.read.json，来解析json的模式。
但如果我用下面的代码。
val klines = spark.
readStream.
format("kafka").
option("kaf
浏览 1提问于2020-06-04得票数 0
回答已采纳
4回答
如何在Python Spark structured streaming中使用foreach在输出上触发操作。
query = wordCounts\
.writeStream\
.outputMode('update')\
.foreach(func)\
.start()
def func():
ops(wordCounts)
浏览 0提问于2018-01-11得票数 2
有一个CSV文件的数据湖，这些文件全天都在更新。我正在尝试使用Trigger.Once功能创建一个Spark Structured作业，以定期将新数据写入到Parquet数据湖中的CSV数据湖中。
这就是我所拥有的：
val df = spark
.readStream
.schema(s)
.csv("s3a://csv-data-lake-files")
下面的命令将所有数据写入到拼图湖中，但在写入所有数据后并未停止(我必须手动取消该作业)。
processedDf
.writeStream
.trigger(Trigger.Once)
.forma
浏览 13提问于2017-08-16得票数 12
2回答
我正在使用spark structured streaming从kafka主题中读取事件，并对其进行处理并将其写入到parquet中。我必须根据我在事件中获得的键将输出写入不同的文件夹。我尝试使用结构化流式处理示例始终指向特定的文件夹.Do我需要为每个文件夹启动一个流吗？
df.writeStream.format("parquet").option("path", "path/to/destination/dir").start()
浏览 2提问于2018-08-13得票数 0
2回答
我正在使用Spark Structured streaming；我的DataFrame具有以下模式
root
|-- data: struct (nullable = true)
| |-- zoneId: string (nullable = true)
| |-- deviceId: string (nullable = true)
| |-- timeSinceLast: long (nullable = true)
|-- date: date (nullable = true)
如何使用Parquet格式编写writeStream并写入数据(包
浏览 0提问于2018-03-04得票数 2
我试图从kafka获得数据以激发结构化流，但我无法检查我是否做得好。我想在控制台上打印卡夫卡的数据，但是控制台上什么都没有。这可能是因为卡夫卡大量的数据，但我不知道。
我使用的是Windows 10，我检查了卡夫卡的端口是由"netstat -an \ findstr TARGET_IP“创建的。TARGET_IP的意思是卡夫卡制片人的IP。根据上面的结果，我检查了"tasklist /FI "PID eq 5406"“。5406是java.exe的PID，用于PID 5406的内存不断增加。
public static void main( String[]
浏览 0提问于2019-07-11得票数 1
回答已采纳
我实现了一个火花作业，在结构化流中读取kafka主题的流。
val df = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "mykafka.broker.io:6667")
.option("subscribe", "test-topic")
.option("kafka.security.protocol", "SASL_SSL")
.option("kafka
浏览 4提问于2020-09-18得票数 2
回答已采纳
我在运行这段代码。streamingInputDF是卡夫卡主题的readStream。我运行的是spark 2.4.0和Kafka 0.10.2
val activityQuery = streamingInputDF.writeStream
.format("console")
.outputMode("append")
.start()
.awaitTermination()
当writeStream在streamingInputDF上被调用时，我看到卡夫卡代理抛出了这个异常
[2018-11-24 21:46:42,842] ERROR Closing s
浏览 0提问于2018-11-25得票数 0
回答已采纳
我想举一个很简单的例子。我有一个卡夫卡readStream阅读卡夫卡的主题。我运行的是spark 2.4.0和Kafka 0.10.2
var streamingInputDF =
spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "twitter-topic")
.load()
控制台writeStream
val
浏览 0提问于2018-11-24得票数 0
回答已采纳
我正在使用spark structured从kafka读取数据。
val readStreamDF = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", config.getString("kafka.source.brokerList"))
.option("startingOffsets", config.getString("kafka.source.startingOffsets"))
.optio
浏览 1提问于2019-08-02得票数 0
3回答
我正在尝试转向星火流数据集(结构化流)，但我得到了一个AnalysisException (摘录如下)。
有人能确认在结构化流(Spark2.0)中确实不支持旋转吗？
线程“主”org.apache.spark.sql.AnalysisException中的异常:必须使用writeStream.start()执行流源查询；卡夫卡在org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.org$apache$spark$sql$catalyst$analysis$UnsupportedOperationChecke
浏览 9提问于2017-12-01得票数 8
回答已采纳
只需设置一个hadoop/kafka/星火，一个节点演示环境。在pyspark中，我尝试读取(.readStream) Kafka消息，并在hadoop中将它写入(.writeStream)到json文件。奇怪的是，在hadoop "output/test“目录下，我可以找到一个创建的json文件，但只有一个消息。卡夫卡的所有新消息都不会更新json文件。但是我想把卡夫卡的所有信息都存储在一个json文件中。
我尝试过控制台(writeStream.format(“控制台”)或kafak(writeStream.format("kafka"))的接收器类型，它正常工作。
浏览 2提问于2022-08-27得票数 0
我目前正在制作一个使用Spark Structured Streaming的原始日志数据聚合器。
Inputstream由一个文本文件目录构成：
// == Input == //
val logsDF = spark.readStream
.format("text")
.option("maxFilesPerTrigger", 1)
.load("input/*")
然后解析日志...
// == Parsing == //
val logsDF2 = ...
..。和聚合
// == Aggregation == //
v
浏览 1提问于2018-09-11得票数 4
1回答
我有一个由卡夫卡流创造的数据。我想把它简化为一个值，然后在我的程序中使用这个值。
```scala
导入sparkSession.implicits._
val = sparkSession
.readStream
.format(“卡夫卡”)
.option("kafka.bootstrap.servers“)
.option(“订阅”，"theTopic")
.load()
val结果= df
.selectExpr(“将值转换为字符串)为json")
.map(json => getAnInt(Json)
.reduce { (x，y) =>
浏览 0提问于2019-07-20得票数 0
回答已采纳
1回答
从火花流数据提取数据
、、、、
我刚开始玩星火流。
我从卡夫卡那里得到了一个类似于下面的事件。我必须从dataframe中提取路径，从路径中读取数据并将其写入目的地。
{"path":["/tmp/file_path/file.parquet"],"format":"parquet","entries":null}
知道如何提取路径并格式化火花流数据吗？
这就是我想要达到的目标，
val df: DataFrame = spark.readStream.format("kafka").
option("kafka.
浏览 5提问于2022-11-14得票数 0
我们需要在卡夫卡源代码中使用maxOffsetsPerTrigger，在结构化流中使用Trigger.Once()，但是基于这个，似乎在spark 3中读到了allAvailable。在这种情况下，有什么方法可以达到速率限制吗？
下面是spark 3中的示例代码：
def options: Map[String, String] = Map(
"kafka.bootstrap.servers" -> conf.getStringSeq("bootstrapServers").mkString(","),
"subscribe
浏览 6提问于2020-12-02得票数 1
回答已采纳
我正在寻找一个解决方案，写入火花流数据到卡夫卡。我正在使用以下方法向kafka写入数据
df.selectExpr("to_json(struct(*)) AS value").writeStream.format("kafka")
但我的问题是，在写给卡夫卡的时候，数据显示如下
{"country":"US","plan":postpaid,"value":300}
{"country":"CAN","plan":0.0,"value"
浏览 0提问于2019-03-09得票数 0
我设计了一个Nifi流，将以Avro格式序列化的JSON事件推送到Kafka topic中，然后尝试在Spark Structured streaming中消费它。
虽然Kafka part运行良好，但Spark Structured streaming无法读取Avro事件。它失败，并出现以下错误。
[Stage 0:> (0 + 1) / 1]2019-07-19 16:56:57 ERROR Utils:91 - Aborting task
org.apache.avro
浏览 19提问于2019-07-19得票数 0
我对Spark-Streaming和Kafka有意见。在运行示例程序从Kafka主题消费并将微批结果输出到终端时，当我设置选项时，我的作业似乎挂起了： df.option("startingOffsets", "earliest") 从最新的偏移量开始工作很好，结果随着每个微批次的流过而打印到终端。 我在想，也许这是一个资源问题--我正在尝试从一个有相当多数据的主题中阅读。但是，我似乎没有内存/cpu问题(使用本地*集群运行此作业)。这项工作似乎从来没有真正开始过，而只是悬而未决： 19/09/17 15:21:37 INFO Metadata: Cluste
浏览 21提问于2019-09-18得票数 4
回答已采纳
1回答
我在Spark Structured streaming中读取Kafka中的事件，需要一个接一个地处理事件并写入redis。我为此编写了一个UDF，但它给了我spark上下文错误。 conf = SparkConf()\
.setAppName(spark_app_name)\
.setMaster(spark_master_url)\
.set("spark.redis.host", "redis")\
.set("spark.redis.port", "6379")\
.set("spark.redis.auth&#
浏览 10提问于2020-07-13得票数 0
回答已采纳
我正在尝试在更新输出模式下使用spark structured写入文件。我找到了，只要配置了控制台格式，它就能正常工作。但如果我将输出模式更改为：
val query = sessionUpdates
.writeStream
.outputMode("update")
.format("json")
.option("path", "/work/output/data")
.option("checkpointLocation", "/work/output/checkpoint&#
浏览 12提问于2018-02-28得票数 3
我正在尝试通过spark structured streaming从Kafka中读取数据。但是，在Spark 2.4.0.中，您不能为流设置组id (参见How to set group.id for consumer group in kafka data source in Structured Streaming?)。 然而，由于没有设置，spark只是生成组Id，而我停留在GroupAuthorizationException： 19/12/10 15:15:00 ERROR streaming.MicroBatchExecution: Query [id = 747090ff-12
浏览 166提问于2019-12-10得票数 1
回答已采纳
我们有使用Spark Structured实现的流应用程序，它试图从Kafka主题读取数据并将其写入HDFS位置。
有时应用程序会失败，并出现异常：
_spark_metadata/0 doesn't exist while compacting batch 9
java.lang.IllegalStateException: history/1523305060336/_spark_metadata/9.compact doesn't exist when compacting batch 19 (compactInterval: 10)
我们无法解决此问题。
我找到的唯一解
浏览 0提问于2019-05-31得票数 6
1回答
我正在实现一个用例来试用Spark Structured Streaming API。源数据是从Kafka主题中读取的，经过一些转换后，结果会写入控制台。
我希望打印中间输出以及结构化流查询的最终结果。
以下是代码片段：
val trips = getTaxiTripDataframe() //this function consumes kafka topic and desrialize the byte array to create dataframe with required columns
val filteredTrips = trips.filter(col
浏览 4提问于2019-07-31得票数 1
回答已采纳
1回答
我想要记录从spark结构化数据流的传入流中读取到数据库的记录数量。我正在使用foreachbatch来转换传入的流批处理，并将其写入所需的位置。如果在特定的小时内没有记录，我想记录读取的0条记录。但是，当没有流时，foreach批处理不会执行。有人能帮我吗？我的代码如下： val incomingStream = spark.readStream.format("eventhubs").options(customEventhubParameters.toMap).load() val query=incomingStream.writeStream.foreachBatc
浏览 10提问于2020-06-22得票数 1
1回答
我从Spark Structured Streaming应用程序获得表单的状态消息：
18/02/12 16:38:54 INFO StreamExecution: Streaming query made progress: {
"id" : "a6c37f0b-51f4-47c5-a487-8bd269b80142",
"runId" : "061e41b4-f488-4483-a290-403f1f7eff03",
"name" : null,
"timestamp" :
浏览 1提问于2018-02-12得票数 3
回答已采纳
1回答
我的代码类似于
df = spark.readStream.option("header","true") \
.schema(df_schema)\
.csv(df_file)
df2 = df.filter(df.col == 1)
df3 = df2.withColumn("new_col", udf_f(df2.some_col))
dfc = df3.where(df3.new_col == 2).count()
query = dfc.writeStream.outputMode("append").f
浏览 8提问于2018-03-02得票数 1
下面是我在foreachBatch中使用spark structured streaming的代码
df.writeStream.trigger(Trigger.ProcessingTime("10 seconds")).foreachBatch((batchDF: DataFrame, batchId: Long) => {
batchDF.persist
batchDF.createOrReplaceTempView("all_notifis");
batchDF.write.mode(SaveMode.
浏览 0提问于2020-05-28得票数 0
4回答
我想读一下卡夫卡的话题，然后用火花流的方式把它写到kudu表上。 我的第一种方法 // sessions and contexts
val conf = new SparkConf().setMaster("local[2]").setAppName("TestMain")
val sparkSession = SparkSession.builder().config(conf).getOrCreate()
val sparkContext = sparkSession.sparkContext
val kuduContext = new KuduConte
浏览 51提问于2017-10-26得票数 2
回答已采纳
嗨，我是从卡夫卡主题中读到的，我想要处理从卡夫卡收到的数据，例如，对未完成的数据进行过滤，删除停止词，最后再写回另一个卡夫卡主题。
// read from kafka
val readStream = existingSparkSession
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", hostAddress)
.option("subscribe", "my.raw") // Always rea
浏览 2提问于2021-03-14得票数 3
我很难使用Spark Structured Streaming在kafka主题中读取我的JSON数据。 上下文： 我正在构建一个简单的管道，在这里我使用kafka从MongoDb读取数据(这个数据库通常是从另一个应用程序中填充的)，然后我想在Spark中获取这些数据。 为此，我使用了Spark Structured Streaming，这似乎是可行的。 下面是我的代码： import org.apache.spark.rdd
import org.apache.spark.sql.avro._
import org.apache.spark.sql.{Column, SparkSession
浏览 13提问于2019-03-11得票数 0
回答已采纳
2回答
我是作为JSON从kafka_2.11-0.10.0.1到Spark2.1的流式计量器读取记录。我转向了结构化流媒体；尽管卡夫卡用户确认了输入的数据，但我的控制台和writeStream没有移动。我正在测试使用
pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
我的代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSe
浏览 0提问于2017-12-15得票数 3
回答已采纳
下面是我使用kafka和pyspark编写的第一个程序。代码似乎运行无异常，但我的查询的输出是空的。
我正在发起火花和卡夫卡。后来，在Kafka initiation中，我订阅了topic = "quickstart-events“，并从终端生成了该主题的消息。但是当我运行这段代码时，它给了我空白的数据帧。
我该如何解析？
代码：
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext, SparkSession, DataFrame
from pyspark.sql.types im
浏览 11提问于2021-09-12得票数 0
我一直收到这个错误消息：
The message is 1169350 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
正如在其他StackOverflow帖子中所指出的，我正在尝试在生产者中设置“max.request.size”配置，如下所示：
.writeStream
.format("kafka")
.option(
"kafka.bootstrap.
浏览 13提问于2020-05-26得票数 0
回答已采纳
4回答
我正在使用Spark Structured从Kafka队列中读取。读完卡夫卡之后，我在dataframe上应用了filter。我正在将这个经过过滤的数据帧保存到一个拼图文件中。这将生成许多空拼图文件。有没有什么方法可以让我不再写空文件？
df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", KafkaServer) \
.option("subscribe", KafkaTopics) \
.load
浏览 1提问于2017-09-27得票数 2
1回答
只显示前20行
、、
我有一个流式数据集。它从卡夫卡流中读取并写入控制台。
val outputStream = inputStream.writeStream.format("console")
.option("truncate", value = false)
.option("checkpointLocation", "checkpoint")
outputStream.start()
spark.streams.awaitAnyTermination()
浏览 2提问于2022-05-06得票数 0
回答已采纳
我正在更新模式下运行我的Spark结构化流作业，并且无法确定是否有可能获得每个更新的批处理ID。例如，当您以更新模式输出到控制台时，Spark将在输出时显示每个批号：
-------------------------------------------
Batch: 0
-------------------------------------------
...
-------------------------------------------
Batch: 1
-------------------------------------------
...
以此类推。我需要把同样的信息
浏览 0提问于2019-09-17得票数 0
回答已采纳
1回答
火花流写卡夫卡与延迟后x分钟
、、、、
我们有一个火花流应用程序。体系结构如下
从火花到卡夫卡。
Spark应用程序正在使用qubole/ Kinesis进行来自Kinesis的结构化流。然后将数据进行聚合，然后推送给卡夫卡。
我们的用例需要延迟4分钟才能进入卡夫卡。
加窗时间为2分钟，水印为4分钟。
val windowedCountsDF = messageDS
.withWatermark("timestamp", "4 minutes")
.groupBy(window($"timestamp", "2 minutes", "2 minute
浏览 0提问于2019-06-06得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
