Spark Structured Streaming允许使用unifrom DataFrame和Dataset API处理实时数据流。
在我的结构化流作业中，我在updateAcrossEvents方法中更新Spark Accumulators，但是当我试图在StreamingListener中打印它们时，它们总是0。下面是代码。....
如何在kafka中摄取两个生产者的数据并使用spark结构化流加入？
我试图从两个kafka主题中读取数据，但我无法加入并找到最终的数据帧。我的kafka主题是CSVStreamRetail和OrderItems。 val spark = SparkSession .builder ...
我的目标是从redis中摄取流式数据并进行处理，如何通过spark结构化流连接和处理数据？我如何通过spark结构化流连接和处理数据？
我正在寻找从Spark数据管道访问数据的最佳方案。场景如下。我正在从Kafka主题中读取数据，创建一个流式数据框架，然后对其进行清理和... ...
当结构化流中使用的Spark DataFrame的底层数据在源端更新时，会发生什么？
我有一个用例，在这个用例中，我将一个流式DataFrame和一个静态DataFrame连接在一起。静态DataFrame是从一个parquet表（一个包含parquet文件的目录）中读取的。这个parquet数据是...
从Spark Streaming中获取异常 "没有注册输出操作，所以没有执行"。
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") ....
Spark *Structured* Streaming中的RecordTooLargeException异常
我一直收到这个错误信息。当序列化时，消息是1169350字节，这比你在max.request.size配置中配置的最大请求大小要大。由于...
我在我的结构化流作业中引发了一个自定义异常来测试失败，如下所示。我看到查询被终止，但不能理解为什么驱动脚本没有以非零的方式失败......。
假设我有一个流数据框A和一个大的静态数据框B，假设A的大小为< 10000条记录。然而，B是一个大得多的数据帧，大小在数百万条范围内。...
在Spark结构化流中对createOrReplaceTempView进行查询时出现错误。
下面是我在spark结构化流中的代码，in foreachBatch df.writeStream.trigger(Trigger.ProcessingTime("10秒")).foreachBatch((batchDF: DataFrame, batchId: Long) => { ...
错误。在kafka中使用Spark结构化流来读写数据到另一个主题。
我正在做一个小任务，使用kafka主题读取access_logs文件，然后我计算状态，并将状态的计数发送到另一个kafka主题。但是我一直收到错误信息，比如，当我使用no ...
在spark结构化流媒体中，Json字符串应该作为Kafka主题消耗，而不是模式。
我需要消耗Kafka主题，它为每一行产生动态Json字符串，我不能解析没有schema的Json字符串。在我的案例中，Schema可以是动态的。有什么办法可以将Kafka topic(value)转换为......？
为什么我的AWS Glue开发端点在尝试从Confluent Kafka Topic (RBACLDAP)读取时，会抛出NoClassDefFoundError错误？
当我试图通过AWS Glue Endpoint运行spark流作业时，我收到了一个java.lang.NoClassDefFoundError：orgapachekafkacommonsecurityauthAuthenticateCallbackHandler错误，该错误读取......
你好~我准备用Kafka+Spark结构化流开发一个小批量的程序，但我很困惑，到底是用Python还是Scala，哪个更快？但是我很困惑，到底是用python还是scala，哪个更快。如果有什么好 ...
Spark结构化流使用spark-acid writeStream（带检查点）抛出org.apache.hadoop.fs.FileAlreadyExistsException。
在我们的Spark应用中，我们使用Spark结构化流。它使用Kafka作为输入流，& HiveAcid作为Hive表的writeStream。对于HiveAcid，它是开源的库，叫做spark acid，来自qubole：... ...
在Spark结构化流中，在createOrReplaceTempView上写查询时遇到问题。
下面是我的代码在spark结构化流内foreachBatch df.writeStream.trigger(Trigger.ProcessingTime("10秒")).foreachBatch((batchDF: DataFrame, batchId: Long)=> { batchDF......。
我使用spark.readStream().text(
目前在Spark Structured Streaming中还没有对PubSub源的支持。有人在Spark Structured Streaming中写了一个自定义的源码来从PubSub中读取吗？这是个可行的方法吗？
使用Spark结构化流从多个Kafka主题读取并写入不同的汇的最佳方式是什么？
我正试图编写一个Spark结构化流作业，从多个Kafka主题（可能是100个）中读取，并根据主题名称将结果写入S3上的不同位置。我已经...
我从源卡夫卡读取了火花结构化的流数据帧。我想将此数据帧传递给函数，并将该函数的结果写入某个目标。案例类JsonSchema（...