spark-structured-streaming 相关问题

Spark Structured Streaming允许使用unifrom DataFrame和Dataset API处理实时数据流。

将一列中的 json 数据从 Apache Kafka 流数据扩展到数据帧 PySpark 中的多列时出现问题

亲爱的，我练习流数据。我决定创建 Apache Kafka。我想从 KafkaProducer 读取数据，然后我创建了一个用于流查询的 Kafka 源： df = 火花 \ 。读 \ .表格...

json apache-spark pyspark apache-kafka spark-structured-streaming

回答 1 投票 0

PySpark（Spark v3.4.1）结构化流如何实现累积聚合数据写入spark sink？

我正在从事 pyspark 结构化流编程，以生成流数据的一些累积聚合。我已经使用mongodb和kafka作为spark读取流数据。我已经尝试过 foreachBatch...

apache-spark pyspark spark-structured-streaming

回答 1 投票 0

PySpark 数据框转换 pyspark

我有一个下面的数据框，我需要将其转换如下。我正在使用 PySpark 3.4.1 。 +------------------------+------------------------ -------------------------------------------------- ---------...

apache-spark pyspark apache-spark-sql spark-streaming spark-structured-streaming

回答 1 投票 0

Spark 结构化流因 Kubernetes 中执行器内存不足而失败

我正在运行一个 Spark 结构化流作业，从对象存储桶中读取数据，进行一些转换和过滤，调用 groupby 和聚合。将写入内容发布到对象存储中我...

apache-spark amazon-s3 pyspark spark-structured-streaming minio

回答 1 投票 0

结构化流+Kafka集成numInputRows为0但偏移量增加

结构化流+ Kafka集成，spark版本3.3，我发现了一些奇怪的东西，一开始我可以得到正确的numInputRows，但是大约5次获取numInputRows将是0，bue endOff...

apache-spark apache-kafka spark-structured-streaming

回答 1 投票 0

pyspark 中的用例流

我正在 Azure 上使用 Databricks，我的数据托管在 ADLS2 上。当前运行时版本是 10.4 LTS（如果需要我可以升级）我有一个表 Pimproduct： ID 姓名行动 dlk_last_modified

python azure pyspark azure-databricks spark-structured-streaming

回答 1 投票 0

Spark 流使用 Abris 从 Kafka 读取，并且最新的架构注册表不会同步

我有一个 Spark 流，它从 Kafka Avro 消息中读取并根据最新版本的架构生成数据帧。我正在使用 abris 来执行此操作，看起来就像这样。导入 za.co.absa...

scala apache-spark apache-kafka spark-structured-streaming confluent-schema-registry

回答 1 投票 0

kafka消费者组在spark结构化流媒体中的意义

计划构建 Spark 结构化流应用程序，该应用程序从 Kafka 主题读取 json 数据，解析数据并写入任何存储。 val df = 火花 .readStream .format("卡夫克...

apache-spark apache-kafka spark-structured-streaming

回答 1 投票 0

spark scala kafka avro 反序列化器

我正在使用来自kafka的流数据帧。 value 列中的数据采用 avro 格式。我想将数据反序列化为结构类型。目前我已经编写了一个 udf 函数来调用该方法

scala apache-kafka deserialization avro spark-structured-streaming

回答 1 投票 0

使用 Apache Spark 进行结构化流处理

我可以在结构化流中创建多个批次，而其中只有一个包含连续数据的文件吗？我已经使用多个输入文件创建了多个批次。但是，现在我想生成多个

apache-spark pyspark spark-streaming spark-structured-streaming

回答 1 投票 0

使用 Spark 进行文件流传输

我正在使用来自 s3 的 Spark 文件流。代码运行良好，但对于第一批，所有文件都被视为目录中的有效文件。请帮助我是火花新手。我的代码如下。我正在使用来自 s3 的 Spark 文件流。代码运行良好，但对于第一批，所有文件都被视为目录中的有效文件。请帮助我是火花新手。我的代码如下。 <pre> Dataset<Row> df = spark.readStream().format("csv").schema(getOLRSchema()).load( "s3a://filePath/*.csv"); query = df.writeStream().foreachBatch(new VoidFunction2<Dataset<Row>, Long>() { @Override public void call(Dataset<Row> ds, Long batchId) throws Exception { long timestamp = System.currentTimeMillis(); ds.write().parquet("hdfs://ha-cluster/testStreaming/"); System.out.println("Time taken to complete: " + (System.currentTimeMillis() - timestamp)); } }) .option("checkpointLocation", "hdfs://ha-cluster/mnt/vol2/Ajit/checkpoint") .option("latestFirst", "true") .option("maxFilesPerTrigger", "32") .trigger(Trigger.ProcessingTime(120000)) .outputMode("append") .start(); query.awaitTermination(); </pre> Spark 文档解释说，只有新注意到的文件才会被处理，现有的文件将被忽略

java apache-spark amazon-s3 spark-structured-streaming

回答 1 投票 0

Spark 连续结构化流不显示输入速率或处理速率指标

我正在独立集群上运行我的 Spark 连续结构化流应用程序。但是我注意到平均输入/秒或平均进程/秒等指标没有在结构上显示（作为 NaN）...

apache-spark spark-streaming spark-structured-streaming data-engineering

回答 1 投票 0

无法在Jupyter笔记本上使用kafka jars

我正在使用 Spark 结构化流从单节点 Kafka 读取数据。在 Mac 上本地运行以下设置。我可以通过 Spark-Submit 阅读，但在 Jupyter Notebook 中不起作用。来自 pyspark.sql 我...

python apache-spark apache-kafka spark-structured-streaming

回答 1 投票 0

Databricks Notebook 中推断架构失败

我在Databricks中编写了一个spark结构化流。第一段代码是检查我的实体是否存在增量表。如果没有，则创建增量表。在这里，我想我们...