实现了滑动窗口计数和TopN排序, 比较有意思, 具体分析一下代码
Topology
这是一个稍微复杂些的topology, 主要体现在使用不同的grouping方式, fieldsGrouping和globalGrouping
String spoutId = "wordGenerator"; String counterId = "counter"; String intermediateRankerId = "intermediateRanker"; String totalRankerId = "finalRanker"; builder.setSpout(spoutId, new TestWordSpout(), 5); builder.setBolt(counterId, new RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, new Fields("word")); builder.setBolt(intermediateRankerId, new IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, new Fields("obj")); builder.setBolt(totalRankerId, new TotalRankingsBolt TOP_N)).globalGrouping(intermediateRankerId);
RollingCountBolt
首先使用RollingCountBolt, 并且此处是按照word进行fieldsGrouping的, 所以相同的word会被发送到同一个bolt, 这个field id是在上一级的declareOutputFields时指定的
RollingCountBolt, 用于基于时间窗口的counting, 所以需要两个参数, the length of the sliding window in seconds和the emit frequency in seconds
new RollingCountBolt(9, 3), 意味着output the latest 9 minutes sliding window every 3 minutes
1. 创建SlidingWindowCounter(SlidingWindowCounter和SlotBasedCounter参考下面)
counter = new SlidingWindowCounter(this.windowLengthInSeconds / this.windowUpdateFrequencyInSeconds);
如何定义slot数? 对于9 min的时间窗口, 每3 min emit一次数据, 那么就需要9/3=3个slot
那么在3 min以内, 不停的调用countObjAndAck(tuple)来递增所有对象该slot上的计数
每3分钟会触发调用emitCurrentWindowCounts, 用于滑动窗口(通过getCountsThenAdvanceWindow), 并emit (Map<obj, 窗口内的计数和>, 实际使用时间)
因为实际emit触发时间, 不可能刚好是3 min, 会有误差, 所以需要给出实际使用时间
2. TupleHelpers.isTickTuple(tuple), TickTuple
前面没有说的一点是, 如何触发emit? 这是比较值得说明的一点, 因为其使用Storm的TickTuple特性.
这个功能挺有用, 比如数据库批量存储, 或者这里的时间窗口的统计等应用
"__system" component会定时往task发送 "__tick" stream的tuple
发送频率由TOPOLOGY_TICK_TUPLE_FREQ_SECS来配置, 可以在default.ymal里面配置
也可以在代码里面通过getComponentConfiguration()来进行配置,
public Map<String, Object> getComponentConfiguration() { Map<String, Object> conf = new HashMap<String, Object>(); conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, emitFrequencyInSeconds); return conf;
配置完成后, storm就会定期的往task发送ticktuple
只需要通过isTickTuple来判断是否为tickTuple, 就可以完成定时触发的功能
public static boolean isTickTuple(Tuple tuple) { return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) \\ SYSTEM_COMPONENT_ID == "__system" && tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID); \\ SYSTEM_TICK_STREAM_ID == "__tick" }
最终, 这个blot的输出为, collector.emit(new Values(obj, count, actualWindowLengthInSeconds));
obj, count(窗口内的计数和), 实际使用时间
SlotBasedCounter
基于slot的counter, 模板类, 可以指定被计数对象的类型T
这个类其实很简单, 实现计数对象和一组slot(用long数组实现)的map, 并可以对任意slot做increment或reset等操作
关键结构为Map<T, long[]> objToCounts, 为每个obj都对应于一个大小为numSlots的long数组, 所以对每个obj可以计numSlots个数
incrementCount, 递增某个obj的某个slot, 如果是第一次需要创建counts数组
getCount, getCounts, 获取某obj的某slot值, 或某obj的所有slot值的和
wipeSlot, resetSlotCountToZero, reset所有对象的某solt为0, reset某obj的某slot为0
wipeZeros, 删除所有total count为0的obj, 以释放空间
public final class SlotBasedCounter<T> implements Serializable { private static final long serialVersionUID = 4858185737378394432L; private final Map<T, long[]> objToCounts = new HashMap<T, long[]>(); private final int numSlots; public SlotBasedCounter(int numSlots) { if (numSlots <= 0) { throw new IllegalArgumentException("Number of slots must be greater than zero (you requested " + numSlots + ")"); } this.numSlots = numSlots; } public void incrementCount(T obj, int slot) { long[] counts = objToCounts.get(obj); if (counts == null) { counts = new long[this.numSlots]; objToCounts.put(obj, counts); } counts[slot]++; } public long getCount(T obj, int slot) { long[] counts = objToCounts.get(obj); if (counts == null) { return 0; } else { return counts[slot]; } } public Map<T, Long> getCounts() { Map<T, Long> result = new HashMap<T, Long>(); for (T obj : objToCounts.keySet()) { result.put(obj, computeTotalCount(obj)); } return result; } private long computeTotalCount(T obj) { long[] curr = objToCounts.get(obj); long total = 0; for (long l : curr) { total += l; } return total; } /** * Reset the slot count of any tracked objects to zero for the given slot. * * @param slot */ public void wipeSlot(int slot) { for (T obj : objToCounts.keySet()) { resetSlotCountToZero(obj, slot); } } private void resetSlotCountToZero(T obj, int slot) { long[] counts = objToCounts.get(obj); counts[slot] = 0; } private boolean shouldBeRemovedFromCounter(T obj) { return computeTotalCount(obj) == 0; } /** * Remove any object from the counter whose total count is zero (to free up memory). */ public void wipeZeros() { Set<T> objToBeRemoved = new HashSet<T>(); for (T obj : objToCounts.keySet()) { if (shouldBeRemovedFromCounter(obj)) { objToBeRemoved.add(obj); } } for (T obj : objToBeRemoved) { objToCounts.remove(obj); } } }
SlidingWindowCounter
SlidingWindowCounter只是对SlotBasedCounter做了进一步的封装, 通过headSlot和tailSlot提供sliding window的概念
incrementCount, 只能对headSlot进行increment, 其他slot作为窗口中的历史数据
核心的操作为, getCountsThenAdvanceWindow
1. 取出Map<T, Long> counts, 对象和窗口内所有slots求和值的map
2. 调用wipeZeros, 删除已经不被使用的obj, 释放空间
3. 最重要的一步, 清除tailSlot, 并advanceHead, 以实现滑动窗口
advanceHead的实现, 如何在数组实现循环的滑动窗口
public final class SlidingWindowCounter<T> implements Serializable { private static final long serialVersionUID = -2645063988768785810L; private SlotBasedCounter<T> objCounter; private int headSlot; private int tailSlot; private int windowLengthInSlots; public SlidingWindowCounter(int windowLengthInSlots) { if (windowLengthInSlots < 2) { throw new IllegalArgumentException("Window length in slots must be at least two (you requested " + windowLengthInSlots + ")"); } this.windowLengthInSlots = windowLengthInSlots; this.objCounter = new SlotBasedCounter<T>(this.windowLengthInSlots); this.headSlot = 0; this.tailSlot = slotAfter(headSlot); } public void incrementCount(T obj) { objCounter.incrementCount(obj, headSlot); } /** * Return the current (total) counts of all tracked objects, then advance the window. * * Whenever this method is called, we consider the counts of the current sliding window to be available to and * successfully processed "upstream" (i.e. by the caller). Knowing this we will start counting any subsequent * objects within the next "chunk" of the sliding window. * * @return */ public Map<T, Long> getCountsThenAdvanceWindow() { Map<T, Long> counts = objCounter.getCounts(); objCounter.wipeZeros(); objCounter.wipeSlot(tailSlot); advanceHead(); return counts; } private void advanceHead() { headSlot = tailSlot; tailSlot = slotAfter(tailSlot); } private int slotAfter(int slot) { return (slot + 1) % windowLengthInSlots; } }
IntermediateRankingsBolt
这个bolt作用就是对于中间结果的排序, 为什么要增加这步, 应为数据量比较大, 如果直接全放到一个节点上排序, 会负载太重
所以先通过IntermediateRankingsBolt, 过滤掉一些
这里仍然使用, 对于obj进行fieldsGrouping, 保证对于同一个obj, 不同时间段emit的统计数据会被发送到同一个task
IntermediateRankingsBolt继承自AbstractRankerBolt(参考下面)
并实现了updateRankingsWithTuple,
void updateRankingsWithTuple(Tuple tuple) { Rankable rankable = RankableObjectWithFields.from(tuple); super.getRankings().updateWith(rankable); }
逻辑很简单, 将Tuple转化Rankable, 并更新Rankings列表
参考AbstractRankerBolt, 该bolt会定时将Ranking列表emit出去
Rankable
Rankable除了继承Comparable接口, 还增加getObject()和getCount()接口
public interface Rankable extends Comparable<Rankable> { Object getObject(); long getCount(); }
RankableObjectWithFields
RankableObjectWithFields实现Rankable接口
1. 提供将Tuple转化为RankableObject
Tuple由若干field组成, 第一个field作为obj, 第二个field作为count, 其余的都放到List<Object> otherFields中
2. 实现Rankable定义的getObject()和getCount()接口
3. 实现Comparable接口, 包含compareTo, equals
public class RankableObjectWithFields implements Rankable public static RankableObjectWithFields from(Tuple tuple) { List<Object> otherFields = Lists.newArrayList(tuple.getValues()); Object obj = otherFields.remove(0); Long count = (Long) otherFields.remove(0); return new RankableObjectWithFields(obj, count, otherFields.toArray()); }
Rankings
Rankings维护需要排序的List, 并提供对List相应的操作
核心的数据结构如下, 用来存储rankable对象的list
List<Rankable> rankedItems = Lists.newArrayList();
提供一些简单的操作, 比如设置maxsize(list size), getRankings(返回rankedItems, 排序列表)
核心的操作是,
public void updateWith(Rankable r) { addOrReplace(r); rerank(); shrinkRankingsIfNeeded(); }
上一级的blot会定期的发送某个时间窗口的(obj, count), 所以obj之间的排序是在不断变化的
1. 替换已有的, 或新增rankable对象(包含obj, count)
2. 从新排序(Collections.sort)
3. 由于只需要topN, 所以大于maxsize的需要删除
AbstractRankerBolt
首先以TopN为参数, 创建Rankings对象
private final Rankings rankings; public AbstractRankerBolt(int topN, int emitFrequencyInSeconds) { count = topN; this.emitFrequencyInSeconds = emitFrequencyInSeconds; rankings = new Rankings(count); }
在execute中, 也是定时触发emit, 同样是通过emitFrequencyInSeconds来配置tickTuple
一般情况, 只是使用updateRankingsWithTuple不断更新Rankings
这里updateRankingsWithTuple是abstract函数, 需要子类重写具体的update逻辑
public final void execute(Tuple tuple, BasicOutputCollector collector) { if (TupleHelpers.isTickTuple(tuple)) { emitRankings(collector); } else { updateRankingsWithTuple(tuple); } }
最终将整个rankings列表emit出去
private void emitRankings(BasicOutputCollector collector) { collector.emit(new Values(rankings)); getLogger().info("Rankings: " + rankings); }
TotalRankingsBolt
该bolt会使用globalGrouping, 意味着所有的数据都会被发送到同一个task进行最终的排序.
TotalRankingsBolt同样继承自AbstractRankerBolt
void updateRankingsWithTuple(Tuple tuple) { Rankings rankingsToBeMerged = (Rankings) tuple.getValue(0); super.getRankings().updateWith(rankingsToBeMerged); }
唯一的不同是, 这里updateWith的参数是个rankable列表, 在Rankings里面的实现一样, 只是多了遍历
最终可以得到, 全局的TopN的Rankings列表
相关推荐
618商品网页制作编程示例开发案列优质学习资料资源工具与案列应用场景开发文档教程资料
提供的源码资源涵盖了安卓应用、小程序、Python应用和Java应用等多个领域,每个领域都包含了丰富的实例和项目。这些源码都是基于各自平台的最新技术和标准编写,确保了在对应环境下能够无缝运行。同时,源码中配备了详细的注释和文档,帮助用户快速理解代码结构和实现逻辑。 适用人群: 这些源码资源特别适合大学生群体。无论你是计算机相关专业的学生,还是对其他领域编程感兴趣的学生,这些资源都能为你提供宝贵的学习和实践机会。通过学习和运行这些源码,你可以掌握各平台开发的基础知识,提升编程能力和项目实战经验。 使用场景及目标: 在学习阶段,你可以利用这些源码资源进行课程实践、课外项目或毕业设计。通过分析和运行源码,你将深入了解各平台开发的技术细节和最佳实践,逐步培养起自己的项目开发和问题解决能力。此外,在求职或创业过程中,具备跨平台开发能力的大学生将更具竞争力。 其他说明: 为了确保源码资源的可运行性和易用性,特别注意了以下几点:首先,每份源码都提供了详细的运行环境和依赖说明,确保用户能够轻松搭建起开发环境;其次,源码中的注释和文档都非常完善,方便用户快速上手和理解代码;最后,我会定期更新这些源码资源,以适应各平台技术的最新发展和市场需求。
提供的源码资源涵盖了安卓应用、小程序、Python应用和Java应用等多个领域,每个领域都包含了丰富的实例和项目。这些源码都是基于各自平台的最新技术和标准编写,确保了在对应环境下能够无缝运行。同时,源码中配备了详细的注释和文档,帮助用户快速理解代码结构和实现逻辑。 适用人群: 这些源码资源特别适合大学生群体。无论你是计算机相关专业的学生,还是对其他领域编程感兴趣的学生,这些资源都能为你提供宝贵的学习和实践机会。通过学习和运行这些源码,你可以掌握各平台开发的基础知识,提升编程能力和项目实战经验。 使用场景及目标: 在学习阶段,你可以利用这些源码资源进行课程实践、课外项目或毕业设计。通过分析和运行源码,你将深入了解各平台开发的技术细节和最佳实践,逐步培养起自己的项目开发和问题解决能力。此外,在求职或创业过程中,具备跨平台开发能力的大学生将更具竞争力。 其他说明: 为了确保源码资源的可运行性和易用性,特别注意了以下几点:首先,每份源码都提供了详细的运行环境和依赖说明,确保用户能够轻松搭建起开发环境;其次,源码中的注释和文档都非常完善,方便用户快速上手和理解代码;最后,我会定期更新这些源码资源,以适应各平台技术的最新发展和市场需求。
背景:一维卡尔曼滤波的MATLAB例程,背景为温度估计。 代码介绍:运行程序后,可以自己输入温度真实值 滤波前的值和滤波后的值分别于期望值(真实值)作差,可以得到误差曲线图 更多介绍:https://blog.csdn.net/callmeup/article/details/139374191
数仓建模
提供的源码资源涵盖了安卓应用、小程序、Python应用和Java应用等多个领域,每个领域都包含了丰富的实例和项目。这些源码都是基于各自平台的最新技术和标准编写,确保了在对应环境下能够无缝运行。同时,源码中配备了详细的注释和文档,帮助用户快速理解代码结构和实现逻辑。 适用人群: 这些源码资源特别适合大学生群体。无论你是计算机相关专业的学生,还是对其他领域编程感兴趣的学生,这些资源都能为你提供宝贵的学习和实践机会。通过学习和运行这些源码,你可以掌握各平台开发的基础知识,提升编程能力和项目实战经验。 使用场景及目标: 在学习阶段,你可以利用这些源码资源进行课程实践、课外项目或毕业设计。通过分析和运行源码,你将深入了解各平台开发的技术细节和最佳实践,逐步培养起自己的项目开发和问题解决能力。此外,在求职或创业过程中,具备跨平台开发能力的大学生将更具竞争力。 其他说明: 为了确保源码资源的可运行性和易用性,特别注意了以下几点:首先,每份源码都提供了详细的运行环境和依赖说明,确保用户能够轻松搭建起开发环境;其次,源码中的注释和文档都非常完善,方便用户快速上手和理解代码;最后,我会定期更新这些源码资源,以适应各平台技术的最新发展和市场需求。
提供的源码资源涵盖了安卓应用、小程序、Python应用和Java应用等多个领域,每个领域都包含了丰富的实例和项目。这些源码都是基于各自平台的最新技术和标准编写,确保了在对应环境下能够无缝运行。同时,源码中配备了详细的注释和文档,帮助用户快速理解代码结构和实现逻辑。 适用人群: 这些源码资源特别适合大学生群体。无论你是计算机相关专业的学生,还是对其他领域编程感兴趣的学生,这些资源都能为你提供宝贵的学习和实践机会。通过学习和运行这些源码,你可以掌握各平台开发的基础知识,提升编程能力和项目实战经验。 使用场景及目标: 在学习阶段,你可以利用这些源码资源进行课程实践、课外项目或毕业设计。通过分析和运行源码,你将深入了解各平台开发的技术细节和最佳实践,逐步培养起自己的项目开发和问题解决能力。此外,在求职或创业过程中,具备跨平台开发能力的大学生将更具竞争力。 其他说明: 为了确保源码资源的可运行性和易用性,特别注意了以下几点:首先,每份源码都提供了详细的运行环境和依赖说明,确保用户能够轻松搭建起开发环境;其次,源码中的注释和文档都非常完善,方便用户快速上手和理解代码;最后,我会定期更新这些源码资源,以适应各平台技术的最新发展和市场需求。
LaNi5建模参数与变量
web期末作业设计网页 基于springboot+vue+MySQL实现的在线考试系统+源代码+文档
Windows11_InsiderPreview_EnterpriseVL_x64_zh-cn_26080.iso.003
Python库是一组预先编写的代码模块,旨在帮助开发者实现特定的编程任务,无需从零开始编写代码。这些库可以包括各种功能,如数学运算、文件操作、数据分析和网络编程等。Python社区提供了大量的第三方库,如NumPy、Pandas和Requests,极大地丰富了Python的应用领域,从数据科学到Web开发。Python库的丰富性是Python成为最受欢迎的编程语言之一的关键原因之一。这些库不仅为初学者提供了快速入门的途径,而且为经验丰富的开发者提供了强大的工具,以高效率、高质量地完成复杂任务。例如,Matplotlib和Seaborn库在数据可视化领域内非常受欢迎,它们提供了广泛的工具和技术,可以创建高度定制化的图表和图形,帮助数据科学家和分析师在数据探索和结果展示中更有效地传达信息。
京八件296x150cm.psd
提供的源码资源涵盖了安卓应用、小程序、Python应用和Java应用等多个领域,每个领域都包含了丰富的实例和项目。这些源码都是基于各自平台的最新技术和标准编写,确保了在对应环境下能够无缝运行。同时,源码中配备了详细的注释和文档,帮助用户快速理解代码结构和实现逻辑。 适用人群: 这些源码资源特别适合大学生群体。无论你是计算机相关专业的学生,还是对其他领域编程感兴趣的学生,这些资源都能为你提供宝贵的学习和实践机会。通过学习和运行这些源码,你可以掌握各平台开发的基础知识,提升编程能力和项目实战经验。 使用场景及目标: 在学习阶段,你可以利用这些源码资源进行课程实践、课外项目或毕业设计。通过分析和运行源码,你将深入了解各平台开发的技术细节和最佳实践,逐步培养起自己的项目开发和问题解决能力。此外,在求职或创业过程中,具备跨平台开发能力的大学生将更具竞争力。 其他说明: 为了确保源码资源的可运行性和易用性,特别注意了以下几点:首先,每份源码都提供了详细的运行环境和依赖说明,确保用户能够轻松搭建起开发环境;其次,源码中的注释和文档都非常完善,方便用户快速上手和理解代码;最后,我会定期更新这些源码资源,以适应各平台技术的最新发展和市场需求。
提供的源码资源涵盖了安卓应用、小程序、Python应用和Java应用等多个领域,每个领域都包含了丰富的实例和项目。这些源码都是基于各自平台的最新技术和标准编写,确保了在对应环境下能够无缝运行。同时,源码中配备了详细的注释和文档,帮助用户快速理解代码结构和实现逻辑。 适用人群: 这些源码资源特别适合大学生群体。无论你是计算机相关专业的学生,还是对其他领域编程感兴趣的学生,这些资源都能为你提供宝贵的学习和实践机会。通过学习和运行这些源码,你可以掌握各平台开发的基础知识,提升编程能力和项目实战经验。 使用场景及目标: 在学习阶段,你可以利用这些源码资源进行课程实践、课外项目或毕业设计。通过分析和运行源码,你将深入了解各平台开发的技术细节和最佳实践,逐步培养起自己的项目开发和问题解决能力。此外,在求职或创业过程中,具备跨平台开发能力的大学生将更具竞争力。 其他说明: 为了确保源码资源的可运行性和易用性,特别注意了以下几点:首先,每份源码都提供了详细的运行环境和依赖说明,确保用户能够轻松搭建起开发环境;其次,源码中的注释和文档都非常完善,方便用户快速上手和理解代码;最后,我会定期更新这些源码资源,以适应各平台技术的最新发展和市场需求。
MySQL中的嵌套查询(也称为子查询)是指在一个查询中嵌套另一个查询。子查询可以出现在SELECT、INSERT、UPDATE或DELETE语句中,通常用于返回一个结果集,该结果集可以作为外部查询的一部分进行进一步处理. 嵌套查询可以非常强大,但它们也可能影响查询性能,特别是当子查询包含复杂的逻辑或处理大量数据时。在设计数据库查询时,应该考虑查询的效率和性能。
提供的源码资源涵盖了安卓应用、小程序、Python应用和Java应用等多个领域,每个领域都包含了丰富的实例和项目。这些源码都是基于各自平台的最新技术和标准编写,确保了在对应环境下能够无缝运行。同时,源码中配备了详细的注释和文档,帮助用户快速理解代码结构和实现逻辑。 适用人群: 这些源码资源特别适合大学生群体。无论你是计算机相关专业的学生,还是对其他领域编程感兴趣的学生,这些资源都能为你提供宝贵的学习和实践机会。通过学习和运行这些源码,你可以掌握各平台开发的基础知识,提升编程能力和项目实战经验。 使用场景及目标: 在学习阶段,你可以利用这些源码资源进行课程实践、课外项目或毕业设计。通过分析和运行源码,你将深入了解各平台开发的技术细节和最佳实践,逐步培养起自己的项目开发和问题解决能力。此外,在求职或创业过程中,具备跨平台开发能力的大学生将更具竞争力。 其他说明: 为了确保源码资源的可运行性和易用性,特别注意了以下几点:首先,每份源码都提供了详细的运行环境和依赖说明,确保用户能够轻松搭建起开发环境;其次,源码中的注释和文档都非常完善,方便用户快速上手和理解代码;最后,我会定期更新这些源码资源,以适应各平台技术的最新发展和市场需求。
数据集格式:Pascal VOC格式+YOLO格式(不包含分割路径的txt文件,仅仅包含jpg图片以及对应的VOC格式xml文件和yolo格式txt文件) 图片数量(jpg文件个数):3575 标注数量(xml文件个数):3575 标注数量(txt文件个数):3575 标注类别数:10 标注类别名称:[“Weevil”,“beetle”,“grasshopper”,“mango_hopper”,“mango_mealybug”,“moth”,“sawfly”,“slug”,“stem_borer”,“wasp”] 更多信息:https://blog.csdn.net/FL1623863129/article/details/139383493
历经二十余年的高速发展,我国保险行业的市场竞争已经达到白热化的程度,在同一个城市往往有数十家主体参与保险业务的竞争。保险企业的竞争异常激烈,想要在现在这复杂多变的市场环境中跳出来,必须好好的进行业务的管理。广大保户在享受着竞争带来的实惠同时也变得更加挑剔,哪家保险公司能提供更好的服务往往成为其最终决定的关键因素。在这种环境下,越来越多的保险公司使用管理系统来提供更好的服务。目前保险公司的业务和管理信息平台,大多依从原来的内部开发思路搭建。一旦有新的应用要求,新开发出的系统又要考虑和以前系统的单独集成。而老系统的升级又会引发大量的系统更新和检测工作。因此一个好的系统对于企业是很重要的。
本设计对国内外大棚种植中温度与湿度控制系统进行简要分析,针对大棚智能化控制系统存在的众多因素,将温度与湿度环境监测同单片机控制、上位机记录控制相互结合,提出了基于单片机的设施农业温湿度智能控制系统设计方案。系统采用模块化设计方法,利用上位机与下位机控制结合的方式。下位机实现检测各点温度值与湿度值,上位机通过VB建立一个信息管理系统。
python实验在科学计算和可视化中的应用实验.zip