一个千万级数据的统计方法尝试

现在，我们有一个文件，样子长成这个样子的：

该数据共有两列数据，col1，col2.col1是字符串类型的，col2为数字类型。这样的数据一共有多少呢？

一共有：25165824。

 

现在我们的目标是统计col1中每个值出现的次数，并把对应的col2的值加起来，并且得到平均值。

这样是放在关系数据库里，用SQL的话，十分容易搞定，SQL如下：

 

 

select col1                        name
         ,count(1)                 count
         ,sum(col2)               sum
         ,sum(col2)/count(1) avg
from   table_t
group by col1;

 

就这么简单。

 

然而假如没有数据库，怎么办呢？

 

其实解决方案很多，本文尝试使用一种最简单的，不需要使用任何第三放的软件或者代码支持的方法搞定这个问题。

面对这种格式固定的流式数据，最好的统计方法就是awk。（在求count的问题上，作者也曾尝试使用sort | uniq -c 的方法，但数据量太大，sort太慢，最终放弃。）

 

方案一：

直接扫描25165824条数据，逻辑如下：

BEGIN{
    FS=",";
}
{
    if($1 in count){
        count[$1]+=1;
    }
    else{
        count[$1] =1;
    }
    if($1 in sum){
        sum[$1] +=$2;
    }
    else{
        sum[$1] = $2;
    }
}
END{
    for(x in count){
        print x":"count[x]":"sum[x];
    }
} 

简单吧，逐行扫描全部数据，分别统计count、和sum值。最后求avg就是一个除法的问题了，此处不讨论。

改方案在本人thinkpad r61i上耗时18s～20s。

 

方案二：

 

将全部数据分成不同的分，分别计算每份数据结果，然后进行汇总求值。呵呵，思想有点类似map/reduce。只不过我就一台电脑哈。

在该方案中，作者尝试了不同的划分方法。

分别将数据每10000000、5000000、3000000分成一份。

然后分别进行统计，在全部统计之后，将结果汇总起来。

在计算正确的前提下，耗时分别为13s、11s、10s (经过多次测试之后的平均值，在测试时，基本完全消耗一核CPU。由于是双核系统，cpu load基本上在0.9～1.1之间。)

 

显然，该方案比直接跑全部数据要来的快。基本上减少了6～7s的时间。但也发现，在不同的数据分割方案之间，时间相差不大。可能是计算任务已经占满了CPU，在该条件下已经极限了吧。

 

下面是计算逻辑，以每5000000条数据为一份举例：

 

首先是整体计算逻辑：

cal.sh

#!/bin/bash

rm a b c d e f

awk -f map.awk taa > a &
awk -f map.awk tab > b &
awk -f map.awk tac > c &
awk -f map.awk tad > d &
awk -f map.awk tae > e &
awk -f map.awk taf > f &

while true
do
    if [ -s a ] && [ -s b ] && [ -s c ] && [ -s d ] && [ -s e ] && [ -s f ] ; then
        break;
    fi
done

cat a>>f
cat b>>f
cat c>>f
cat d>>f
cat e>>f


awk -f reduce.awk f

通过代码，不难理解其中的逻辑吧。

 

其次是用来统计每份数据的map.awk文件：

map.awk:      

BEGIN{
    FS=",";
}
{
    if($1 in count){
        count[$1]+=1;
    }
    else{
        count[$1] =1;
    }
    if($1 in sum){
        sum[$1] +=$2;
    }
    else{
        sum[$1] = $2;
    }
}
END{
    for(x in count){
        print x":"count[x]":"sum[x];
    }
}

该逻辑与之前全部数据扫描的逻辑一致，只是每次面对的数据不同而已。

 

最后是用于将每份数据汇总起来，计算最终结果的逻辑：

reduce.awk:

BEGIN{
    FS=":";
    printf("%20s%20s%20s%20s","col1","count","sum","avg");
}
{
    if($1 in count){
        count[$1] += $2;
    }
    else{
        count[$1]  = $2;
    }
    if($1 in sum){
        sum[$1] += $3;
    }
    else{
        sum[$1]  = $3;
    }
}
END{
    for(x in count){
        printf("%20s%20i%20i%20.2f\n",x,count[x],sum[x],sum[x]/count[x]);
    }
}

其实将每份数据汇总到一起的逻辑是cal.sh中的那一堆"cat >>"，该reduce.awk只是将每份数据计算的结果汇总，计算并打印最终结果，最终结果如下：

 

 

总结：

 

本文尝试使用awk解决数据统计的问题，通过切分数据充分利用cpu资源进行数据统计。

如果该任务跑在分布式存储系统上，并通过真正的map/reduce进行并行计算，我想想过会更好的吧。

 

呵呵，敬请期待，探索还在继续...