我有传感器数据，它经常给出信号，如图所示，这里我试图找到信号=1的正信号的时间间隔。07-02 12:21:23 | 2016-07-02 13:27:07 | 3我曾尝试使用map()函数获取时间间隔，但无法成功，因为我无法使用map()函数同时获取行值。
浏览 19提问于2019-07-22得票数 1
回答已采纳
我们使用Apache在特定的时间间隔内从Twitter和Reddit等多个来源获取数据(例如，30秒)。然后，我们想把它发送到Apache，可能它应该将Twitter和Reddit消息组合成一个主题，这样Spark就可以在给定的时间间隔内接收来自两个源的数据。
浏览 1提问于2020-04-15得票数 0
回答已采纳
spark streaming fileStream如何从一个时间间隔到另一个时间间隔识别监控目录中的新文件？
它是基于新的文件名、文件创建时间戳还是任何其他方法？fileStream(String directory, Class<K> kClass, Class<V> vClass, Class<F> fClass, Function<org.apache.hadoop.fs.Path,Boolean>
浏览 1提问于2015-04-25得票数 5
2回答
我已经创建了spark应用程序，它从Apache flume获取输入数据。我将spark批处理间隔设置为4分钟，这样spark将每隔4分钟处理一次数据。但是我有一些昂贵的spark批处理，这需要相当多的时间(比如30分钟)，所以在这段时间内，大约7个spark批处理将在队列中等待，一旦昂贵的批处理执行完成，它将一个接一个地开始处理。通过这种方式，我的火花保持忙碌的最大时间。那么我怎样才能避免排队呢？
浏览 1提问于2019-09-28得票数 3
根据我们的实验，当状态变成超过一百万个对象时，有状态的星火流内部处理成本会花费大量的时间。因此，延迟会受到影响，因为我们必须增加批处理间隔以避免不稳定行为(处理时间>批处理间隔)。，那些花了这么多时间处理用户状态的星火内部处理/基础设施成本到底是多少？除了简单地增加批处理间隔之外，还有其他减少处理时间的选择吗？org.apache.spark.HashPartitioner;
import org.apache
浏览 2提问于2015-09-10得票数 24
回答已采纳
1回答
Spark流编程
、、
手头的问题可以概括为：
我有一组user_id，我启动了一个Spark Streaming任务，接收流数据并计算一些user_id的累积统计数据，它一直在运行。我是Spark Streaming作业的新手，希望我能得到一些关于上述想法的最佳实现的提示，伪代码优先。非常感谢。
浏览 0提问于2017-03-31得票数 0
3回答
我正在做一个将学生数据转换为时间间隔的小项目。程序只是读取数据，并从标记列中选择标记(整数)，以便在按升序对它们进行排序后将它们转换为间隔。任何人都可以在这一部分帮助我，非常感谢：import org.apache.spark.sql.functions._
var ascending=mrk.sort
var interval=ascendi
浏览 20提问于2017-07-17得票数 1
回答已采纳
Spark streaming textFileStream和fileStream可以监控目录并处理Dstream RDD中的新文件。如何获取DStream RDD在该特定时间间隔正在处理的文件名？
浏览 0提问于2015-03-13得票数 9
如何在Spark streaming中获取当前批量时间戳(DStream)？在执行过程中，我需要当前的时间戳来验证输入数据中的时间戳。如果我与当前时间进行比较，那么每次RDD转换执行时，时间戳可能会有所不同。
有没有办法获得时间戳，当特定的Spark streaming微批处理已经开始或它属于哪个微批处理间隔？
浏览 4提问于2015-12-23得票数 4
我面临一个问题，即我试图在固定的时间间隔内创建常规检查点，但如果自上次创建检查点以来对delta表没有任何更改，我将得到以下错误：java.lang.IllegalStateExceptionat org.apache.spark.sql.delta.Checkpoints$.writeCheckpoint(Checkpoints.scala:328)
at org.apache.spark.sql.delta.Checkpoin
浏览 3提问于2021-08-05得票数 1
回答已采纳
是针对每个批处理间隔查询偏移量还是以不同的频率查询偏移量？
浏览 1提问于2015-11-17得票数 1
我正在学习spark streaming，并从一些生产日志中找出一些散列标签。但是我不明白这个reduceByKeyAndWindow在这里是如何工作的？为什么我们要递减第二个参数中的值呢？
浏览 29提问于2019-03-17得票数 0
我正在尝试通过帮助twitter-streaming.But来保存来自twitter的保存推文，但我有一个问题:我的程序在一段时间后停止工作(取决于接近4-5秒的1毫秒的批处理间隔)。当批处理间隔接近100毫秒时，我看到一些记录，如19/08“很大”并且任何数据不可用时，我只看到关于Spark UI开始和结束的消息。package TwitterAn
浏览 7提问于2019-08-07得票数 0
回答已采纳
1回答
每个间隔数据使用RDDs并行处理，每个间隔之间没有任何数据共享。考虑示例，它生成在该间隔中接收的所有单词的计数。请考虑以下示例输入：输出：sample 1with1spark 1间隔-
浏览 2提问于2015-05-05得票数 8
回答已采纳
2回答
这是我正在尝试做的事情的一个简化版本。我想在我的udf中做一些计数。这样做的一种方法是将长累加器传递给udf，并在deserializeProtobuf函数的if else循环中递增累加器。但是不能让语法正常工作。有人能帮我吗？有没有更好的办法？ parsed_data = CustomClass.parseFrom(input_stream)
if (con
浏览 3提问于2021-03-23得票数 0
我是Apache Spark的新手。我认为问题可能是因为Spark尝试连接端口的时间间隔太短，但我不确定如何更改它，这是否真的是问题所在。
浏览 0提问于2015-01-16得票数 0
1回答
如果我执行以下计算以求出年龄：-- func.current_date() - df.DateOfBirth
有什么建议可以从间隔数据类型中提取年份值吗
浏览 1提问于2020-11-18得票数 0
1回答
import spark.implicits._
window($"timestamp", "10 minutes", "5 minutes"), .count()
我非常喜欢Spark我知道类以及如何构建自己的聚合
浏览 3提问于2021-04-30得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
