Hadoop中如何进行数据的采样和统计？代码举例讲解

发表于2023年8月17日2023年6月11日作者 IT之美

在Hadoop中进行数据采样和统计的主要方式是:

1、使用Hive的TABLESAMPLE实现采样:

对SELECT查询的结果按比例或数量进行随机采样。
支持BERNOULLI与SYSTEM采样方法。
BERNOULLI按比例进行行采样,SYSTEM按行数进行采样。
底层依赖于HDFS数据的随机读取与选择。

2、在MapReduce中自定义Sampler进行采样:

编写Sampler类实现nextSample方法进行随机行选择。
在RecordReader的read方法中调用Sampler选择是否读取该行。
这需要编写自定义的RecordReader与Sampler组件。

3、在MapReduce中统计分析全部数据:

编写Mapper解析数据并输出标签作为key,计数器作为value。
Reducer汇总所有相同标签的数据计数器。
最终输出每种标签的数量统计结果。
这需要编写Mapper、Reducer与Partitioner组件。

4、在数据库中使用SQL实现数据采样与统计:

使用TABLESAMPLE clause对查询结果进行随机行采样。
使用COUNT、SUM与GROUP BY进行数据统计与分析。
这需要数据已入库至数据库系统中,性能会有一定损失。

5、 Hive数据采样与统计示例:

创建输入表:

CREATE TABLE records (
    user_id INT, 
    action STRING  
)
ROW FORMAT DELIMITED  
FIELDS TERMINATED BY '\t';

加载数据:

LOAD DATA INPATH '/data/records、txt' INTO TABLE records;

按50%比例采样:

SELECT * FROM records TABLESAMPLE(50 PERCENT);

统计每个action的数量:

SELECT action, COUNT(*) AS count
FROM records 
GROUP BY action;

结果返回action与对应的数量。

数据采样和统计的主要作用是:
1、对大规模数据集进行随机下采样,产生具有代表性的数据样本。
2、统计不同数据维度与指标之间的频率与分布特征。
3、支持数据分析与挖掘过程中的抽样验证与效果评估。
4、帮助理解数据的概况,发现数据的结构与特点。

来看一些简单示例:
1、 Hive使用TABLESAMPLE采样与GROUP BY统计:

创建如下表:

CREATE TABLE records (
    user_id INT,  
    action STRING  
)  
ROW FORMAT DELIMITED  
FIELDS TERMINATED BY '\t';

按50%比例采样:

SELECT * FROM records TABLESAMPLE(50 PERCENT);

统计action的数量:

SELECT action, COUNT(*) AS count  
FROM records  
GROUP BY action;

2、自定义MapReduce作业采样与统计:

编写Sampler实现随机采样:

public class SampleMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
   private Sampler sampler;  

   @Override
   public void setup(Context context) {
      sampler = new Sampler(0、5);  
   }

   @Override
   public void map(LongWritable key, Text value, Context context)  
      throws IOException, InterruptedException {  
      if (sampler、nextSample()) { 
         // 随机选择50%数据进行后续处理   
      }  
   }
}

编写Reducer统计action次数:

ublic class CountReducer extends  
      Reducer<Text, IntWritable, Text, IntWritable> {
   private IntWritable result = new IntWritable();  

   @Override
   public void reduce(Text action, Iterable<IntWritable> counts, 
                     Context context)  
      throws IOException, InterruptedException {  
      int sum = 0;
      for (IntWritable count : counts) {
         sum += count、get();
      }  
      result、set(sum);
      context、write(action, result);
   }
}

设置MapReduce作业使用SampleMapper与CountReducer
运行作业实现对action进行50%采样与统计