想知道什么(以及如何修改)会触发Spark Sturctured流查询(配置了拼图文件输出接收器)以将数据写入拼图文件。我定期提供流输入数据(使用StreamReader读入文件)，但它不会为作为输入提供的每个文件将输出写入Parquet文件。一旦我给它提供了几个文件，它往往会很好地编写一个Parquet文件。我希望能够强制每个新的文件作为输入提供到拼图文
浏览 7提问于2019-03-28得票数 1
1回答
我们使用这个sql并调用spark。spark.sqlContext.sql(statement) // hive-metastore integration is enabled
这导致火花驱动程序占用过多内存，我们是否可以使用火花流(或结构化流)以管道方式流，而不是收集驱动程序上的所有内容，然后发送给客户端？我们不想(在典型的流应用程序中)立即发送数据，而是希望在客户要求(拉)数据时向客户发送流数据。
浏览 1提问于2020-07-12得票数 0
我使用spark(3.0.0)结构化流从kafka读取主题。我使用了joins，然后使用了mapGropusWithState来获取流数据，因此，根据我对火花官方指南：的理解，我不得不使用更新模式。下面的火花官方指南部分没有提到DB sink，它也不支持为update mode：写到files
目前，我将其输出到console，并希望将数据存储在文件或DB中。因此，我的问题是:在我的情况下，如何将流数据写入数据库或文件？我是否必须将数
浏览 5提问于2020-09-16得票数 0
我使用spark-structured-streaming作为消费者从kafka获取数据，按照指南参考https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html然后将数据保存到hdfs作为拼图文件。这是我的问题:程序运行良好，但一些容器很少失败(但它确实发生了)，导致了一些损坏的拼接文件。它将导致错误，如不是拼图文件(长度太小: 4)或[
浏览 20提问于2019-05-25得票数 0
目前，我的Spark结构化流是这样的(Sink部件只显示)： aggregationQuery.writeStreamseconds")) .option("path", "hdfs://<myip
浏览 0提问于2018-05-18得票数 2
1回答
结构化流式传输指标性能？
、、、
在尝试了一些监视结构化流性能和输入/输出指标的方法后，我发现一种可靠的方法是附加streamingQueryListener来输出streamingQueryProgress，以获得输入/输出数量。除了SparkUI，将queryProgress输出到文件或Kafka的最好方法是什么？在spark streaming和spark struc
浏览 1提问于2018-05-19得票数 2
我的意图是使用结构化流从一个Kafka主题消费，做一些处理，并存储到EMRFS/S3的拼花格式。val event = spark.readStream.format("kafka")
sc.hadoopConfiguration.set("parquet
浏览 0提问于2018-04-09得票数 13
我有一个Azure数据湖存储容器，它充当由Apache处理JSON文件的着陆区。spark.read.json("/mnt/input_location/*.json")
创建一个带有已处理的文
浏览 2提问于2021-04-28得票数 3
回答已采纳
我希望将卡夫卡主题的所有聚合数据一次写到一个拼花文件中(或者至少在最后有一个拼花文件)。Dataset<Row> df = spark.readStream()
.format("kafka")当我试图将它写到parquet时，我只会得到多个空的parquet文件。.config("spark.master", "loc
浏览 1提问于2019-05-29得票数 0
我想从tar-gzip文件(tgz)中进行流处理，其中包括我实际存储的CSV数据。当我的数据以CSV文件的形式出现时，我已经设法用spark 2.2实现了结构化流，但实际上，数据是以压缩的csv文件的形式出现的。在处理CSV流之前，是否有一种通过结构化流执行的触发器进行解压缩的方法？，产生预期的输出。当我试图将这些文件放在给定的路径上时，我不会得到任何异常，批处理输出</e
浏览 12提问于2017-12-30得票数 2
回答已采纳
我们正在考虑在一个项目中使用Spark结构化流。输入和输出是S3桶上的拼图文件。是否有可能以某种方式控制输出文件的大小？我们的目标是输出文件大小为10-100 of。据我所知，在传统的批处理方法中，我们可以根据输入数据集的大小调整分区的数量来确定输出文件的大小，在结构化流中类似吗？
浏览 0提问于2019-02-14得票数 2
我正在使用Spark结构化流写入Elasticsearch 6.2.0 Sink： .writeStream .queryName("ElasticSink")
.option("checkpointLocation", s"
浏览 3提问于2018-09-03得票数 1
回答已采纳
1回答
在尝试了许多更改之后，我仍然遇到了将Flume自定义接收器和Spark Stream绑定到同一端口的问题。我使用的是Flume 1.6.0和Spark 2.0.0。使用Spark+flume集成指南，我构建了.conf文件和spark .py文件。我首先启动flume代理，但当我尝试在同一端口上使用Spark-streaming (读取)时，它显示fail to bind。我尝试了多个端口，但据我所知，无法找到任何网络限制(完全权限)。
浏览 17提问于2016-09-28得票数 0
spark.sparkContext.addSparkListener(new SparkListener() { "processedRowsPerSecond" : 3013.733905579399 "sink" : {
"description" : "
浏览 0提问于2018-07-25得票数 3
有没有可能将Kafka流源支持的数据帧写入到AWS Redshift中，我们过去曾使用写入到Redshift中，但我认为它不会与DataFrame##writeStream一起工作。我在中遇到的一种可能的方法是将文件写入S3，然后使用具有S3对象路径的清单文件调用。在结构化流的情况下，我如何控制写入S3的文件？还有一个单独的触发器，用于在将5个文件写入S3后创建清单文件。
浏览 14提问于2018-01-19得票数 0
回答已采纳
1回答
是否有方法为星火结构化文件流源指定起始偏移量？我正在尝试从HDFS源源不断地发送文件： .parquetcheckpointLocation", "/tmp/streaming-test/checkpoint")
.format(&q
浏览 1提问于2018-07-18得票数 3
回答已采纳
Spark 2.2引入了Kafka的结构化流媒体源代码。据我所知，它依靠HDFS检查点目录来存储偏移量，并保证“只传递一次”消息。但旧的docks (如)表示，Spark Streaming检查点不能跨应用程序或Spark升级恢复，因此不太可靠。如果要将Kafka源的偏移量存储到事务型DB中，如何从结构化流批处理中获取偏移量？RDD转换为HasOffsetRanges来完成
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offs
浏览 1提问于2017-09-11得票数 28
回答已采纳
1回答
我有一个处理文件记录的方案。文件中的数据定期添加(每毫秒一次)。因此，我需要读取文件并处理它，同时只处理新添加的记录。
private void processData
浏览 1提问于2017-07-13得票数 2
2回答
我已经为我的结构化流应用程序启用了WAL。我在哪里能找到WAL日志的位置？我能够在前缀receivedBlockMetadata中看到我的星火流进程的WAL。但是，我没有看到为结构化流创建的任何前缀
浏览 2提问于2020-02-24得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
