码迷，mamicode.com
首页 > 其他好文 > 详细

Hadoop 一: NCDC 数据准备

时间：2017-10-01 16:57:22      阅读：149      评论：0      收藏：0      [点我收藏+]

标签：ber   utils   byte   clojure   完成   padding   not   doc   guide   

本文介绍Hadoop- The Definitive Guide一书中的NCDC数据准备,为后面的学习构建大数据环境;

环境

3节点 Hadoop 2.7.3 集群; java version "1.8.0_111"

1 下载数据

从NCDC下载20,21世纪天历史气数据;官网按年份命名文件夹,每个文件内包含N个gz打包的(*.op.gz)全年各地区天气数据文件和一个全年天气数据打包tar文件,比如1971年;

034700-99999-1971.op.gz
035623-99999-1971.op.gz
035833-99999-1971.op.gz
035963-99999-1971.op.gz
036880-99999-1971.op.gz
040180-16201-1971.op.gz
061800-99999-1971.op.gz
080870-99999-1971.op.gz
gsod_1971.tar

*1971.op.gz就是该年的某地区某天数据打包,而*1971.tar就是对全年*.op.gz文件的打包;只需要下载tar文件,再解压即可得到全年天气数据;在这里下载从1902年到2017年tar文件;

#!/bin/bash
for i in {1902..2017}
do
    cd /home/lanstonwu/hapood/ncdc
    wget --execute robots=off -r -np -nH --cut-dirs=4 -R index.html* ftp://ftp.ncdc.noaa.gov/pub/data/gsod/$i/*.tar
done

2 上传数据

为了便于使用,文件下载完成后,推荐使用hadoop将全年的天气数据合并为一个文件;由于下载的数据保存在本地,为了使用hadoop并行处理这些数据,需要将数据上传到HDFS;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

/**
 * 将本地文件上传到hadoop集群 hdfs 提示权限不足时设置环境变量export HADOOP_USER_NAME=hadoop再运行
 * 
 * @author lanstonwu
 *
 */
public class UpLoadFile {
    public static void main(String[] args) throws IOException {
        // hdfs目录
        String target = "hdfs://192.168.56.12:9000/gsod";
        // 本地文件目录
        File file = new File("/home/lanstonwu/hapood/ncdc");
        if (file.exists()) {
            File[] files = file.listFiles();
            if (files.length == 0) {
                System.out.println("文件夹是空的!");
                return;
            } else {
                for (File file2 : files) {// 遍历本地文件目录
                    if (file2.isDirectory()) {
                        System.out.println("文件夹:" + file2.getAbsolutePath() + "," + file2.getName());
                    } else {
                        System.out.println("文件:" + file2.getAbsolutePath() + ",name:" + file2.getName());
                        // 读取本地文件
                        FileInputStream fis = new FileInputStream(new File(file2.getAbsolutePath()));
                        Configuration config = new Configuration();
                        // Returns the FileSystem for this URI‘s scheme and authority
                        FileSystem fs = FileSystem.get(URI.create(target + "/" + file2.getName()), config);
                        // Create an FSDataOutputStream at the indicated Path
                        OutputStream os = fs.create(new Path(target + "/" + file2.getName()));
                        // 复制数据
                        IOUtils.copyBytes(fis, os, 4096, true);
                        System.out.println("拷贝完成...");
                    }
                }
            }
        } else {
            System.out.println("文件不存在!");
        }
    }
}

3 合并数据

由于hadoop处理大数据文件比处理小数据文件更有优势,这里将tar文件内的全年gz打包数据合并为一个文件;因为仅仅合并数据,用map即可,无需reduce,用hadoop的streaming并行完成这个工作;首先准备处理文件清单;

$ vi ncdc_file_list.txt

hdfs://gp-sdw1:9000/gsod/gsod_1981.tar
hdfs://gp-sdw1:9000/gsod/gsod_1977.tar
hdfs://gp-sdw1:9000/gsod/gsod_1978.tar
hdfs://gp-sdw1:9000/gsod/gsod_1979.tar
hdfs://gp-sdw1:9000/gsod/gsod_1980.tar
hdfs://gp-sdw1:9000/gsod/gsod_1981.tar
hdfs://gp-sdw1:9000/gsod/gsod_1982.tar
hdfs://gp-sdw1:9000/gsod/gsod_1983.tar
.....

文件清单中记录所有要处理的文件,每一行即代表一个文件,hadoop streaming逐行读取传递给map函数处理;接着编写map脚本,每一个步骤有序号和说明;

#!/bin/bash

HADOOP_HOME=/opt/hadoop/2.7.3
cd /tmp
#1 NLineInputFormat give a signle line:offset is key,hdfile is HDFS 
read offset hdfile

#2 restrive file from hdfs
echo "reporter:status:Restrivering $hdfile" >&2
$HADOOP_HOME/bin/hadoop fs -get $hdfile .

#3 get short name from tar file
target=`basename $hdfile .tar`

#4 create directory by name of target
mkdir $target

#5 un-tar the local file to target directory
tar xvf `basename $hdfile` -C $target

#6 un-zip the local file and merge them to one file
echo "reporter:status:Un-gzipping $target" >&2
for file in $target/*
do 
    gunzip -c $file>>$target.all
    echo "repoter:status:Processed $file" >&2
done

#7 Put gzipped version into HDFS
echo "reporter:status:Gzipping $target and putting in HDFS" >&2
gzip -c $target.all | $HADOOP_HOME/bin/hadoop fs -put - /ncdc_year_gz/$target.gz

#8 remove the local file
rm -Rf $target
rm -f $target.all
rm -f $target.tar

hadoop从HDFS中读取文件到本地(第2步),获取文件名(第3步),根据获取到的文件名创建目录(第4步),解压该年的全年数据到目录里(第5步),循环解压和读取全年数据合并到一个文件里(第6步),将合并的文件压缩并上传到HDFS ncdc_year_gz目录(第7步),删除本地文件目录和文件(第8步).reporter 的目的是返回状态信息,便于监控mapper运行.注意:必须设置HADOOP_HOME变量,如果不设置该变量,所有调用hadoop的地方必须全路径,因为在运行时操作系统上配置的HADOOP_HOME变量是不可见,会导致运行报如下错误;

No such file or directory
PipeMapRed.waitOutputThreads(): subprocess failed with code 127

1.4 运行mapper 将准备好的NCDC文件清单上传到HDFS(hadoop集群节点需要);

$ hadoop fs -put ncdc_file_list.txt /

运行map;

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar   -D mapred.reduce.tasks=0   -D mapred.map.tasks.speculative.execution=false   -D mapred.task.timeout=12000000   -input /ncdc_file_list.txt   -inputformat org.apache.hadoop.mapred.lib.NLineInputFormat   -output output   -mapper load_ncdc_map.sh   -file /home/hadoop/script/load_ncdc_map.sh

禁用reduce,设置超时,设置input为准备好的ncdc清单文件,设置mapper和file为map脚本.

17/10/01 13:05:36 WARN streaming.StreamJob: -file option is deprecated, please use generic option -files instead.
packageJobJar: [/home/hadoop/script/load_ncdc_map.sh, /tmp/hadoop-unjar708897410907700502/] [] /tmp/streamjob2755689666173396550.jar tmpDir=null
17/10/01 13:05:37 INFO client.RMProxy: Connecting to ResourceManager at gp-sdw1/192.168.56.12:8032
17/10/01 13:05:37 INFO client.RMProxy: Connecting to ResourceManager at gp-sdw1/192.168.56.12:8032
17/10/01 13:05:38 INFO mapred.FileInputFormat: Total input paths to process : 1
17/10/01 13:05:38 INFO mapreduce.JobSubmitter: number of splits:114
17/10/01 13:05:38 INFO Configuration.deprecation: mapred.reduce.tasks is deprecated. Instead, use mapreduce.job.reduces
17/10/01 13:05:38 INFO Configuration.deprecation: mapred.map.tasks.speculative.execution is deprecated. Instead, use mapreduce.map.speculative
17/10/01 13:05:38 INFO Configuration.deprecation: mapred.task.timeout is deprecated. Instead, use mapreduce.task.timeout
17/10/01 13:05:38 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1506832924184_0001
17/10/01 13:05:39 INFO impl.YarnClientImpl: Submitted application application_1506832924184_0001
17/10/01 13:05:39 INFO mapreduce.Job: The url to track the job: http://gp-sdw1:8088/proxy/application_1506832924184_0001/
17/10/01 13:05:39 INFO mapreduce.Job: Running job: job_1506832924184_0001
17/10/01 13:05:46 INFO mapreduce.Job: Job job_1506832924184_0001 running in uber mode : false
17/10/01 13:05:46 INFO mapreduce.Job:  map 0% reduce 0%
17/10/01 13:06:00 INFO mapreduce.Job:  map 1% reduce 0%
17/10/01 13:06:04 INFO mapreduce.Job:  map 2% reduce 0%
17/10/01 13:06:09 INFO mapreduce.Job:  map 3% reduce 0%
17/10/01 13:06:12 INFO mapreduce.Job:  map 4% reduce 0%
17/10/01 13:06:17 INFO mapreduce.Job:  map 5% reduce 0%
17/10/01 13:06:23 INFO mapreduce.Job:  map 6% reduce 0%
17/10/01 13:06:25 INFO mapreduce.Job:  map 7% reduce 0%
17/10/01 13:06:28 INFO mapreduce.Job:  map 11% reduce 0%
17/10/01 13:06:32 INFO mapreduce.Job:  map 12% reduce 0%
17/10/01 13:06:34 INFO mapreduce.Job:  map 13% reduce 0%
17/10/01 13:06:37 INFO mapreduce.Job:  map 14% reduce 0%
17/10/01 13:06:38 INFO mapreduce.Job:  map 17% reduce 0%
17/10/01 13:06:39 INFO mapreduce.Job:  map 19% reduce 0%
17/10/01 13:06:56 INFO mapreduce.Job:  map 20% reduce 0%
17/10/01 13:07:02 INFO mapreduce.Job:  map 21% reduce 0%
17/10/01 13:07:12 INFO mapreduce.Job:  map 22% reduce 0%
17/10/01 13:07:14 INFO mapreduce.Job:  map 23% reduce 0%
17/10/01 13:07:16 INFO mapreduce.Job:  map 24% reduce 0%
17/10/01 13:07:17 INFO mapreduce.Job:  map 25% reduce 0%
17/10/01 13:07:52 INFO mapreduce.Job:  map 27% reduce 0%

技术分享
Status即为map脚本reporter返回信息;map完成,检查hadoop 合并后的文件;

$ hadoop fs -ls /ncdc_year_gz

-rw-r--r--   3 hadoop supergroup   14809707 2017-10-01 13:11 /ncdc_year_gz/gsod_1966.gz
-rw-r--r--   3 hadoop supergroup   14771822 2017-10-01 13:13 /ncdc_year_gz/gsod_1967.gz
-rw-r--r--   3 hadoop supergroup   13592592 2017-10-01 13:12 /ncdc_year_gz/gsod_1968.gz
-rw-r--r--   3 hadoop supergroup   20475061 2017-10-01 13:14 /ncdc_year_gz/gsod_1969.gz
-rw-r--r--   3 hadoop supergroup   20012492 2017-10-01 13:14 /ncdc_year_gz/gsod_1970.gz
-rw-r--r--   3 hadoop supergroup   11205341 2017-10-01 13:12 /ncdc_year_gz/gsod_1971.gz
-rw-r--r--   3 hadoop supergroup    4556815 2017-10-01 13:11 /ncdc_year_gz/gsod_1972.gz
-rw-r--r--   3 hadoop supergroup   21961972 2017-10-01 13:18 /ncdc_year_gz/gsod_1974.gz
-rw-r--r--   3 hadoop supergroup   23030229 2017-10-01 13:18 /ncdc_year_gz/gsod_1976.gz
-rw-r--r--   3 hadoop supergroup   23293175 2017-10-01 13:18 /ncdc_year_gz/gsod_1978.gz
-rw-r--r--   3 hadoop supergroup   24564712 2017-10-01 13:18 /ncdc_year_gz/gsod_1980.gz
-rw-r--r--   3 hadoop supergroup   29662599 2017-10-01 13:19 /ncdc_year_gz/gsod_1988.gz
-rw-r--r--   3 hadoop supergroup   29092407 2017-10-01 13:19 /ncdc_year_gz/gsod_1993.gz
-rw-r--r--   3 hadoop supergroup   25363736 2017-10-01 13:19 /ncdc_year_gz/gsod_1994.gz
-rw-r--r--   3 hadoop supergroup   22179093 2017-10-01 13:19 /ncdc_year_gz/gsod_1995.gz

Hadoop 一: NCDC 数据准备

标签：ber   utils   byte   clojure   完成   padding   not   doc   guide   

原文地址：http://www.cnblogs.com/lanston/p/hadoop_ncdc_data_prepare.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！