Hadoop小试牛刀之dump数据

 

各位好，最近小弟接到一个需求，需要dump数据从云梯(淘宝的hadoop环境)到TAIR(淘宝的缓存系统)。

原先的大概设计：

先从云梯原始表分析出部分数据，再用Datax(数据同步工具)将这些数据导入mysql单表，再起一个单机多线程任务跑（这里需要8个多小时）。

我的一个简单想法，既然数据在云梯上，为什么不直接在云梯上直接跑MapReduce任务呢？然后我就开始这么干了。。。

 

 1.由于原始表在hive环境中，大概20亿条记录，而我感兴趣的只是其中部分数据。所以先用hivesql过滤数据，sql如下:

 

set mapred.reduce.tasks=45;//产生的hive文件个数控制
set hive.job.hooks.autored.enable=false;

create external table if not exists ecrm_buyer_benefit (
 buyer_id              string
,seller_id            string
,end_time                string
)
comment 'stat_buyer_benefit'
row format delimited
fields terminated by ','
lines terminated by '\n'
stored as TEXTFILE;

INSERT OVERWRITE TABLE ecrm_buyer_benefit
select buyer_id,seller_id,max(end_time)
from r_ecrm_buyer_benefit  b
where
b.pt=20121126000000
and
b.end_time>'2012-11-27 00:00:00'
and b.promotion_type='3'
and (b.status='1' or b.status='0')
group by buyer_id,seller_id;

 

 

说明

a.通过参数控制hivesql跑出来的文件数目，方便后续跑MapReduce任务时控制map数量，避免同时并发数太高，把TAIR写挂了

b.使用TEXTFILE类型文件，hive会默认压缩生成.deflate文件

c.查询分区表，取最大时间的记录，业务逻辑可以忽略哈

 

2.执行完之后，在表名下，我们会看到生成了45个文件，每个文件大概30M。通过./hadoop dfs -dus

  察看总大小为1.3G,共1.2亿条记录

3.编写MapReduce代码

a.业务逻辑比较简单，主要考虑异常数据，由于是build数据，所以Map主要负责build，reduce主要负责搜集异常记录，对异常数据需要做重试。同时用counter记录所有异常数据。记录异常:

private void recordError(OutputCollector<LongWritable,Text> output,LongWritable key,String line,Reporter reporter){
reporter.getCounter("counter_group", "error_counter").increment(1l);
try {
output.collect(key, new Text(line));
} catch (IOException e) {
e.printStackTrace();
System.out.println("output error,line:"+line);
}
}

 

reducer只输出异常信息：

public void reduce(LongWritable key, Iterator<Text> values,OutputCollector<LongWritable, Text> output, Reporter reporter) throws IOException {
while (values.hasNext()) {
output.collect(key, values.next());
}
}

 

运行结果1.2G条记录起了45个mapper同时跑，在TAIR 1wTPS下，运行了8个半小时。当然这个后续还有很多可以优化的，比如使用map多线程(默认单线程)等。

 

整个开发还是比较方便的，主要的问题就是如何提交第三方依赖包。这里使用maven的assemble将所有依赖的类打入同一个jar。

pom配置：

<plugins>
         <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <configuration>
          <source>1.6</source>
          <target>1.6</target>
          <encoding>UTF-8</encoding>
        </configuration>
      </plugin>

        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <archive>
                    <manifest>
                        <mainClass>com.taobao.ump.dump.DumpStarter</mainClass>
                    </manifest>
                </archive>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
        </plugin>
    </plugins>

 

执行打包：mvn clean compile assembly:single

 ，在target下生成10M的jar包，提交hadoop执行即可

 

3.小节

使用hadoop进行并行开发还是很方便的。不过我这里用来做IO任务，其实不是很合适，hadoop应该是为cpu任务设计的。