javafu
- 浏览: 58259 次
- 性别:
- 来自: 北京
-
文章分类
最新评论
统计高并发UV可行的方案（类似WordCount的计算去重word总数）：
bolt1通过fieldGrouping 进行多线程局部汇总，下一级blot2进行单线程保存session_id和count数到Map且进行遍历，可以得到：
Pv、UV、访问深度（按每个session_id 的浏览数）
2014-05-01 UV数（按日期统计）
既然去重，必须持久化。两种持久化数据：
1、内存（适用中小型数据）
数据结构Map
2、no-sql 分布式数据库，如Hbase（适用大型数据）
1、数据源
2、日期格式化处理类
3、多线程局部汇总深度数据
4、单线程汇总数据
5、topoly类
6、pom.xml文件
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.test</groupId>
<artifactId>StormMavenProject</artifactId>
<packaging>jar</packaging>
<version>0.0.1-SNAPSHOT</version>
<name>StormMavenProject</name>
<url>http://maven.apache.org</url>
<dependencies>
<dependency>
<groupId>org.ow2.asm</groupId>
<artifactId>asm</artifactId>
<version>5.0.3</version>
</dependency>
<dependency>
<groupId>org.clojure</groupId>
<artifactId>clojure</artifactId>
<version>1.7.0</version>
</dependency>
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.3.2</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
<version>3.0.3</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.8</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>log4j-over-slf4j</artifactId>
<version>1.6.6</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.8</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>minlog</artifactId>
<version>1.3.0</version>
</dependency>
<dependency>
<groupId>org.objenesis</groupId>
<artifactId>objenesis</artifactId>
<version>2.1</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>reflectasm</artifactId>
<version>1.10.1</version>
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>servlet-api</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.1.0</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-rename-hack</artifactId>
<version>1.1.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>ring-cors</groupId>
<artifactId>ring-cors</artifactId>
<version>0.1.5</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.10</version>
</dependency>
</dependencies>
<build>
<finalName>StormMavenProject</finalName>
</build>
</project>
7、日期处理类
8、测试结果
bolt1通过fieldGrouping 进行多线程局部汇总，下一级blot2进行单线程保存session_id和count数到Map且进行遍历，可以得到：
Pv、UV、访问深度（按每个session_id 的浏览数）
2014-05-01 UV数（按日期统计）
既然去重，必须持久化。两种持久化数据：
1、内存（适用中小型数据）
数据结构Map
2、no-sql 分布式数据库，如Hbase（适用大型数据）
1、数据源
public class SourceSpout implements IRichSpout{
/**
* 数据源Spout
*/
private static final long serialVersionUID = 1L;
Queue<String> queue = new ConcurrentLinkedQueue<String>();
SpoutOutputCollector collector = null;
String str = null;
public void nextTuple() {
if (queue.size() >= 0) {
collector.emit(new Values(queue.poll()));
}
try {
Thread.sleep(500) ;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
try {
this.collector = collector;
Random random = new Random();
String[] hosts = { "www.taobao.com" };
String[] session_id = { "ABYH6Y4V4SCVXTG6DPB4VH9U123", "XXYH6YCGFJYERTT834R52FDXV9U34", "BBYH61456FGHHJ7JL89RG5VV9UYU7",
"CYYH6Y2345GHI899OFG4V9U567", "VVVYH6Y4V4SFXZ56JIPDPB4V678" };
String[] time = { "2014-01-07 08:40:50", "2014-01-07 08:40:51", "2014-01-07 08:40:52", "2014-01-07 08:40:53",
"2014-01-07 09:40:49", "2014-01-07 10:40:49", "2014-01-07 11:40:49", "2014-01-07 12:40:49" };
for (int i = 0; i < 20; i++) {
queue.add(hosts[0]+"\t"+session_id[random.nextInt(5)]+"\t"+time[random.nextInt(8)]);
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void close() {
// TODO Auto-generated method stub
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
declarer.declare(new Fields("log"));
}
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
public void ack(Object msgId) {
// TODO Auto-generated method stub
System.out.println("spout ack:"+msgId.toString());
}
public void activate() {
// TODO Auto-generated method stub
}
public void deactivate() {
// TODO Auto-generated method stub
}
public void fail(Object msgId) {
// TODO Auto-generated method stub
System.out.println("spout fail:"+msgId.toString());
}
}
2、日期格式化处理类
public class FmtLogBolt implements IBasicBolt{
/**
* 格式化日期
*/
private static final long serialVersionUID = 1L;
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("date","session_id"));
}
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
public void prepare(Map stormConf, TopologyContext context) {
// TODO Auto-generated method stub
}
String eachLog = null;
public void execute(Tuple input, BasicOutputCollector collector) {
eachLog=input.getStringByField("log");
if (eachLog != null && eachLog.length() > 0 ) {
collector.emit(new Values(DateFmt.getCountDate(eachLog.split("\t")[2],DateFmt.date_short),eachLog.split("\t")[1])) ;// 日期, session_id
}
}
public void cleanup() {
// TODO Auto-generated method stub
}
}
3、多线程局部汇总深度数据
public class DeepVisitBolt implements IBasicBolt{
/**
* 多线程局部汇总深度数据
*/
private static final long serialVersionUID = 1L;
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("date_session_id","count"));
}
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
public void prepare(Map stormConf, TopologyContext context) {
// TODO Auto-generated method stub
}
Map<String, Integer> counts = new HashMap<String, Integer>();
public void execute(Tuple input, BasicOutputCollector collector) {
String dateString =input.getStringByField("date");
String session_id = input.getStringByField("session_id");
Integer count = counts.get(dateString+"_"+session_id);
if (count == null) {
count = 0;
}
count ++ ;
counts.put(dateString+"_"+session_id,count) ;
collector.emit(new Values(dateString+"_"+session_id,count)) ;
}
public void cleanup() {
// TODO Auto-generated method stub
};
}
4、单线程汇总数据
public class UVSumBolt implements IBasicBolt{
/**
* 单线程汇总数据
*/
private static final long serialVersionUID = 1L;
Map<String, Integer> counts = new HashMap<String, Integer>();
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
}
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
public void prepare(Map stormConf, TopologyContext context) {
cur_date = DateFmt.getCountDate("2014-01-07", DateFmt.date_short);
}
long beginTime = System.currentTimeMillis() ;
long endTime = 0;
String cur_date = null;
public void execute(Tuple input, BasicOutputCollector collector) {
try {
endTime = System.currentTimeMillis() ;
long PV = 0;// 总数
long UV = 0; // 个数，去重后
String dateSession_id = input.getString(0);
Integer count = input.getInteger(1);
//清空不是当天的数据
if (!dateSession_id.startsWith(cur_date)
&& DateFmt.parseDate(dateSession_id.split("_")[0]).after(
DateFmt.parseDate(cur_date))) {
cur_date = dateSession_id.split("_")[0];
counts.clear();
}
counts.put(dateSession_id, count);
if (endTime - beginTime >= 2000) {//两秒输出一次
// 获取word去重个数，遍历counts 的keySet，取count
Iterator<String> i2 = counts.keySet().iterator();
while (i2.hasNext()) {
String key = i2.next();
if (key != null) {
if (key.startsWith(cur_date)) {
UV++;
PV += counts.get(key);
}
}
}
System.err.println("PV=" + PV + "; UV="+ UV);
}
} catch (Exception e) {
throw new FailedException("SumBolt fail!");
}
}
public void cleanup() {
// TODO Auto-generated method stub
}
}
5、topoly类
public class UVTopo {
/**
* topoly类
*/
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new SourceSpout(), 1);
builder.setBolt("FmtLogBolt", new FmtLogBolt(), 4).shuffleGrouping("spout");
// Fields Grouping：按Field分组，比如按word来分组， 具有同样word的tuple会被分到相同的Bolts， 而不同的word则会被分配到不同的Bolts。
builder.setBolt("sumBolt", new DeepVisitBolt(),4).fieldsGrouping("FmtLogBolt", new Fields("date","session_id"));
builder.setBolt("UvSum", new UVSumBolt(), 1).shuffleGrouping("sumBolt") ;
Config conf = new Config() ;
conf.setDebug(true);
if (args.length > 0) {
try {
StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
} catch (AlreadyAliveException e) {
e.printStackTrace();
} catch (InvalidTopologyException e) {
e.printStackTrace();
} catch (AuthorizationException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}else {
LocalCluster localCluster = new LocalCluster();
localCluster.submitTopology("mytopology", conf, builder.createTopology());
}
}
}
6、pom.xml文件
引用
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.test</groupId>
<artifactId>StormMavenProject</artifactId>
<packaging>jar</packaging>
<version>0.0.1-SNAPSHOT</version>
<name>StormMavenProject</name>
<url>http://maven.apache.org</url>
<dependencies>
<dependency>
<groupId>org.ow2.asm</groupId>
<artifactId>asm</artifactId>
<version>5.0.3</version>
</dependency>
<dependency>
<groupId>org.clojure</groupId>
<artifactId>clojure</artifactId>
<version>1.7.0</version>
</dependency>
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.3.2</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
<version>3.0.3</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.8</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>log4j-over-slf4j</artifactId>
<version>1.6.6</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.8</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>minlog</artifactId>
<version>1.3.0</version>
</dependency>
<dependency>
<groupId>org.objenesis</groupId>
<artifactId>objenesis</artifactId>
<version>2.1</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>reflectasm</artifactId>
<version>1.10.1</version>
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>servlet-api</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.1.0</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-rename-hack</artifactId>
<version>1.1.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>ring-cors</groupId>
<artifactId>ring-cors</artifactId>
<version>0.1.5</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.10</version>
</dependency>
</dependencies>
<build>
<finalName>StormMavenProject</finalName>
</build>
</project>
7、日期处理类
public class DateFmt {
/*
* 日期处理类
*/
public static final String date_long = "yyyy-MM-dd HH:mm:ss" ;
public static final String date_short = "yyyy-MM-dd" ;
public static SimpleDateFormat sdf = new SimpleDateFormat(date_short);
public static String getCountDate(String date,String patton)
{
SimpleDateFormat sdf = new SimpleDateFormat(patton);
Calendar cal = Calendar.getInstance();
if (date != null) {
try {
cal.setTime(sdf.parse(date)) ;
} catch (ParseException e) {
e.printStackTrace();
}
}
return sdf.format(cal.getTime());
}
public static Date parseDate(String dateStr) throws Exception
{
return sdf.parse(dateStr);
}
public static void main(String[] args) throws Exception{
// System.out.println(DateFmt.getCountDate("2014-03-01 12:13:14", DateFmt.date_short));
System.out.println(parseDate("2014-05-02").after(parseDate("2014-05-01")));
}
}
8、测试结果
- 大小: 191.2 KB
分享到：
发表评论
-
ITridentSpout、FirstN(取Top N)实现、 流合并和join
2017-05-25 10:01 1052一、ITridentSpout 基于事务 static int ... -
Trident实战之计算网站PV
2017-05-24 13:24 6611、Trident实战之计算网站PV /** * ... -
Trident API和概念
2017-05-23 10:57 773一、Trident API——Spout ITride ... -
Trident入门
2017-05-22 13:44 534英文原址：https://github.com/nathanm ... -
分布式远程调用drpc实例
2017-05-22 10:53 430一、DRPC定义 分布式dRPC（distributed RP ... -
不透明分区事务IOpaquePartitionedTransactional实例
2017-05-22 10:54 6901、spout public class MyOpaq ... -
分区事务IPartitionedTransactionalSpout实例
2017-05-21 11:02 5961.分区事务spout public class My ... -
普通事务ITransactionalSpout实例之按天统计数据
2017-05-20 16:56 5001、普通事务Spout /** * 普通事务Spou ... -
普通事务ITransactionalSpout实例
2017-05-20 15:45 8351、普通事务Spout /** * 普通事务Spou ... -
Storm事务API
2017-05-19 16:00 625Spout ITransactionalSpout<T& ... -
Storm批处理事务原理详解
2017-05-19 15:54 2138事务：Storm容错机制通 ... -
集群统一启动和停止shell脚本开发
2017-05-17 09:56 4711、cd 2、ls -al 显示隐藏目录 3、rm -rf ... -
storm高并发PV统计，利用zookeeper锁输出汇总值
2017-05-14 14:42 917汇总型方案： 1、shuffleGrouping下，pv(单线 ... -
storm高并发PV统计
2017-04-16 17:54 710一、PV统计思考 方案需要考虑分析多线程下，注意线程安全问题。 ... -
Storm高并发运用WordSum
2017-04-16 14:21 10851、创建发射所有字符串统计总个数及去重个数处理类 pub ... -
storm分组策略介绍
2017-04-16 11:46 713一、storm数据来源 Spout的数据源： MQ：直接流数 ... -
Storm高并发介绍
2017-04-16 10:18 620并发度： worker：指的是component (spo ... -
Storm 字符统计Demo
2017-04-14 13:57 5451、数据源读取，字符发射spout类 /** * 字符 ... -
Storm 本地模式
2017-04-09 22:25 407本地模式，是在eclipse等编译器编写strom运行文件 ... -
Storm启动配置
2017-03-29 17:40 686一、安装Storm wget ...
相关推荐
- 除了基础的Apache Storm，课程中还提到了JStorm，它是阿里巴巴开源的一个高性能、高稳定性的Storm版本，针对大规模集群和高并发场景进行了优化。 6. **HBase存储与State运用**： - 在项目1中，我们使用HBase...
项目的实施过程中，需要注意的问题包括如何高效地进行数据聚合，如何确保在高并发下的数据一致性，以及如何优化HTTP长连接以保证实时数据推送的稳定性和效率。通过这种方式，1号店电商实时数据分析系统能够提供对非...
数据批处理平台包含不同类型的Job，例如基础类Url处理、统计分布类PV/UV计算、安全分析类漏洞分析等。流数据处理平台需要具备横向扩展、负载均衡、错误恢复、任务迁移、集中控制和引擎异构等特点。 4. 安全服务协作...
轴类零件加工工艺设计.zip
资源内项目源码是来自个人的毕业设计，代码都测试ok，包含源码、数据集、可视化页面和部署说明，可产生核心指标曲线图、混淆矩阵、F1分数曲线、精确率-召回率曲线、验证集预测结果、标签分布图。都是运行成功后才上传资源，毕设答辩评审绝对信服的保底85分以上，放心下载使用，拿来就能用。包含源码、数据集、可视化页面和部署说明一站式服务，拿来就能用的绝对好资源！！！ 项目备注 1、该资源内项目代码都经过测试运行成功，功能ok的情况下才上传的，请放心下载使用！ 2、本项目适合计算机相关专业(如计科、人工智能、通信工程、自动化、电子信息等)的在校学生、老师或者企业员工下载学习，也适合小白学习进阶，当然也可作为毕设项目、课程设计、大作业、项目初期立项演示等。 3、如果基础还行，也可在此代码基础上进行修改，以实现其他功能，也可用于毕设、课设、作业等。 下载后请首先打开README.txt文件，仅供学习参考, 切勿用于商业用途。
seaborn基本绘图人力资源数据集
移动机器人(sw三维）
自制html网页源代码查看器
3吨叉车的液压系统设计().zip
1_实验三 扰码、卷积编码及交织.ppt
北京交通大学软件学院自命题科目考试大纲.pdf
雅鲁藏布江流域 shp矢量数据 (范围＋DEM).zip
基于RUST的数据结构代码示例，栈、队列、图等
NIFD：2024Q1房地产金融报告
详细介绍及样例数据：https://blog.csdn.net/li514006030/article/details/146916652
【工业机器视觉定位软件Vision-Detect】基于C#的WPF与Halcon开发的工业机器视觉定位软件（整套源码），开箱即用 有用户登录，图片加载，模板创建，通讯工具，抓边抓圆，良率统计，LOG日志，异常管理，九点标定和流程加载保存等模块，功能不是很完善，适合初学者参考学习。 资源介绍请查阅：https://blog.csdn.net/m0_37302966/article/details/146912206 更多视觉框架资源：https://blog.csdn.net/m0_37302966/article/details/146583453
内容概要：本文档详细介绍了Java虚拟机（JVM）的相关知识点，涵盖Java内存模型、垃圾回收机制及算法、垃圾收集器、内存分配策略、虚拟机类加载机制和JVM调优等内容。首先阐述了Java代码的编译和运行过程，以及JVM的基本组成部分及其运行流程。接着深入探讨了JVM的各个运行时数据区，如程序计数器、Java虚拟机栈、本地方法栈、Java堆、方法区等的作用和特点。随后，文档详细解析了垃圾回收机制，包括GC的概念、工作原理、优点和缺点，并介绍了几种常见的垃圾回收算法。此外，文档还讲解了JVM的分代收集策略，新生代和老年代的区别，以及不同垃圾收集器的工作方式。最后，文档介绍了类加载机制、JVM调优的方法和工具，以及常用的JVM调优参数。 适合人群：具备一定Java编程基础的研发人员，尤其是希望深入了解JVM内部机制、优化程序性能的技术人员。 使用场景及目标：①帮助开发人员理解Java代码的编译和执行过程；②掌握JVM内存管理机制，包括内存分配、垃圾回收等；③熟悉类加载机制，了解类加载器的工作原理；④学会使用JVM调优工具，掌握常用调优参数，提升应用程序性能。 其他说明：本文档内容详尽，适合用作面试准备材料和技术学习资料，有助于提高开发人员对JVM的理解和应用能力。
Android项目原生java语言课程设计，包含LW+ppt
戴德梁行&中国房地产协会：2021亚洲房地产投资信托基金研究报告
Android项目原生java语言课程设计，包含LW+ppt