有没有办法使用readStream()方法以spark structured的形式从HashSet中读取数据？
在云计算领域，使用readStream()方法以spark structured的形式从HashSet中读取数据是不可能的。readStream()方法是Spark Structured Streaming中用于读取流式数据的方法，它支持从各种数据源（如文件系统、消息队列、数据库等）读取数据，并将其转换为DataFrame或Dataset进行处理。
然而，HashSet是Java中的一种数据结构，用于存储唯一的元素集合，并不支持以流式的方式读取数据。HashSet是基于哈希表实现的，它提供了高效的插入、删除和查找操作，但不保证元素的顺序。
如果想要以spark structured的形式读取数据，可以考虑将数据存储在支持流式读取的数据源中，例如Apache Kafka、Apache Pulsar等消息队列系统，或者使用支持流式数据处理的数据库，如Apache Cassandra、MongoDB等。这些数据源可以与Spark Structured Streaming集成，通过相应的数据源连接器读取数据，并将其转换为DataFrame或Dataset进行处理。
对于具体的实现细节和代码示例，可以参考Spark官方文档中有关Spark Structured Streaming的相关章节。
相关·内容
1回答
有没有办法使用readStream()方法以spark structured的形式从HashSet中读取数据？
apache-spark、hashmap
使用火花结构流方法的readStream()方法从HashMap读取数据。
浏览 10提问于2018-12-14得票数 0
1回答
通过spark structured streaming读取正在写入的文件
apache-spark、spark-structured-streaming
我在我的应用程序中使用spark structured。我有这样的用例，我需要在写文件的时候去读它。我尝试了spark structured，如下所示：df_str = spark.readStr
浏览 0提问于2021-01-02得票数 0
1回答
如何将kafka主题中的json字符串读入pyspark dataframe？
python、apache-spark、pyspark、apache-kafka
我正在尝试将json消息从Kafka主题读取到PySpark数据帧中。messages_json = sc.parallelize(message_lst)我想知道有没有办法使用Spark structured streaming或类似的方法来获得相同的数据帧。更新2:我设
浏览 2提问于2021-06-04得票数 0
4回答
如何读取一次流数据集并输出到多个接收器？
apache-spark、spark-structured-streaming
我有一个Spark Structured Streaming Job，它从S3读取数据，转换数据，然后将其存储到一个S3接收器和一个Elasticsearch接收器。目前，我先执行readStream一次，然后执行writeStream.format("").start()两次。当这样做时，似乎Spark从S3源读取数据两次，每个接收器一次。有没有一种更有效的<
浏览 1提问于2017-09-19得票数 14
1回答
基于google云存储的Spark结构化流媒体
apache-spark、apache-spark-sql、google-cloud-dataproc、spark-structured-streaming
我正在运行几个使用google云存储上的Avro数据的batch Spark管道。我需要更新一些管道，使其更实时，并且想知道spark structured是否可以以流的方式直接消费来自gcs的文件，即parkContext.readstream.from(...)可以应用于从外部源的存储桶下连续生成的Apache beam已经有像File.Ma
浏览 1提问于2018-01-05得票数 2
1回答
Spark readStream不会拾取输入文件中的架构更改。如何修复它？
apache-spark、pyspark、spark-streaming、spark-structured-streaming
在将CDC数据合并到基本增量表之前，我有以下脚本使用spark structured streaming读取CDC数据。streamDf = spark \ .format('csv') \ .optioncheckpoint") \
浏览 35提问于2021-07-08得票数 0
回答已采纳
4回答
如何在写拼图文件时避免空文件？
apache-spark、pyspark、spark-structured-streaming
我正在使用Spark Structured从Kafka队列中读取。读完卡夫卡之后，我在dataframe上应用了filter。我正在将这个经过过滤的数据帧保存到一个拼图文件中。这将生成许多空拼图文件。有没有什么方法可以让我不再写空文件？df = spark \ .format("kafka") \
浏览 1提问于2017-09-27得票数 2
1回答
基于kafka分区的结构化流读
apache-kafka、spark-structured-streaming、apache-spark-2.0
我使用spark structured来读取来自Kafka主题的传入消息，并根据传入消息写入多个镶嵌表，因此我创建了一个readStream，因为Kafka源很常见，并且为每个镶嵌表在循环中创建了单独的写入流这工作得很好，但是readstream正在造成一个瓶颈，因为它为每个writeStream创建了一个readStream，并且没有办法缓存已经读取的数据帧。v
浏览 1提问于2018-10-10得票数 0
2回答
我们可以在spark结构流批量模式下从Kafka中从特定偏移量获取数据吗
spark-structured-streaming
在kafka中，我动态地获得新的主题，并且我必须使用来自特定偏移量的火花流来处理它。是否有可能从变量传递json值。例如，考虑下面的代码 .read .option("kafka.bootstrap.servers", "host1:port1,host2...我试图在字符串中传递值并调用它，但它不起作用……如果我在startingOffsets中给出
浏览 0提问于2020-06-03得票数 0
2回答
Spark Structured Streaming -从嵌套目录读取文件
apache-spark、spark-streaming
我有一个客户端，它将CSV文件放在嵌套目录中，如下所示，我需要实时读取这些文件。我正在尝试使用Spark Structured Streaming来实现这一点。csv/user/data/sub1/sub2/2_1.csv代码：
val csvDF = <em
浏览 63提问于2018-07-31得票数 3
1回答
Spark structured streaming无权访问组
scala、apache-spark、apache-kafka、spark-streaming、spark-structured-streaming
我正在尝试通过spark structured streaming从Kafka中读取数据。但是，在Spark 2.4.0.中，您不能为流设置组id (参见How to set group.id for consumer group in kafka data source in Structured-13d5-4bda-ad21-7d8e43ebf518-1874352823-driver-2 有什么办法可以绕过这个
浏览 166提问于2019-12-10得票数 1
回答已采纳
1回答
在结构化流应用编程接口(pyspark)中使用redshift作为readStream的JDBC源
apache-spark、amazon-redshift、spark-structured-streaming
我正在寻找一个包，或者使用redshift作为结构化数据流的源的以前的实现。spark.readStream \ .option('url', redshift_url) \
.option('forward_spark
浏览 2提问于2020-11-18得票数 0
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
1回答
如何跟踪火花流中的数据延迟b/w EventHub和Blob
apache-spark、spark-streaming、azure-eventhub
我工作的火花流基本上是从EventHub读取接近实时的数据，并转储到斑点位置，我将需要实现水印，以查看什么是延迟(数据滞后) b/w的EventHub和斑点位置。正如我们所知，EventHub只保存2天的数据，我需要确保没有数据丢失b/w EventHub和blob位置。有没有办法在Spark structured</
浏览 13提问于2019-11-06得票数 0
2回答
压缩批次9时IllegalStateException：_spark_metadata/0不存在
scala、apache-spark、apache-kafka、spark-structured-streaming
我们有使用Spark Structured实现的流应用程序，它试图从Kafka主题读取数据并将其写入HDFS位置。有时应用程序会失败，并出现异常：java.lang.IllegalStateException我找到的唯一解决方案是删除检查点位置文件，这将使作业在我们再次运行应用程序时从头开始读
浏览 0提问于2019-05-31得票数 6
1回答
基于spark结构流的Xml解析
apache-spark、pyspark、xml-parsing、spark-structured-streaming
我正在尝试使用PySpark Structured Streaming on Databricks中的Kinesis source分析数据。 我创建了一个Dataframe，如下所示。kinDF = spark.readStream.format("kinesis").如果我以xml_df = spark.read.format("xml").options(rowTag='Consu
浏览 36提问于2019-08-15得票数 1
1回答
处理数据- Spark结构流
apache-spark、pyspark、apache-kafka、spark-structured-streaming
据我所知，spark structured是通过使用检查点实现容错的。
我想读卡夫卡的故事。假设我使用了检查点，然后由于某种原因，我的代码崩溃了/我停止了它，然后我希望当我重新运行代码时，它会恢复处理过的数据。我的问题是，在读取配置中，如果我将偏移量设置为最早，那么在重新运行代码后，我将再次读取相同的数据，如果我将最新
浏览 1提问于2019-04-03得票数 0
1回答
Spark Structured Streaming with Kafka source，在查询运行时更改主题分区的数量
apache-spark、apache-kafka、partitioning、spark-structured-streaming、spark-streaming-kafka
我设置了一个从Kafka主题读取的Spark结构化流查询。如果在Spark查询运行时更改了主题中的分区数量，Spark似乎不会注意到，并且不会消耗新分区上的数据。除了停止和重新启动查询之外，有没有办法让Spark检查同一主题中的新分区？
编辑:我使用的是Spark 2.4.4。我从卡夫卡<em
浏览 32提问于2019-11-08得票数 0
2回答
用消防软管从分区文件夹中读取JSON
apache-spark、apache-spark-sql、databricks、spark-structured-streaming
Kinesis将文件的持久性(在本例中为时间序列JSON )管理为一个文件夹层次结构，该层次结构由YYYY/MM/DD/HH (直到24小时编号)...great划分。那么我如何使用Spark2.0来读取这些嵌套的子文件夹，并从所有的叶json文件中创建一个静态的Dataframe呢？数据阅读器有“选项”吗？我的下一个目标是成为一个流DF，火龙将新文件持久化到s3中，<
浏览 4提问于2016-10-30得票数 8
1回答
Spark structured streaming read from kafka json编码问题
json、apache-spark、apache-kafka、apache-kafka-connect、spark-structured-streaming
我很难使用Spark Structured Streaming在kafka主题中读取我的JSON数据。上下文： 我正在构建一个简单的管道，在这里我使用kafka从MongoDb读取数据(这个数据库通常是从另一个应用程序中填充的)，然后我想在Spark中获取这些数据。为此，我<
浏览 13提问于2019-03-11得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
