在Spark Streaming中处理太迟的数据

文章/答案/技术大牛

发布

1回答

scala、apache-spark、spark-structured-streaming

水印允许使用窗口在一段时间内考虑将延迟到达的数据包含在已经计算的结果中。它的前提是跟踪到一个时间点，在这个时间点之前，假设不会有更多的延迟事件到达，但如果它们到达了，它们仍然是discarded。有没有办法存储丢弃的数据，以便稍后用于对帐？例如，在我的结构化流媒体中，我将水印设置为1小时。我每10分钟做一次窗口操作，并在20分钟后收到一个较晚的事件。是否有一种方法

浏览 28提问于2019-07-03得票数 0

回答已采纳

3回答

使用Spark Streaming时限制Kafka批量大小

apache-spark、apache-kafka、spark-streaming、kafka-consumer-api

是否可以限制Kafka消费者返回Spark Streaming的批量大小？我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。

浏览 0提问于2016-10-11得票数 18

回答已采纳

1回答

使用while循环中的Spark处理日志文件

apache-spark、pyspark

我有一个服务器，每1秒生成一些日志文件，我想使用Apache Spark处理这个文件。我的问题是，对于这种无限运行并处理批量或一组生成文件的应用程序，最好的方法是什么。我应该使用一个无限的while循环，还是应该在cron job甚至像airflo

浏览 0提问于2017-04-18得票数 0

1回答

这个问题脱离了我关于管理AmazonDynamoDbClient节流和重试的另一个问题。然而，我认为解决方案可能存在之前，我甚至到发电机电话。我的高级过程如下:我有一个scala应用程序，它使用Apache读取大型CSV文件，并对它们执行一些聚合，然后将它们写入dynamo。我将其部署到EMR以提供可伸缩性。问题是，一旦聚合完成，我们就有数以百万计的记录准备进入发电机，但我们有一个写容量的发电机。它们不需要立即插入，但是最好能控制每秒多少次，这样我们就可以根据用例对其进行微调。下面是我到目前为止所得到的代

浏览 3提问于2017-03-28得票数 3

2回答

什么是“偏移从X改为0”错误与KafkaSource在星火结构化流？

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

在带有检查点的Spark结构化流应用程序中，我得到了“偏移量从X更改为0，一些数据可能被漏掉”的错误，但它似乎并没有真正造成任何问题。我想弄清楚这个错误到底意味着什么。我的设置如下。我在另一个码头容器中有一个Spark结构化流(2.1.1)应用程序。溪流消耗卡夫卡的数据。它们还在重新挂载到指定卷中的位置使用检查点，以确保在重新启动之间保留元<e

浏览 2提问于2017-05-09得票数 3

1回答

如何在Spark streaming中避免批量大小的突然峰值？

apache-spark、apache-kafka、spark-streaming

我正在从kafka流式传输数据，并试图将每批事件的数量限制在10个事件。在处理10-15批次后，批次大小突然出现峰值。下面是我的设置： spark.streaming.backpressure.pid.min

浏览 4提问于2017-11-17得票数 0

1回答

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

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

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

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

1回答

如何限制星图操作？

apache-spark

我有一个S3 json数据集，它是KMS客户端加密DynamoDB的转储(即每条记录都是独立加密的KMS客户端)。我想知道是否有什么办法限制这些星图操作？

浏览 7提问于2017-05-13得票数 1

1回答

spark streaming中限制Kafka消费数据

apache-kafka、spark-streaming、kafka-consumer-api

Spark从kafka获取数据。我想通过spark-streaming限制记录消耗。关于卡夫卡的数据非常庞大。我使用spark.streaming.kafka.maxRatePerPartition=1属性来限制spark中的记录。但是仍然在5分钟内，我收到了13400条消息。我的spark程序每5分钟不能处理超过1000条消息

浏览 17提问于2018-02-06得票数 0

1回答

Spark Job未将消息发布到Kafka主题

scala、apache-spark、apache-kafka、spark-streaming、kafka-producer-api

我已经编写了spark作业来读取一个文件，将数据转换为json并将数据发布到Kafka:我尝试了所有选项，如1. posting thread.sleep 2.更改linger.ms小于thread.sleep.But什么都不工作out..it只是不发布任何东西到kafKa .I尝试过producer.flush()/producer.close().No错误在log.But中仍然没有发布任何东西。如果我写一个简单的独立制作人来发布消息到相同的kafka主题，

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

3回答

持续信息JobScheduler:59 -在我的Spark独立集群中添加作业时间*毫秒

apache-spark、spark-streaming、apache-spark-standalone

我们正在使用具有8核和32 We的Spark独立集群，具有相同配置的3个节点集群。有时批处理流在不到1秒的时间内完成。在某些情况下，控制台中会出现以下日志，该时间超过10秒。

浏览 0提问于2016-03-29得票数 3

1回答

Spark Streaming Kafka直接消费者消费速度下降

scala、amazon-web-services、apache-spark、apache-kafka、spark-streaming

我使用在亚马逊网络服务中运行的spark独立集群(spark和spark-streaming-kafka版本1.6.1)，并使用检查点目录StreamingContext.getOrCreate(config.sparkConfig.checkpointDir, createStreamingContext)的</

浏览 2提问于2016-12-19得票数 2

1回答

hadoop同步日志的体系结构

hadoop、real-time

我在几个云提供商之间有不同的环境，比如windows服务器、rackspace中的linux服务器、aws..etc。在这和内部网络之间有一道防火墙。我需要构建一个实时服务器环境，在这个环境中，所有新生成的IIS日志、apache日志都将同步到内部大数据环境。我知道有些工具，如Splunk或Sumologic可能会有所帮助，但我们需要在开源技术中实现这种逻辑。由于防火墙的存在，我假设我只能从云提供商那里提取日志而不是推送。谁能和我分享

浏览 4提问于2015-12-28得票数 0

回答已采纳

1回答

无法使用Spark* Structured Streaming覆盖默认值"spark.sql.shuffle.partitions“*

scala、apache-spark、spark-structured-streaming

我想直接在代码中覆盖spark.sql.shuffle.partitions参数： val sparkSession = SparkSession .appName("SPARK") sparkSession.conf.set("spark.sql.shuffle.partitions", 2) 但此设置不会生效，因为在日志中我收到以下警告消息虽然在spar

浏览 39提问于2021-04-30得票数 2

1回答

火花流|将不同的数据帧并行写入多个表

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

我正在从Kafka读取数据并加载到数据仓库中，从一个Kafka主题中我正在创建一个数据框架，在应用了所需的转换之后，我正在从中创建多个DFs并将这些DFs加载到不同的表中，但这个操作是按顺序进行的。| | |-- primary: boolean (nullable = true) 我分别为attribute1Formatted和attribute2Formatted创建了两个不同的数据

浏览 1提问于2021-07-26得票数 1

2回答

数据源io.pivotal.greenplum.spark.GreenplumRelationProvider不支持流写入。

scala、apache-kafka、spark-streaming、greenplum

我试图读取卡夫卡的数据，并将其上传到格林梅利的数据库使用火花。我使用的是格林梅-火花连接器，但我正在获取数据源，io.pivotal.greenplum.spark.GreenplumRelationProvider不支持流写入。是否格林梅源不支持流媒体数据？我曾尝试将数据源命名为“绿梅”，并将"io.pivotal.greenplum.spark.GreenplumRelationProvider“转换为

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

回答已采纳

1回答

从Mesos集群收集日志

logging、apache-spark、flume、mesos

我的团队正在亚马逊EC2实例上部署一个新的集群。经过一些研究，我们决定使用Apache Mesos作为集群管理器，使用Spark进行计算。我们问自己的第一个问题是，对于每个不同的框架，从所有机器收集日志的最佳方式是什么。到目前为止，我们开发了一些自定义的bash/python脚本，它们从预定义的位置收集日志，将它们压缩并将压缩后的文件发送到S3。这种轮换由cron作业激活，该作业每小时运行一次。我一直在寻找

浏览 0提问于2015-06-26得票数 1

1回答

显示Spark结构化流作业使用的事件数

apache-spark、spark-structured-streaming

我有2个线性火花结构化流式作业，从一个卡夫卡主题复制数据到另一个。是否可以在Spark UI中发布/查看消费/产生的事件数量？

浏览 13提问于2020-04-15得票数 1

回答已采纳

1回答

在哪里可以找到与spark* structured streaming相关的所有属性？*

scala、spark-streaming、spark-structured-streaming

我想知道是否有与spark structured streaming相关的所有属性的列表？例如，在文档中，我们可以找到：当我按照spark sql上的配置文档中的建议执行spark.s

浏览 2提问于2018-09-03得票数 0

0回答

dstream.checkpoint(checkpointInterval)和spark.streaming.receiver.writeAheadLog.enable之间是否有联系

spark-streaming

根据spark streaming文档，已排队但未处理的数据流将在每个checkpointInterval之后写入检查点目录。但同时有一个配置spark.streaming.receiver.writeAheadLog.enable = true，根据spark文档“从接收器接收的所有数据都会写入配置检查点目录中的预写日志”。spark给出的每个

浏览 3提问于2017-12-07得票数 1

点击加载更多