spark-streaming 相关问题

Spark 中的 StreamQueryListener 不执行 onQueryProgress() 中的代码

我正在从 Databricks 增量表作为流读取数据并将其写入另一个增量表（使用屏幕截图中的控制台以便于调试），我想使用 StreamingQueryListener(...

Pyspark：通过服务主体向事件中心进行身份验证

我有一个服务主体，它有权读取 Eventhub 上的主题。我想阅读其中的主题并尝试以下操作： # 使用服务主体 (SAS) 的事件中心连接字符串

如何在 Spark 中动态应用数组列类型

我有一个 PySpark DataFrame，其中包含一个字符串列，其中包含结构为对象数组的 JSON 数据。但是，这些 JSON 对象的架构可能因行而异。 这是两个例子...

如何从 Spark 作业中检索通过 Dataproc 提交的作业的 jobId

我想获取在 Spark 上下文中运行的 Spark 作业的 jobId。 Dataproc 是否将此信息存储在 Spark Context 中？

Spark 3.0 - 从 MQTT 流读取数据

我想将数据流从基于 mosquitto 的 MQTT 主题读取到我的 Spark 3.0 应用程序中。我尝试通过以下方式使用 Bahir 库： 数据集 df = SparkSession .

Autoloader 未在流模式下拾取 .text 文件

我正在使用 Databricks Autoloader 以流（微批量）模式处理文件。源文件采用.text 格式。虽然创建了检查点并且流没有失败，但 Delta ta...

PySpark 流式传输与 AWS Kinesis Datastream 连接

我正在尝试将 AWS Kinesis Data Stream 读入 PySpark sql 数据帧。 这是我的Python代码 将 pyspark 导入为 ps 火花=（ ps.sql.SparkSession.builder .config(地图= { '火花...

没有记录被处理并且所有检查点文件数据不一致

我正在尝试使用 AWS Glue Streaming ETL 作业使用触发器进行读写。AvailableNow 与 Kinesis Data Streams 一起使用，就像我与 Kafka 一起使用一样，但没有处理任何记录，并且所有检查点文件都有

AWS Glue 流 + Kinesis + 触发器现已推出

我正在尝试使用 AWS Glue Streaming ETL 作业使用触发器进行读写。AvailableNow 与 Kinesis Data Streams 一起使用，就像我与 Kafka 一起使用一样，但没有处理任何记录，并且所有检查点文件都有

使用事件中心管理 PySpark Streaming 中的数据封装

使用 PySpark 流式传输数据时，我收到的主要消息封装在名为“body”的键中。 Spark.readStream.format("eventhubs").options(**ehConf).load() 难道是……

Spark 结构化流：流-流连接与不写入的聚合

平台：Databricks Notebooks|语言：PySpark 上下文：我正在尝试在流数据管道中构建一个节点，该节点评估所有存在的行数（例如 count(*) == Expected_co...

PySpark 流代码中的动态键/列重命名

我正在尝试动态重命名流数据中的列，但标准方法似乎不适用于流。 我的代码如下。 json_df = Spark.readStream.format("前夕...

在 Spark 结构化流处理期间重命名 Spark UI 中的 jobId

我能够使用 setJobDescription 重命名 SparkUI 中的作业描述，仅购买我的预处理数据被重命名（缓存数据集），但主要作业/阶段未重命名 例如： 我的缓存...

静态和流数据帧之间的 Spark 结构化流连接

我正在阅读有关 Spark 结构化流连接的信息，并遇到了静态到流数据集支持的连接类型。我有一个问题，为什么不支持它，因为据我所知...

隐藏 Spark 属性，使其不显示在 Spark Web UI 中，而不实现安全过滤器

位于 http://:4040 的应用程序 Web UI 在“环境”选项卡中列出了 Spark 属性。 将显示通过spark-defaults.conf、SparkConf 或命令行显式指定的所有值。 嗬...

在apache Iceberg 上的同一个表中合并多个流的问题

我在不同字段的同一个表中进行了多个火花流写入。 Iceberg 文档说了以下内容：Iceberg 支持使用乐观并发进行多个并发写入...

Spark 流作为事件处理/处理解决方案（微服务）

Spark 批处理为我们的业务带来了很多价值，因为它非常容易水平扩展（我们将 AWS EMR 与 YARN 结合使用）。 然而，我们最新的专有技术带来了新的挑战

PySpark StreamingQueryException：Elasticsearch Spark 连接器的 java.lang.NoSuchMethodError

我正在从事 PySpark 流作业，需要将流数据从 Kafka 写入 Elasticsearch。我正在使用： 火花版本：3.5.2 Elasticsearch Spark 连接器：org.elasticsearch:elasticsea...

错误 SparkContext：无法添加文件 java.io.FileNotFoundException：找不到 Spark 的 Jar

请根据我使用的代码帮助我修复上述错误 proccesing_data.py代码用于使用spark-streaming处理数据 导入日志记录 从 pyspark.sql 导入 SparkSession 来自 pys...

spark 结构化流 - 使用 availableNow 触发器从 kafka 读取

我尝试使用 Spark Stream API 从 Kafka 读取数据并将结果作为增量表写入 S3。对我来说，在 S3 上放置更少的对象很重要，因此我使用 coalesce(2) 在每个批次中创建两个对象。