spark structured streaming和batch的接收器相同吗？
Spark Structured Streaming和批处理的接收器是不同的。
Spark Structured Streaming是一种基于Spark SQL引擎的流式处理框架，它允许以类似于批处理的方式处理连续的数据流。它提供了一种高级API，可以将流数据视为连续的表，并使用SQL查询、DataFrame和Dataset API进行处理。Structured Streaming支持事件时间和处理时间，并提供了容错和Exactly-Once语义。
批处理是一种离线处理方式，它将数据分成一批一批的进行处理。批处理的接收器通常是从文件系统、数据库或其他数据源中读取一批数据，并将其加载到Spark中进行处理。批处理的特点是数据有限且静态，处理过程是离线的。
因此，Spark Structured Streaming和批处理的接收器是不同的。Structured Streaming接收器可以从各种数据源（如Kafka、Flume、HDFS等）读取连续的数据流，而批处理接收器通常从静态的数据源（如文件系统、数据库等）读取一批数据。
相关·内容
我想使用Spark structured来聚合从RabbitMQ消费的数据。我知道与apache kafka有官方的spark结构化流媒体集成，我想知道是否也存在与RabbitMQ的集成？由于我不能切换现有的消息传递系统(RabbitMQ)，我想使用kafka-connect在消息传递系统(Rabbit to kafka)之间移动数据，然后使用Spark structured streaming。
有人知道更好的解决方案吗？
浏览 0提问于2018-04-16得票数 0
1回答
我正在运行几个使用google云存储上的Avro数据的batch Spark管道。我需要更新一些管道，使其更实时，并且想知道spark structured是否可以以流的方式直接消费来自gcs的文件，即parkContext.readstream.from(...)可以应用于从外部源的存储桶下连续生成的Avro文件。
Apache beam已经有像File.MatchAll().continuously()，Watch，watchnewFiles这样的东西，允许beam管道监控新文件并以流的方式读取它们(从而消除了对pubsub或通知系统的需要)，Spark structured stream
浏览 1提问于2018-01-05得票数 2
我使用spark(3.0.0)结构化流从kafka读取主题。
我使用了joins，然后使用了mapGropusWithState来获取流数据，因此，根据我对火花官方指南：的理解，我不得不使用更新模式。
下面的火花官方指南部分没有提到DB sink，它也不支持为update mode：写到files
目前，我将其输出到console，并希望将数据存储在文件或DB中。
因此，我的问题是:在我的情况下，如何将流数据写入数据库或文件？我是否必须将数据写入kafka，然后使用kafka连接将数据读回文件/db？
附注：我跟踪了这些文章以获得aggregated流查询。
- https://stackov
浏览 5提问于2020-09-16得票数 0
我正在使用Spark structured来处理Kafka队列中的大量数据，并执行一些繁重的ML计算，但我需要将结果写入Elasticsearch。
我尝试使用ForeachWriter，但无法在其中获取SparkContext，另一种选择可能是在ForeachWriter中执行HTTP Post。
现在，我正在考虑写我自己的ElasticsearchSink。
有没有为Spark Structured streaming创建Sink的文档？
浏览 8提问于2017-02-24得票数 2
回答已采纳
我正在尝试使用Spark Structured Streaming将聚合数据写入Kafka。下面是我的代码：
dataset
.writeStream()
.queryName(queryName)
.outputMode(OutputMode.Append())
.format("kafka")
.option("kafka.bootstrap.servers", kafkaBootstrapServers)
.option("topic", "topic")
.trigg
浏览 1提问于2018-04-10得票数 1
我使用spark-structured-streaming作为消费者从kafka获取数据，按照指南参考https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html 然后将数据保存到hdfs作为拼图文件。 这是我的问题:程序运行良好，但一些容器很少失败(但它确实发生了)，导致了一些损坏的拼接文件。它将导致错误，如不是拼图文件(长度太小: 4)或[.parquet不是拼图文件。期望的幻数在尾部80，65，82，49，但在读取它们时发现56，52，53，51]。我必须将它们移动到其他目录，并确保hi
浏览 20提问于2019-05-25得票数 0
我是个新手，正在学习spark structured，
我有以下使用complete作为输出模式的代码
import java.util.Date
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.StructType
object StreamingWordCount {
def main(args: Array[String]): Unit = {
val spark = SparkS
浏览 0提问于2018-07-27得票数 1
我曾尝试编写Spark MemoryStream单元测试用例，但我的测试用例程序中没有导入SharedSparkSession。
**import org.apache.spark.sql.test.SharedSparkSession
class MemoryStreamTest extends AnyFunSuite with SharedSparkSession {
....
}**
我的build.sbt文件配置如下
**scalaVersion := "2.12.0"
val sparkVersion = "3.0.0"
libraryDepe
浏览 0提问于2021-04-08得票数 0
我们每15分钟就会收到几百万条记录。在spark structured streaming中，将相同in的当前记录集与以前的记录集连接的最佳方式是什么？如何在重启后重新初始化之前的状态？我们已经尝试了HBase来存储之前的状态，但是它变得非常慢。如果我们使用spark任意会话，如何在重启后重新初始化以前的状态？我们现在已经在Kafka streams中实现了这一点。但是想知道在spark structured streaming中是否有实现的方法。
浏览 2提问于2020-07-15得票数 2
目前在阅读flink的文档时，我发现flink中有很多支持，比如翻滚窗口、滑动窗口、会话窗口和自定义窗口的实现
谁能告诉我现在或将来有多少窗口语义是在spark structured或spark streaming中内置或定义的？是否有可能在spark structured streaming中的上实现一个应用程序？
谢谢你的点子和建议~
浏览 0提问于2018-11-05得票数 0
我的理解是Spark structured Streaming是建立在Spark SQL之上的，而不是Spark streaming。因此，下面的问题是，应用于spark流的属性是否也适用于spark结构化流，例如：
spark.streaming.backpressure.initialRate spark.streaming.backpressure.enabled spark.streaming.receiver.maxRate
浏览 1提问于2018-09-03得票数 1
1回答
我们正在尝试用Apache Spark streaming替换Apache Storm。在storm中；我们根据“客户ID”对流进行分区，以便具有“客户ID”范围的消息将被路由到相同的螺栓(工人)。我们这样做是因为每个worker都会缓存客户详细信息(来自DB)。因此，我们分成4个分区，每个螺栓(工人)将拥有整个范围的1/4。
我确实看到了Spark和Storm的比较；这是对Spark的限制。我希望我们在Spark Streaming中有一个解决方案
浏览 1提问于2015-06-06得票数 1
如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？
我从Kafka获得消息流，我想对它应用Map操作，对于每个键，我想查询像Cassandra这样的数据存储，并获得该键的更多信息，并在流上应用进一步的操作。我如何使用Spark Structured Streaming 2.2.0来做到这一点？
浏览 2提问于2017-09-07得票数 1
我读到Spark Structured Streaming不支持将Kafka消息读取为JSON的模式推断。有没有一种方法可以像Spark Streaming一样检索模式：
val dataFrame = spark.read.json(rdd.map(_.value()))
dataFrame.printschema
浏览 1提问于2018-01-21得票数 13
我正在构建一个卡夫卡吞食模块在EMR 5.11.1，火花2.2.1。我的意图是使用结构化流从一个Kafka主题消费，做一些处理，并存储到EMRFS/S3的拼花格式。
控制台接收器按预期工作，文件接收器不工作。
在spark-shell中
val event = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", <server list>)
.option("subscribe", <topic>)
.load()
val eventdf
浏览 0提问于2018-04-09得票数 13
给定一个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 structured streaming。我想知道spark in batch模式是如何知道从哪个偏移量读取的？如果我指定"startingOffsets“为”最早的“，我只会得到最新的记录，而不是分区中的所有记录。我在两个不同的集群中运行了相同的代码。集群A(本地机器)获取了6条记录，集群B( TST集群-第一次运行)获取了1条记录。
df = spark \
.read \
.format("kafka") \
.option("kafka.bootstrap.servers",
浏览 0提问于2020-10-24得票数 0
4回答
如何在Python Spark structured streaming中使用foreach在输出上触发操作。
query = wordCounts\
.writeStream\
.outputMode('update')\
.foreach(func)\
.start()
def func():
ops(wordCounts)
浏览 0提问于2018-01-11得票数 2
在K8s群集上运行结构化流式处理时，我收到警告和错误。以下是我的部分代码： val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", brokerServers)
.option("subscribe", topicName)
.load()
val query = df.writeStream
.outputMode("append")
.foreachBatch
浏览 11提问于2019-10-12得票数 0
我有一个Spark Structured Streaming Job，它从S3读取数据，转换数据，然后将其存储到一个S3接收器和一个Elasticsearch接收器。
目前，我先执行readStream一次，然后执行writeStream.format("").start()两次。当这样做时，似乎Spark从S3源读取数据两次，每个接收器一次。
有没有一种更有效的方法来写入同一管道中的多个接收器？
浏览 1提问于2017-09-19得票数 14
1回答
我在Spark Streaming 2.2.1工作中有16个接收器。过了一段时间，一些接收器处理的记录越来越少，最终每秒只能处理一条记录。可以在屏幕截图上观察到这些行为：
虽然我知道根本原因可能很难找到并且不明显，但有没有方法可以进一步调试这个问题？目前我不知道从哪里开始挖掘。会不会和背压有关？
Spark流属性：
spark.app.id application_1599135282140_1222
spark.cores.max 64
spark.driver.cores 4
spark.driver.extraJavaOptions -XX:+PrintFlagsFin
浏览 0提问于2020-09-24得票数 2
显然在Spark streaming中没有对Cassandra接收器的内置支持。我在网上找到了这个例子，它基于ForEachWriter为Spark structured streaming实现了一个自定义的Cassandra接收器： https://dzone.com/articles/cassandra-sink-for-spark-structured-streaming 我知道我们需要创建一个ForeachWriter实现，它负责打开到接收器(Cassandra)的连接，写入数据并关闭连接。所以CassandraSinkForeach和CassandraDriver类是有意义的。 但
浏览 31提问于2019-02-24得票数 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
我想调用一个web服务来获取Spark Structured Streaming中的一些数据。有可能吗？多么?
浏览 0提问于2019-12-07得票数 5
1回答
是否有方法为星火结构化文件流源指定起始偏移量？
我正在尝试从HDFS源源不断地发送文件：
spark.sql("SET spark.sql.streaming.schemaInference=true")
spark.readStream
.parquet("/tmp/streaming/")
.writeStream
.option("checkpointLocation", "/tmp/streaming-test/checkpoint")
.format("parquet")
.opti
浏览 1提问于2018-07-18得票数 3
回答已采纳
我们有一个在YARN集群上运行的Spark Streaming application。
它接收来自Kafka topics的消息。
实际上，我们的处理时间超过了批处理间隔。
Batch Interval : 1 Minute
Processing Time : 5 Minutes
我想知道，如果在处理时间之间接收到一些数据会发生什么情况，在处理结束之前这些数据会在内存中可用吗？还是会在后续的数据获取中被覆盖？
我们使用Direct Streaming approach从Kafka主题中获取数据。
我应该使用Window based操作吗？例如，如果我有Window length as 5 M
浏览 10提问于2017-02-07得票数 6
回答已采纳
我想使用Apache Spark Structured Streaming和Kafka，Spark Structured Streaming支持Kafka 0.10及更高版本，我的Kafka集群使用kafka版本0.8.2.1。我想将当前kafka 0.8.2.1集群中的一些主题复制到基于2.2.0的新Kafka集群中。 为此，我尝试在Kafka 2.2.0集群上使用kafka-console-consumer来监听来自kafka集群0.8.2.1的消息，并将kafka-console-consumer的结果通过管道传输到kafka 2.2.0集群上的kafka-console-produc
浏览 17提问于2019-05-23得票数 1
我的计划是：
1. using spark streaming to load data from kafka every period like 1 minute.
2. convert the data loading every 1 min into DataFrame.
3. upsert the DataFrame into a Hive table (a table storing all history data)
目前，我成功地实现了步骤1-2。
我想知道是否有任何实用的方法来实现step3。详细说明：
1. load the latest history table with
浏览 0提问于2020-05-06得票数 0
我正在寻找一种在Spark中打印流媒体应用程序的执行计划的方法。我知道这是可能的print the plan of a SQL Spark application。但是，我想展示流应用程序的逻辑和物理计划。下面是我的应用程序： package org.sense.spark.app
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{HashPartitioner, SparkConf}
object TestStreamCombineByKey {
def ma
浏览 11提问于2019-09-30得票数 0
回答已采纳
我正在尝试连接我的结构化流式spark 2.4.5与kafka，但所有的时间，我尝试这个数据源提供程序出现错误。遵循我的scala代码和sbt构建： import org.apache.spark.sql._
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger
object streaming_app_demo {
def main(args: Array[String]): Unit = {
浏览 86提问于2020-05-04得票数 0
回答已采纳
我正在用Java/Scala编写一个自定义的Spark structured streaming源代码(使用v2接口和Spark 2.3.0)。
在测试与Spark offsets/checkpoint的集成时，我得到以下错误：
18/06/20 11:58:49 ERROR MicroBatchExecution: Query [id = 58ec2604-3b04-4912-9ba8-c757d930ac05, runId = 5458caee-6ef7-4864-9968-9cb843075458] terminated with error
java.lang.ClassCastExc
浏览 0提问于2018-06-20得票数 1
我尝试将MQTT与PySpark结构流结合使用。
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
spark = SparkSession \
.builder \
.appName("Test") \
.master("local[4]") \
.getOrCreate()
# Custom Structured Streaming re
浏览 5提问于2018-07-26得票数 3
我经常使用HDFS和Kafka，我注意到Kafka比HDFS更可靠。所以现在使用Spark-structured-streaming，我很惊讶检查点只有HDFS。用Kafka来勾点会更快更可靠。那么，有没有可能在没有HDFS的情况下使用spark structured？奇怪的是，我们只能在Kafka中将HDFS用于流数据。或者，有没有可能告诉斯帕克忘记ChekpPointing并在程序中管理它？ Spark 2.4.7 谢谢
浏览 39提问于2021-05-07得票数 2
正在尝试测试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
我想了解"spark.streaming.blockInterval"在星火流DirectAPI中扮演了什么角色，据我理解，"spark.streaming.blockInterval"用于计算分区，即#partitions = (receivers x* batchInterval) /blockInterval，但在DirectAPI中，火花流分区等于no。卡夫卡隔断。
"spark.streaming.blockInterval"是如何在DirectAPI中使用的？
浏览 0提问于2016-12-17得票数 0
回答已采纳
我们有使用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
package com.scala.sparkStreaming
import org.apache.spark._
import org.apache.spark.streaming._
object Demo1 {
def main(assdf:Array[String]){
val sc=new SparkContext("local","Stream")
val stream=new StreamingContext(sc,Seconds(2))
val rdd1=stream.textFileStream(&
浏览 2提问于2020-05-22得票数 1
我有一个像这样的小溪 +------+-------------------+------+
|group | time| label|
+------+-------------------+------+
| a|2020-01-01 10:49:00|red |
| a|2020-01-01 10:51:00|yellow|
| a|2020-01-01 12:49:00|blue |
| b|2020-01-01 12:44:00|red |
| b|2020-01-01 12:46:00|blue |
浏览 20提问于2021-03-25得票数 0
注意:我使用的是Spark 2.2.0。尝试从Zeppelin笔记本运行Scala代码时出现错误
%spark
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.streaming.{Time, Seconds, StreamingContext}
...
...
case class Record(time: Time, topic: String, count: Integer)
...
...
import org.apache.spark.stre
浏览 1提问于2017-12-07得票数 1
1回答
我正在尝试在Scala中运行一个Spark应用程序来连接到ActiveMQ。出于这个目的，我正在使用Bahir format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")。当我在我的built.sbt中使用Bahir2.2时，应用程序运行正常，但在将其更改为Bahir3.0或Bahir4.0时，应用程序无法启动，并且它给出一个错误： [error] (run-main-0) java.lang.NoClassDefFoundError: org/apache/hadoop/fs/FSDataIn
浏览 24提问于2020-12-09得票数 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
回答已采纳
如何在Spark Structured Streaming中获取Kafka header字段(在Kafka 0.11+中引入)？
我看到headers实现是在Spark 3.0中添加的，但在2.4.5中没有。我看到默认情况下spark-sql-kafka-0-10使用的是kafka-client 2.0。
如果不能使用Spark阅读Kafka的标题，那么你能推荐其他的替代方案吗？
浏览 0提问于2020-03-09得票数 0
我想了解一下Spark Structured Streaming的单元测试方面。我的场景是，我从Kafka获取数据，并使用Spark Structured Streaming使用它，并在数据上应用一些转换。
我不确定如何使用Scala和Spark来测试它。有人能告诉我如何使用Scala在结构化流媒体中进行单元测试吗？我还是个流媒体新手。
浏览 3提问于2019-07-05得票数 2
2回答
单源多水槽的结构化流
、、、、
我们正在创建一个Spark结构化流应用程序，其中有一个源和多个Sink (Kafka & HDFS)。我们在星火2.3.1集群上，因此在DataStreamWriter上使用foreachBatch是遥不可及的。因此，我选择多个Stream方法作为被提倡的和多个其他地方。不幸的是，这个流中只有一个在所有批处理中工作，而另一个仍然处于休眠状态。我在这里缺少什么来解决我的问题？
val spark = SparkSession
.builder
.appName("Spark-Structured-Streaming")
.enabl
浏览 3提问于2020-05-30得票数 1
回答已采纳
1回答
我启动了一个结构化流(Spark3.1.2)，它写到kafka (一个驱动程序，K8s上的5个执行器)
dataFrame
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "kafka-host:9092")
.option("topic", "my-topic")
当所有输入都被处理时，卡夫卡主题“my”包含了38087条新消息。所以38087行都写进了卡夫卡。度量被写入到Consol
浏览 7提问于2022-02-11得票数 1
2回答
我正在开发一个实时流应用程序，该应用程序可以从Kafka broker中轮询数据，并且我正在调整以前默认使用Spark结构化流的代码(带有微批处理)。但是，我不知道如何使用连续流而不是微批量流来获得类似的行为。这是一段可以工作的代码：
query = df.writeStream \
.foreachBatch(foreach_batch_func) \
.start()
这就是我到目前为止对连续流的尝试：
query = df \
.writeStream \
.foreach(example_func) \
.
浏览 1提问于2020-09-29得票数 5
我的用户名中有空格，当使用kafka运行spark structured streaming时，它无法找到.ivy2文件夹和相应的.jar文件的路径。
浏览 51提问于2020-04-24得票数 1
我遵循spark streaming + flume集成的guid。但我最终无法获得任何事件。()有人能帮我分析一下吗？在fume中，我创建了"avro_flume.conf“文件，如下所示：
描述/配置源
a1.sources = r1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 123.57.54.113
a1.sources.r1.port = 4141
描述接收器
a1.sinks = k1
a1.sinks.k1.type = avro
浏览 0提问于2015-04-18得票数 0
2回答
中提到的示例允许我在TCP流中接收数据包并监听端口9999。
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 core
浏览 5提问于2016-12-25得票数 3
回答已采纳
1回答
在遗留的spark流中，有一个度量标准，您可以看到该特定批处理正在处理多少条记录。有没有办法在spark structured streaming (3.x.x)指标中显示类似的东西？ ?
浏览 14提问于2021-07-23得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
