Spark Structured Streaming允许使用unifrom DataFrame和Dataset API处理实时数据流。
Azure databricks 自动加载器 Spark Streaming 无法读取输入文件
我已经使用自动加载器功能设置了流作业，输入位于 azure adls gen2 中，采用 parquet 格式。下面是代码。 df = Spark.readStream.format("cloudFiles")\ .选择...
delta mergeSchema 无法将 MemoryStream 与 Spark 检查点结合使用
我正在使用 Spark 的 MemoryStream 测试 DeltaWriter 类来创建流（而不是 readStream），并且我想使用选项“mergeSchema”将结果作为增量文件写入 s3 上：...
我是结构化流媒体新手，希望根据 json 消息中的日期列创建分区列。 这是示例消息： {“日期”：“2022-03-01”，“c...
将数据从增量表写入事件中心的流作业在数据块中失败 - 超时问题
这里是我的代码，用于将数据从增量表写入事件中心（消费者团队将从这里消费数据）： 导入 org.apache.spark.eventhubs._ 导入 org.apache.spark.sql.streaming.Trigger._...
Trigger.AvailableNow 用于 PySpark (Databricks) 中的 Delta 源流查询
Databricks 文档中的所有示例均采用 Scala 语言。无法从 PySpark 找到如何使用此触发器类型。是否有等效的 API 或解决方法？
我有一个使用spark.readStream.format('delta')在pyspark中读取的流。数据由多列组成，包括类型、日期和值列。 示例数据框； 类型 日期 价值 1...
Pyspark：AnalysisException：具有流源的查询必须使用 writeStream.start() 执行；
我有多个想要运行的函数，但收到一条错误消息，提示我需要使用 .start()。有人可以告诉我如何继续吗？ 这是我的代码的一部分： def parse_data(df): 缓存Df = df。
Spark 流 - withWatermark() 具有重复行为
我正在从 Kafka 读取数据（最早开始偏移）并在控制台上写入数据进行一些测试。水印持续时间为 10 秒。遵循 Spark 文档 - https://spark.apache.org/
如何使用 PySpark 正常关闭 StructureStreaming？
作为主题，如何使用 PySpark 优雅地关闭 StructureStreaming？当我调用 Spark.streams.awaitAnyTermination(timeout=100) 时，它总是返回错误：ERROR MicroBatchExecution: Query [id = 97f...
我有一个关于如何处理微批次的问题。会火花 为所有可用的执行器获取多个微批次并并行处理多个微批次或 它只会得到一批...
Spark 流 leftOuter 无法与第三个流连接一起使用
我被这个流 leftOuter join 困住了。我能够流连接 2 个数据帧，并且在水印时间到期后可以获得空值。但如果我加入 3 个数据框，我就无法实现。 ...
在我的代码中有一个特定的要求，因此我必须在流数据帧上使用分区。 但给了我以下错误 流式 DataF 不支持非基于时间的窗口...
即使没有新数据，我们如何强制 Databricks 结构化流每隔几秒从事件中心读取数据？
我们使用 Databricks 结构化流从 azure 事件中心读取数据，并使用 forEachBatch 将数据更新插入到 writeStream 部分中的增量表中 问题是 - 事件...
Stream-Static Join：如何定期刷新（取消持久化/持久化）静态数据帧
我正在构建一个 Spark 结构化流应用程序，我正在其中进行批处理流连接。并且批量数据的来源会定期更新。 所以，我打算做一个坚持/取消坚持...
我在工作过程中遇到了一些模式演变，但我找不到让它发挥作用的方法。 上周，我在 ERP 系统中启用了 5 列，业务需要在...
Spark Streaming 正在从 Kafka 主题读取以及如何将嵌套 Json 格式转换为 dataframe
我能够从 Kafka 主题读取数据，并能够使用 Spark Streaming 在控制台上打印数据。 我希望数据采用数据帧格式。 这是我的代码： 火花 = SparkSession \ .
我有以下 JSON，我正在从 Kafka 读取它，然后尝试使用 from_json 函数转换为 StructType。 schema_session_start = StructType([ StructField("ID", StringType()),...
在 Spark UI 中，具有杀死活动运行阶段的功能： 当使用此按钮杀死阶段时，与该阶段关联的任务将被重新处理？或者他们会
环境：Spark 3.5.0、Scala 2.12.18、openjdk 11.0.20.1 我正在尝试从 Kafka 源流式传输数据，但从 Spark 中得到异常。 看来这通常是由于依赖版本所致
我在 s3 中有管道分隔的 csv 文件，我正在尝试将其加载到 Databricks 中。当使用下面的代码读取时，文件被正确读取： df_test = Spark.read.option("标题", True).