Spark structured streaming无权访问组
Spark structured streaming是Apache Spark的一个模块，用于处理实时流数据。它提供了一种简单且高效的方式来处理连续的数据流，并将其转换为结构化的数据。
Spark structured streaming的主要特点包括：
- 高级API：它提供了一套高级的API，使开发人员可以使用类似于批处理的方式来处理实时数据流。开发人员可以使用SQL、DataFrame和Dataset API来定义流式查询。
- Exactly-once语义：Spark structured streaming支持精确一次的语义，确保每条记录只会被处理一次，避免了数据重复或丢失的问题。
- 容错性：Spark structured streaming具有强大的容错性，可以处理节点故障或数据丢失的情况，并能够自动恢复和重新计算丢失的数据。
- 扩展性：它可以在大规模集群上运行，并且可以根据数据量的增长自动扩展计算资源，以满足实时数据处理的需求。
Spark structured streaming适用于许多实时数据处理场景，包括：
- 实时数据分析和报表：可以使用Spark structured streaming来处理实时生成的数据，并将其转换为结构化的数据，以便进行实时分析和生成报表。
- 实时监控和警报：可以使用Spark structured streaming来监控实时数据流，并根据预定义的规则生成警报或触发其他操作。
- 实时推荐系统：可以使用Spark structured streaming来处理用户行为数据，并实时生成个性化的推荐结果。
- 实时数据清洗和转换：可以使用Spark structured streaming来清洗和转换实时数据流，以满足特定的业务需求。
腾讯云提供了一系列与Spark structured streaming相关的产品和服务，包括：
- 腾讯云数据流计算（Data Flow）：提供了基于Spark structured streaming的实时数据处理服务，支持高可用、弹性扩展和容错性。
- 腾讯云流计算Oceanus：提供了一站式的流式计算平台，支持Spark structured streaming等多种计算引擎，适用于实时数据处理和分析场景。
更多关于Spark structured streaming的信息，可以参考腾讯云的官方文档：
相关·内容
我的理解是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
我正在尝试通过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 streaming编程中，我们可以通过设置以下配置来显式分配kafka消费者组id： val kafkaParams = Map[String, Object](
...
"group.id" -> "use_a_separate_group_id_for_each_stream",
...
)
val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent) 在spark stru
浏览 11提问于2020-03-31得票数 1
回答已采纳
目前在阅读flink的文档时，我发现flink中有很多支持，比如翻滚窗口、滑动窗口、会话窗口和自定义窗口的实现
谁能告诉我现在或将来有多少窗口语义是在spark structured或spark streaming中内置或定义的？是否有可能在spark structured streaming中的上实现一个应用程序？
谢谢你的点子和建议~
浏览 0提问于2018-11-05得票数 0
我经常使用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 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
我们每15分钟就会收到几百万条记录。在spark structured streaming中，将相同in的当前记录集与以前的记录集连接的最佳方式是什么？如何在重启后重新初始化之前的状态？我们已经尝试了HBase来存储之前的状态，但是它变得非常慢。如果我们使用spark任意会话，如何在重启后重新初始化以前的状态？我们现在已经在Kafka streams中实现了这一点。但是想知道在spark structured streaming中是否有实现的方法。
浏览 2提问于2020-07-15得票数 2
我使用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
我尝试将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
正在尝试测试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
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
我想调用一个web服务来获取Spark Structured Streaming中的一些数据。有可能吗？多么?
浏览 0提问于2019-12-07得票数 5
注意:我使用的是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
我想使用Spark structured来聚合从RabbitMQ消费的数据。我知道与apache kafka有官方的spark结构化流媒体集成，我想知道是否也存在与RabbitMQ的集成？由于我不能切换现有的消息传递系统(RabbitMQ)，我想使用kafka-connect在消息传递系统(Rabbit to kafka)之间移动数据，然后使用Spark structured streaming。
有人知道更好的解决方案吗？
浏览 0提问于2018-04-16得票数 0
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 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
回答已采纳
我想使用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
我正在使用Spark structured来处理Kafka队列中的大量数据，并执行一些繁重的ML计算，但我需要将结果写入Elasticsearch。
我尝试使用ForeachWriter，但无法在其中获取SparkContext，另一种选择可能是在ForeachWriter中执行HTTP Post。
现在，我正在考虑写我自己的ElasticsearchSink。
有没有为Spark Structured streaming创建Sink的文档？
浏览 8提问于2017-02-24得票数 2
回答已采纳
给定一个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
回答已采纳
显然在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的结构化流功能，但我需要在一个窗口内做一个reduce by键。基本火花流支持一种称为reduceByKeyAndWindow的方法。我对Spark Structured Streaming文档进行了大量的搜索，但没有找到任何与Spark Streaming的reduceByKeyAndWindow相当的文档。
有没有什么结构化流式操作的组合可以让我在一个窗口中执行reduce by键？
浏览 0提问于2018-04-14得票数 2
1回答
在遗留的spark流中，有一个度量标准，您可以看到该特定批处理正在处理多少条记录。有没有办法在spark structured streaming (3.x.x)指标中显示类似的东西？ ?
浏览 14提问于2021-07-23得票数 0
是否可以使用Kinesis streams作为Spark structured streaming的数据源？我找不到任何可用的连接器。
浏览 11提问于2018-08-26得票数 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回答
我正在运行几个使用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
我想将Java Spark SQL DataFrames转换为结构化流式DataFrames，这样每一批都会统一到结构化流式DataFrame。因此，我可以在DataFrames上使用从批处理源获得的Spark Structured Streaming功能(例如连续作业)。
浏览 24提问于2021-08-22得票数 0
回答已采纳
当涉及到实现机器学习算法时，标准化/规范化数据即使不是关键，也是一个必不可少的点。使用Spark structured streaming以实时的方式做到这一点，是我在过去几周里一直在努力解决的问题。
在历史数据上使用StandardScaler估计器((value(i)-mean) /standard deviation)被证明是很好的，在我的用例中是最好的，以获得合理的聚类结果，但我不确定如何将StandardScaler模型与实时数据相匹配。结构化流式传输不允许这样做。任何建议都将不胜感激！
换句话说，如何在Spark structured streaming中适应模型？
浏览 0提问于2017-05-20得票数 4
我的用户名中有空格，当使用kafka运行spark structured streaming时，它无法找到.ivy2文件夹和相应的.jar文件的路径。
浏览 51提问于2020-04-24得票数 1
如何在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
如何使用"sourceArchiveDir“和"cleanSource=archive”将源CSV文件移动到归档目录？我正在运行下面的代码，但它不移动源文件，但是流处理工作正常，即它将源文件内容打印到控制台。 import org.apache.spark.sql.streaming.{OutputMode, Trigger}
val inputPath =
"/<here is an absolute path to my project dir>/data/input/spark_full_delta/2021-06-21"
spar
浏览 88提问于2021-07-02得票数 1
1回答
结构化流式传输指标性能？
、、、
在尝试了一些监视结构化流性能和输入/输出指标的方法后，我发现一种可靠的方法是附加streamingQueryListener来输出streamingQueryProgress，以获得输入/输出数量。
除了SparkUI，
有没有更好的方法来监控结构化流性能？
将queryProgress输出到文件或Kafka的最好方法是什么？
在spark streaming和spark structured之间比较性能(速度、输入、输出记录)的有效方法是什么？
浏览 1提问于2018-05-19得票数 2
我正在尝试使用以下文档向kafka接收器写入内容：https://docs.databricks.com/spark/latest/structured-streaming/avro-dataframe.html#example-with-schema-registry 但是，我在spark-avro(https://mvnrepository.com/artifact/org.apache.spark/spark-avro_2.11)中找不到有多个参数的to_avro函数。是否有人知道它是否已被移动/重命名，或者我是否应该使用其他库？谢谢!
浏览 19提问于2020-08-06得票数 0
回答已采纳
1回答
我在Spark 2.3.1中使用Spark Structured Streaming，下面是我的代码：
val sparkSession = SparkSession
.builder
.appName("xxx")
.config("spark.serializer",
"org.apache.spark.serializer.KryoSerializer")
.config("spark.rpc.netty.dispatcher.numThreads", "2")
.config("spark.s
浏览 1提问于2018-12-13得票数 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
我有一个像这样的小溪 +------+-------------------+------+
|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(3.0.0)结构化流从kafka读取主题。
我使用了joins，然后使用了mapGropusWithState来获取流数据，因此，根据我对火花官方指南：的理解，我不得不使用更新模式。
下面的火花官方指南部分没有提到DB sink，它也不支持为update mode：写到files
目前，我将其输出到console，并希望将数据存储在文件或DB中。
因此，我的问题是:在我的情况下，如何将流数据写入数据库或文件？我是否必须将数据写入kafka，然后使用kafka连接将数据读回文件/db？
附注：我跟踪了这些文章以获得aggregated流查询。
- https://stackov
浏览 5提问于2020-09-16得票数 0
有返回CSV数据的url。此url的实际数据每10分钟更新一次。我是否需要通过这个url手动下载数据到文件中，然后通过Apache Spark加载这个文件，或者例如，我可以将Spark指向这个url，Spark会不时地通过这个url自动重新下载数据并处理它？例如，是否可以使用Spark Structured Streaming？
浏览 7提问于2020-04-09得票数 0
代码： df_streaming = spark \
.readStream \
.format("kafka") \
... \
.load() \
.xxx()
df_streaming = df_streaming \
.groupBy(["name", "height"]) \
.apply(cal_feature)
stream_writer = df_streaming \
.writeStream \
.format("console")
浏览 8提问于2021-06-16得票数 0
我很难解决以下异常：“必须使用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
回答已采纳
下面是我在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
我设计了一个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 structured streaming中实现Apache pulsar Sink时，我在spark中收到了一个任务不可序列化异常。 我已经尝试将PulsarConfig外推到一个单独的类，并在JDBC函数中调用它，这是我通常对.foreachPartition连接和其他集成到spark structured streaming中的系统所做的操作，如下所示： PulsarSink类 class PulsarSink(
sqlContext: SQLContext,
parameters: Map[Strin
浏览 78提问于2019-09-15得票数 0
回答已采纳
使用pyspark时，哪个选项的性能最好？使用map的UDF或RDD处理？ 我使用spark Structured streaming来使用数据，对于每个微批量，我将DF转换为RDD，并执行一些python graphkit操作，然后再次将RDD转换为DF以写入Kafka流。
浏览 23提问于2020-06-26得票数 1
1回答
我在Kubernetes上运行Spark 3.0应用程序(Spark Structured Streaming)，并尝试使用新的原生Prometheus指标接收器。我能够让它工作，并获得描述here的所有指标。 但是，我真正需要的指标是在启用以下配置时提供的指标: spark.sql.streaming.metricsEnabled，如this Spark Summit presentation中所建议的。现在，即使将该配置设置为"true"，我在/metrics/executors/prometheus下也看不到任何广告中的流指标。要注意的一件事是，我可以在metrics/
浏览 53提问于2020-10-20得票数 3
回答已采纳
我正在尝试从spark structured streaming连接到Kafka。
这是可行的：
spark-shell --master local[1] \
--files /mypath/jaas_mh.conf \
--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 \
--conf "spark.driver.extraJavaOptions=-Djava.security.auth.login.config=jaas_mh.conf" \
--
浏览 19提问于2018-08-25得票数 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
1回答
是否可以使用Snowflake表作为Databricks中spark结构流的源？当我运行以下pyspark代码时：
options = dict(sfUrl=our_snowflake_url,
sfUser=user,
sfPassword=password,
sfDatabase=database,
sfSchema=schema,
sfWarehouse=warehouse)
df = spark.readStream.format
浏览 10提问于2020-02-20得票数 0
我正在尝试使用安装了x-pack的Elasticsearch (ES) 6.1.1 Hadoop来使用Spark Structured Streaming 2.2.1写入数据。这是我的代码(索引已经存在于elastic中)：
val exceptions = spark
.readStream
.text(path)
val advancedQuery = exceptions
.writeStream
.format("org.elasticsearch.spark.sql")
.trigger(Trigger.ProcessingTime(10.seco
浏览 4提问于2018-01-17得票数 0
我们正在使用Qubole Kinesis Connector (jar)进行Spark structured streaming。这曾经工作得很好，但是突然抛出了一个错误"S3 filesystem not found“。
我们可以使用KCL，但我们需要对每个批次进行测试。是否有其他可用的Kinesis连接器可提供无缝写入操作？
浏览 0提问于2020-02-14得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
