zoukankan      html  css  js  c++  java
  • Hadoop流程---从tpch到hive

          刚接触Hadoop,看了一周的Hadoop及其相应的组件,感觉效果不是很明显,于是将找个例子练一下手,跑一个流程,加深对hadoop的理解。

          设计的流程如下:

             TPC_H--->HdFS---->MapRecude---->Hive(paration)----->DB

          即:从TPC_H生产10G的数据,将数据上传到HDFS中,编写MapReduce函数,对数据进行处理,将处理后的数据存放在Hive中,在Hive中仿照TPC_H的22条sql数据,写22条相应的HQL,并将结果保存在db中。

          下面将逐步完成上述的流程

    一、通过TPC_H生产相应的数据

       1. 什么是TPC_H

            TPC-H(商业智能计算测试)是TPC的重要测试标准之一,主要用来模拟真实商业的应用环境。 
            TPC-H 用 3NF 实现了一个数据仓库,共包含 8 个基本关系/表,其中表REGION和表NATION的记录数是固定的(分别为5和25),其它6个表的记录数,则随所设定的参数SF而有所不同,其数据量可以设定从 1GB~3TB 不等。有8个级别供用户选择。 

      2.从TPC_H导出数据

           1) 首先从TPC_H的官方网站下载最新版本的TPC_H压缩包,下载地址为 http://www.tpc.org/tpch/ 。注意下面的操作我是在Linux环境下完成的,在win下也可以,部分设置不一样而已。

           2) 解压下载的压缩文件,会有两个文件,1个是dbgen 1个是ref_data

           3) 在dbgen目录下,将makefile.sute文件复制,并将文件名修改为makefile.接着修改这个makefile文件按的部分参数

           shell>cp makefile.sute makefile

           shell>vim makefile

           具体改的参数有:

              cc = gcc

               ……

               DATABASE = SQLSERVER(也可以选择其他类型的DB)

               MACHINE = LINUX(选择运行的平台)

              WORKLOAD = TPCH

          4) 修改完成后,保存这个文件,然后在命令行进入dbgen的目录下,输入“make”就可以调用make解释器去执行你修改后的makefile了。

          shell>make

         5)  运行dbgen来生产tbl文件。

           shell>./dbgen -s 10

           “./”是一个路径,“./dbgen”表示在当前路径下执行dbgen,其后是参数.简单来说,我们可以替换的参数有(1-s后面的数字,10代表生产10倍的数据(即10G);

         6) 结束后,可以发现在当前的目录下,生成了8个*.tbl文件,总大小为10G

         注:TPC_H不是此次的重点,再次只是简单的介绍一下,如果需要详细的了解,请查看官方网站相关说明,再次有一篇比较入门的介绍TPC_H(以及导出数据)的文章,推荐给大家。http://blog.csdn.net/czpthegreat/article/details/6303071  

    二、将生成的数据从本地复制到HDFS中。

         在本地中,我们将TPC_H生成的数据放在如下目录中,

           /share/bigdata/tpch_data/data/

         这一步操作简单,只需使用如下命令即可

    hadoop fs -put /share/bigdata/tpch_data/data /user/root/tpch

         在本地中,由于目标的文件都很大,比如lineitem.tbl文件有7个多G的数据,打开这么大的文件显然是很费时,而且也没有必要使用这么大的文件来检验mapreduce函数的正确性(这才是主要的原因),在本例中,我们取其中的一部分作为测试数据,检验我们mapreduce函数的正确性。我写了一个shell脚本,通过此脚本我们可以指定要处理的文件,并指定需要截取的行,并将结果输出到指定的文件中。

       shell脚本为:

             

    #!/bin/bash
    printf "Please input the path of file which need to read: "
    read fileName
    printf "Please input the path of file which saves the result: "
    read outPut
    printf "startRow: "
    read startRow
    printf "endRow:"
    read endRow
    #sed -n "$startRow","$endRow"P /share/bigdata/tpch_data/data/$fileName
    sed -n ""$startRow","$endRow"w $outPut" /share/bigdata/tpch_data/data/$fileName

        生成的测试文件放在如下目录中:

         /share/bigdata/tpch_data//data/analytics

        我们也将这些文件复制至hdfs中:

    hadoop fs -put /share/bigdata/tpch_data/data /user/root/tpch

    三、编写并运行MapReduce函数

        本地中我们只是简单的对lineitem.tbl文件中的quantity列进行处理。选取quantity值>30的行。具体代码如下,由于代码比较简单,因此再次我就不介绍了

        

    FilterItemLine.java(这个名字没取好,应该用FilterLineItem)

    package uap.pub.hadoop.tpch.lineitem;
    
    import java.io.IOException;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    import org.apache.hadoop.util.GenericOptionsParser;
    
    public class FilterItemLine {
    
    	/**
    	 * @param args
    	 * @throws IOException
    	 * @throws ClassNotFoundException
    	 * @throws InterruptedException
    	 */
    	public static void main(String[] args) throws IOException,
    			InterruptedException, ClassNotFoundException {
    		Configuration conf = new Configuration();
    //		conf.set("mapred.job.tracker", "192.168.133.101:9001");// 全局
    		String[] otherArgs = new GenericOptionsParser(conf, args)
    				.getRemainingArgs();
    		if (otherArgs.length != 2) {
    			System.err.println("Usage: wordcount <in> <out>");
    			System.exit(2);
    		}
    		Job job = new Job(conf, "Filter ItemLine");
    		job.setJarByClass(FilterItemLine.class);
    		job.setMapperClass(FilterItemLineMapper.class);
    		job.setReducerClass(FilterItemLineReducer.class);
    
    		job.setInputFormatClass(TextInputFormat.class);
    		job.setOutputFormatClass(TextOutputFormat.class);
    
    		job.setMapOutputKeyClass(LongWritable.class);//Map端key输出的格式
    		job.setMapOutputValueClass(Text.class);//Map端value输出的格式
    		job.setOutputKeyClass(Text.class);//Reduce端key输出的格式
    		job.setOutputValueClass(Text.class);//Reduce端value输出的格式
    
    		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    
    		System.exit(job.waitForCompletion(true) ? 0 : 1);
    	}
    
    }
    

      FilterItemLineMapper .java

    package uap.pub.hadoop.tpch.lineitem;
    
    import java.io.IOException;
    import java.util.StringTokenizer;
    
    import org.apache.commons.lang.StringUtils;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    public class FilterItemLineMapper extends
    		Mapper<LongWritable, Text, LongWritable, Text> {
    
    	protected void map(LongWritable key, Text value, Context context)
    			throws IOException, InterruptedException {
    		String line = value.toString();
    		// 获取quantity字段
    		StringTokenizer st = new StringTokenizer(line, "|");
    		int i = 0;
    		String str_quantity = null;
    		while (st.hasMoreTokens()) {
    			String nextToken = st.nextToken();
    			if (i == 4) {
    				str_quantity = nextToken;
    				break;
    			}
    			i++;
    		}
    		if (!StringUtils.isEmpty(str_quantity)) {
    			int quantity = Integer.parseInt(str_quantity);
    			if (quantity >= 30) {
    				context.write(key, value);
    			}
    		}
    
    	}
    }
    

    FilterItemLineReducer.java  

    package uap.pub.hadoop.tpch.lineitem;
    
    import java.io.IOException;
    import java.util.Iterator;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class FilterItemLineReducer extends
            Reducer<LongWritable, Text, Text, Text> {
    
        protected void reduce(LongWritable key, Iterable<Text> values,
                Context context) throws IOException, InterruptedException {
            Iterator<Text> iterator = values.iterator();
            while (iterator.hasNext()) {
                Text value = iterator.next();
                context.write(null, value);//本地情况特殊,不需要对数据进行额外的处理,一般情况下,key不设为null
            }
        }
    }

     代码写好了,下面就是编译代码,打成jar包,运行代码

       1)编译代码

              javac -classpath /share/hadoop/hadoop-core-1.0.4.jar:/share/hadoop/lib/commons-lang-2.4.jar -d . FilterItemLineMapper.java 

              javac -classpath /share/hadoop/hadoop-core-1.0.4.jar:/share/hadoop/lib/commons-lang-2.4.jar -d . FilterItemLineReducer.java

             javac -classpath ./:/share/hadoop/hadoop-core-1.0.4.jar:/share/hadoop/lib/* -d . FilterItemLine.java

        2)打成jar包

         jar -cvf FilterLineItem.jar -C  java/ .

       3) 运行mapreduce

        hadoop jar /share/bigdata/java/FilterLineItem.jar uap.pub.hadoop.tpch.lineitem.FilterItemLine /user/root/tpch/data/lineitem.tbl /user/root/tpch/fileterlineitem_out  

     注意:运行mapreduce方法时,其中间数据存放的位置为:/share/hadooptmp/mapred/local/taskTracker/root/jobcache/

      

     四、将数据导入到hive中

          TPC_H生成的数据有8整个tbl文件,我们将这8个文件导入到hive中去

          在hive中表的创建有外部表和内部表,这两个表的区别想必大家都知道,在此处我主要记录自己在创建表的过程中遇到的问题

          1) 创建内部表    

    CREATE TABLE IF NOT EXISTS REGION(NATIONKEY INT,NAME STRING,COMMENT1 STRING) COMMENT 'The file of TPCH  is region.tbl' 
        > ROW FORMAT DELIMITED                                                                                                   
        > FIELDS TERMINATED BY '|'                                                                                               
        > STORED AS TEXTFILE;   
       # > LOCATION '/user/root/hive/data/nation';不要使用LOCATION
    load data:
    LOAD DATA INPATH '/user/root/tpch/data/region.tbl'INTO TABLE REGION;//After load data,we cannot find region.tbl in the directory '/user/root/tpch/data/region.tbl',on the contray,we find it in the directory '/user/hive/warehouse/region/'

    在建内部表时,不需要使用LOCATION 关键字,如果使用了,则创建后的表的数据(实际的文件)就会放到此目录下,而不是默认的/user/hive/warehouse/目录下

        

         2)创建外部表

         

    hive> CREATE EXTERNAL TABLE IF NOT EXISTS nation (NATIONKEY STRING,NAME STRING,REGIONKEY STRING ,NATION_COMMENT STRING)
        > ROW FORMAT DELIMITED FIELDS TERMINATED BY '|'                                                                    
        > STORED AS TEXTFILE                                                                                               
        > LOCATION '/user/root/tpch/data/nation';一定是个目录,不能是具体的文件
    如果创建的是一个外部表,那么就不需要load数据,否则也会将数据移到hive中,但是与内部表不同的时,此时drop表时,数据并不会被删除,只删除了元数据。
       但是使用外部表的时候,需要指定需要链接的hdfs文件的路径(即hive源数据的地址)。
    从上述的描述来看,LOCATION针对内部表和外部表的处理时不一样的,内部表是,是将数据移到LOCATION指定的位置,而对于外部表而言,LOCATION指定的是需要链接的数据地址。

     除此之外,LOCATION 一定是个目录,不能是个具体的文件,比如下面的例子就报错了

    hive> CREATE EXTERNAL TABLE IF NOT EXISTS nation (NATIONKEY STRING,NAME STRING,REGIONKEY STRING ,NATION_COMMENT STRING)
        > ROW FORMAT DELIMITED FIELDS TERMINATED BY '|'                                                                    
        > STORED AS TEXTFILE                                                                                               
        > LOCATION '/user/root/tpch/data/nation.tbl';//wrong
    FAILED: Error in metadata: MetaException(message:Got exception: org.apache.hadoop.ipc.RemoteException java.io.FileNotFoundException: Parent path is not a directory: /user/root/tpch/data/nation.tbl
    at org.apache.hadoop.hdfs.server.namenode.FSDirectory.mkdirs(FSDirectory.java:949)
    at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.mkdirsInternal(FSNamesystem.java:2068)
    at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.mkdirs(FSNamesystem.java:2029)
    at org.apache.hadoop.hdfs.server.namenode.NameNode.mkdirs(NameNode.java:817)
    at sun.reflect.GeneratedMethodAccessor24.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:563)
    at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1388)
    at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1384)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1121)
    at org.apache.hadoop.ipc.Server$Handler.run(Server.java:1382)
    )

    3)创建分区表

       

  • 相关阅读:
    json
    网页版 treeview使用中遇到的问题
    随机获取一条数据
    oracle
    发送邮件
    DataGrid列的合并
    python简介
    SQLSERVER
    Maven 基础
    Maven 构建jar包
  • 原文地址:https://www.cnblogs.com/chenfei0801/p/3383894.html
Copyright © 2011-2022 走看看