使用Spark structured streaming仅保留最新数据

文章/答案/技术大牛

发布

1回答

apache-spark、pyspark、spark-streaming、databricks

我像这样流式传输数据：time，id，value我只想用最新的value为每个id保留一条记录。解决这个问题的最好方法是什么？更喜欢使用Pyspark

浏览 23提问于2021-09-22得票数 1

2回答

Apache Ignite到亚马逊S3的连接

ignite

我想知道如何将数据从亚马逊S3加载到Apache Ignite集群？是否需要单节点或多节点群集？

浏览 5提问于2017-05-02得票数 1

1回答

使用spark和spark streaming构建服务

apache-spark、pyspark、rabbitmq、spark-streaming

我读过一些关于spark streaming的文章，我想知道是否有可能使用rabbitmq作为代理从自定义来源流式传输数据，并通过spark流提供这些数据，Spark的机器学习和图形处理算法将在这些数据上执行，并将其发送到其他文件系统/数据库/仪表板或客户接收器。附注:我用python编写代码，我没有任何使用spark的经验，我可以称之为我试图实现的微服务吗？谢谢。

浏览 5提问于2020-09-02得票数 0

2回答

如何对Spark结构化流媒体进行单元测试？

apache-spark、spark-structured-streaming、spark-streaming-kafka

我想了解一下Spark Structured Streaming的单元测试方面。我的场景是，我从Kafka获取数据，并使用Spark Structured Streaming使用它，并在数据上应用一些转换。我不确定如何使用Scala和Spark来测试它。有人能告诉我如何使用Scala在结构化流媒体中进行单元测试吗？我还是个流媒体新手。

浏览 3提问于2019-07-05得票数 2

1回答

Spark Structured Streaming source保留策略

apache-spark、spark-structured-streaming

考虑一个关于Kafka主题的连续的JSON数据流，我们希望通过结构化的流来处理它，如下所示： .readStream .那么，有没有什么配置可以消除df中较早的数据或简单地将最早的行出队呢？

浏览 0提问于2018-02-19得票数 1

2回答

如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？

apache-spark、apache-spark-sql

如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？我从Kafka获得消息流，我想对它应用Map操作，对于每个键，我想查询像Cassandra这样的数据存储，并获得该键的更多信息，并在流上应用进一步的操作。我如何使用Spark Structured Streaming 2.2.0来做到这一点？

浏览 2提问于2017-09-07得票数 1

2回答

电火花外壳中的外部包(jars) --如何

apache-spark、pyspark

到目前为止，我读到的资料： - [Spark configuration](https://spark.ap

浏览 0提问于2019-12-11得票数 0

1回答

删除spark-structured-streaming写入的损坏拼接文件时会丢失数据吗？

scala、apache-kafka、parquet、spark-structured-streaming

我使用spark-structured-streaming作为消费者从kafka获取数据，按照指南参考https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html然后将数据保存到hdfs作为拼图文件。但我不确定是否会因为移动而导致数据丢失。我知道spark</

浏览 20提问于2019-05-25得票数 0

1回答

连接spark* structured streaming + kafka出错*

scala、apache-spark、apache-kafka、spark-streaming、spark-structured-streaming

我正在尝试连接我的结构化流式spark 2.4.5与kafka，但所有的时间，我尝试这个数据源提供程序出现错误。._ def main(args: Array[String]): Unit = { println("Spark Structured Streaming with K

浏览 86提问于2020-05-04得票数 0

回答已采纳

1回答

如何查看Spark* Streaming应用的逻辑和物理规划？*

scala、apache-spark、spark-streaming

我正在寻找一种在Spark中打印流媒体应用程序的执行计划的方法。我知道这是可能的print the plan of a SQL Spark application。下面是我的应用程序： package org.sense.spark.app import org.apache.spark.

浏览 11提问于2019-09-30得票数 0

回答已采纳

1回答

使用Spark* Structured *Streaming读取目录时，如何实现只读一次处理？

apache-spark、apache-kafka、spark-structured-streaming

我想使用流处理的概念从本地目录读取文件，然后发布到Apache Kafka。我考虑过使用Spark Structured Streaming。当读取50行文件后流式传输失败时，如何实现检查点。此外，如果我们在结构化流媒体中使用检查点，当代码有任何升级或任何更改时，我们是否会有任何问题。

浏览 0提问于2019-02-25得票数 2

1回答

我想使用Spark structured来聚合从RabbitMQ消费的数据。我知道与apache kafka有官方的spark结构化流媒体集成，我想知道是否也存在与RabbitMQ的集成？由于我不能切换现有的消息传递系统(RabbitMQ)，我想使用kafka-connect在消息传递系统(Rabbit to kafka)之间移动数据，然后使用Spark structured streaming

浏览 0提问于2018-04-16得票数 0

1回答

Nifi和Spark集成

scala、apache-spark、apache-nifi

我想在用Scala编写的Nifi自定义处理器中创建Spark会话，到目前为止，我可以在scala项目上创建spark会话，但是当我在nifi自定义处理器的OnTrigger方法中添加这个spark会话时，spark会话永远不会创建，有什么方法可以实现这一点吗？到目前为止，我已经导入了spark-core和spark-sql库，欢迎任何反馈。

浏览 4提问于2021-03-10得票数 0

1回答

齐柏林飞艇:可以用齐柏林飞艇发送卡夫卡信息吗？

apache-kafka、apache-zeppelin

._2)) 对我来说，Zeppelin相当于一个spark-shell，但我不确定我们是否可以用Spark-shell发送kafka消息。

浏览 4提问于2021-09-06得票数 0

1回答

Apache Spark与Kafka的集成

apache-spark、apache-kafka、spark-structured-streaming、spark-kafka-integration

我正在学习Udemy上关于Kafka和Spark的课程，我正在学习apache spark与Kafka的集成下面是apache spark的代码 SparkSession session = SparkSession.builder-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming --> <!-- <dependency&

浏览 26提问于2020-09-25得票数 4

回答已采纳

1回答

在达到最新偏移量之前，不会将Kafka消息写入拼图文件

apache-spark、apache-kafka、spark-streaming

我让它工作了，但后来我做了更改，现在我必须等到所有消息都被使用完后，拼图文件中才会有任何东西。我的代码如下。import org.apache.spark.sql.SparkSession def main(args: Array[String]): Unit= { .builder .getOrCreate()= "loc

浏览 4提问于2020-10-10得票数 0

回答已采纳

1回答

可以在没有HDFS的情况下使用Spark* Structured吗？*

spark-structured-streaming

我经常使用HDFS和Kafka，我注意到Kafka比HDFS更可靠。所以现在使用Spark-structured-streaming，我很惊讶检查点只有HDFS。用Kafka来勾点会更快更可靠。那么，有没有可能在没有HDFS的情况下使用spark structured？奇怪的是，我们只能在Kafka中将HDFS用于流数据。Spark 2.4.7 谢谢

浏览 39提问于2021-05-07得票数 2

1回答

如何使用MongoDB Sink集成Kafka和Spark Structured

mongodb、scala、apache-spark、apache-spark-sql、spark-structured-streaming

我正在尝试将Kafka与Spark-Structured Streaming集成到MongoDB Sink。如果我出错了，我需要帮助来纠正我的代码。集成了Kafka-Spark和Spark-Mongo。现在尝试集成来自Kafka-Spark-Mongo的管道 import org.apache.spark.sql.streaming.Triggerimport org.a

浏览 48提问于2019-04-15得票数 1

回答已采纳

1回答

如何找到spark结构的流媒体应用的消费者组id？

apache-spark、spark-structured-streaming

在spark streaming编程中，我们可以通过设置以下配置来显式分配kafka消费者组id： val kafkaParams = Map[String, Object]( ...= KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent) 在sparkstructured streaming中，禁止设置自己的消费者组id。在我的程序中，我不会更改消费者组id，而是

浏览 11提问于2020-03-31得票数 1

回答已采纳

3回答

如何在spark数据流结构中使用非基于时间的窗口？

pyspark、apache-spark-sql、spark-streaming

我正在尝试使用window on structured与spark和kafka。我在非基于时间的数据上使用window，所以我得到了这个错误： .filter(col("rank")

浏览 0提问于2019-04-09得票数 5

点击加载更多