hive 用户手册和map参数调整，hivemap

hive 用户手册和map参数调整，hivemap

map/reduce简单的原理介绍

Hadoop Map/Reduce框架为每一个InputSplit产生一个map任务，而每个InputSplit是由该作业的InputFormat产生的。

然后，框架为这个任务的InputSplit中每个键值对调用一次 map(WritableComparable, Writable, OutputCollector, Reporter)操作。
通过调用 OutputCollector.collect(WritableComparable,Writable)可以收集输出的键值对。

Reduce会接收到不同map任务传来的数据，并且每个map传来的数据都是有序的。如果reduce端接受的数据量相当小，
则直接存储在内存中（缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制，表示用作此用途的堆空间的百分比），
如果数据量超过了该缓冲区大小的一定比例（由mapred.job.shuffle.merge.percent决定），则对数据合并后溢写到磁盘中。

Hive外部表，编写正则表达式解析Nginx日志

nginx日志的format配置：

'$proxy_add_x_forwarded_for - $remote_user [$time_local] "$request"'
'$status $request_body "$http_referer"'
'"$http_user_agent" "$http_x_forwarded_for" $request_time  $upstream_response_time';
生成的日志大致格式：
218.202.xxx.xxx – - [2014-08-19 22:17:08.446671] “POST /xxx/xxx-web/xxx HTTP/1.1″ 200 stepid=15&tid=U753HifVPE0DAOn%2F&output=json&language=zh_CN&session=114099628&dip=10920&diu=DBDBF926-3210-4D64-972A7&xxx=056a849c70ae57560440ebe&diu2=2DFDB167-1505-4372-AAB5-99D28868DCB5&shell=e3209006950686f6e65352c3205004150504c450000000000000000000000000000&compress=false&channel=&sign=438BD4D701A960CD4B7C1DE36AA8A877&wua=0&appkey=0&adcode=150700&t=0 HTTP/1.1″ 200 302 “-” “xxx-iphone” 31.0ms

hive建表的语句：
CREATE EXTERNAL TABLE xxx_log(
host STRING,
log_date STRING,
method STRING,
uri STRING,
version STRING,
STATUS STRING,
flux STRING,
referer STRING,
user_agent STRING,
reponse_time STRING
)
PARTITIONED BY(YEAR STRING, MONTH STRING, DAY STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'
WITH SERDEPROPERTIES ( "input.regex" = "([^ ]*)\\s+-\\s+-\\s+\\[([^\]]*)\\]\\s+\"([^ ]*)\\s+(.*?)\\s+([^ ]*)\"\\s+(-|[0-9]*)\\s+(-|[0-9]*)\\s+\"(.+?|-)\"\\s+\"(.+?|-)\"\\s+(.*)",
"output.format.string" = "%1$s %2$s %3$s %4$s %5$s %5$s %6$s %7$s %8$s %9$s %10$s" ) STORED AS TEXTFILE;
然后，将数据导入表中
方法1. 将指定路径下的文件导入，不移动文件
ALTER TABLE nugget_aos_log ADD partition (YEAR='2014', MONTH='08',DAY='19') location '/user/xxx/xx/year=2014/month=08/day=19';
或者
方法2. 导入指定文件，并将文件移动到用户的存储路径下
LOAD DATA inpath '/user/xxx/xx/2014/08/19' overwrite INTO TABLE xxx_log partition (YEAR='2014', MONTH='08',DAY='19');
接下来就可以查询表中的数据，做验证
hive>select * from xxx_log limit 100;
可能会碰到的情况，你的正则表达式在正则表达式工具（推荐：RegexBuddy）中，没有问题，但是在hive表显示时，每个字段值都是NULL，这说明正则表达式存在错误。这个时候可以用下面的办法解决，在hive中执行下面的命令：
hive>describe extended tablename;
– 查看表的详细信息，其中包括了hive表实际的input.regex正则表达式的值，看看建表的正则表达式转义字符是否缺少。
正则表达式的问题解决之后，再drop table xxx_log，再重新建表，导数据。最后能看到，nginx日志被拆分到了hive表的字段中，接下来就可以进行各种统计了。

hive 用户手册
Usage Examples
Creating tables
MovieLens User Ratings
CREATE TABLE u_data (
  userid INT,
  movieid INT,
  rating INT,
  unixtime STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE;
Apache Access Log Tables
add jar ../build/contrib/hive_contrib.jar;
 
CREATE TABLE apachelog (
  host STRING,
  identity STRING,
  user STRING,
  time STRING,
  request STRING,
  status STRING,
  size STRING,
  referer STRING,
  agent STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
  "input.regex" = "([^]*) ([^]*) ([^]*) (-|\\[^\\]*\\]) ([^ \"]*|\"[^\"]*\") (-|[0-9]*) (-|[0-9]*)(?: ([^ \"]*|\"[^\"]*\") ([^ \"]*|\"[^\"]*\"))?",
  "output.format.string" = "%1$s %2$s %3$s %4$s %5$s %6$s %7$s %8$s %9$s"
)
STORED AS TEXTFILE;
Control Separated Tables
CREATE TABLE mylog (
name STRING, language STRING, groups ARRAY<STRING>, entities MAP<INT, STRING>)
ROW FORMAT DELIMITED
  FIELDS TERMINATED BY '\001'
  COLLECTION ITEMS TERMINATED BY '\002'
  MAP KEYS TERMINATED BY '\003'
STORED AS TEXTFILE;
Loading tables
MovieLens User Ratings
Download and extract the data:
wget http://www.grouplens.org/system/files/ml-data.tar+0.gz
tar xvzf ml-data.tar+0.gz
Load it in:
LOAD DATA LOCAL INPATH 'ml-data/u.data'
OVERWRITE INTO TABLE u_data;
Running queries
MovieLens User Ratings
SELECT COUNT(1) FROM u_data;
Running custom map/reduce jobs
MovieLens User Ratings
Create weekday_mapper.py:
import sys
import datetime
 
for line in sys.stdin:
  line = line.strip()
  userid, movieid, rating, unixtime = line.split('\t')
  weekday = datetime.datetime.fromtimestamp(float(unixtime)).isoweekday()
  print '\t'.join([userid, movieid, rating, str(weekday)])
Use the mapper script:
CREATE TABLE u_data_new (
  userid INT,
  movieid INT,
  rating INT,
  weekday INT)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\t';
 
INSERT OVERWRITE TABLE u_data_new
SELECT
  TRANSFORM (userid, movieid, rating, unixtime)
  USING 'python weekday_mapper.py'
  AS (userid, movieid, rating, weekday)
FROM u_data;
 
SELECT weekday, COUNT(1)
FROM u_data_new
GROUP BY weekday;
Note: due to a bug in the parser, you must run the "INSERT OVERWRITE" query on a single line


hadoop map数目的分析和调整
Hadoop中在计算一个JOB需要的map数之前首先要计算分片的大小。计算分片大小的公式是：
goalSize = totalSize / mapred.map.tasks
minSize = max {mapred.min.split.size, minSplitSize}
splitSize = max (minSize, min(goalSize, dfs.block.size))
totalSize是一个JOB的所有map总的输入大小，即Map input bytes。参数mapred.map.tasks的默认值是2，我们可以更改这个参数的值。计算好了goalSize之后还要确定上限和下限。

下限是max {mapred.min.split.size, minSplitSize} 。参数mapred.min.split.size的默认值为1个字节，minSplitSize随着File Format的不同而不同。

上限是dfs.block.size，它的默认值是64兆。

举几个例子，例如Map input bytes是100兆，mapred.map.tasks默认值为2，那么分片大小就是50兆；如果我们把mapred.map.tasks改成1，那分片大小就变成了64兆。

计算好了分片大小之后接下来计算map数。Map数的计算是以文件为单位的，针对每一个文件做一个循环：

1.   文件大小/splitsize>1.1，创建一个split，这个split的大小=splitsize，文件剩余大小=文件大小-splitsize

2.   文件剩余大小/splitsize<1.1，剩余的部分作为一个split

举几个例子：

1.   input只有一个文件，大小为100M,splitsize=blocksize,则map数为2，第一个map处理的分片为64M,第二个为36M

2.   input只有一个文件，大小为65M,splitsize=blocksize，则map数为1，处理的分片大小为65M （因为65/64<1.1）

3.   input只有一个文件，大小为129M,splitsize=blocksize，则map数为2，第一个map处理的分片为64M,第二个为65M

4.   input有两个文件，大小为100M和20M,splitsize=blocksize,则map数为3，第一个文件分为两个map，第一个map处理的分片为64M,第二个为36M，第二个文件分为一个map，处理的分片大小为20M

5.   input有10个文件，每个大小10M，splitsize=blocksize，则map数为10，每个map处理的分片大小为10M

再看2个更特殊的例子：

1.   输入文件有2个，分别为40M和20M，dfs.block.size = 64M， mapred.map.tasks采用默认值2。那么splitSize = 30M ，map数实际为3，第一个文件分为2个map，第一个map处理的分片大小为30M，第二个map为10M；第二个文件分为1个map，大小为20M

2.   输入文件有2个，分别为40M和20M，dfs.block.size = 64M， mapred.map.tasks手工设置为1。

那么splitSize = 60M ，map数实际为2，第一个文件分为1个map，处理的分片大小为40M；第二个文件分为1个map，大小为20M

通过这2个特殊的例子可以看到mapred.map.tasks并不是设置的越大，JOB执行的效率就越高。同时，Hadoop在处理小文件时效率也会变差。

根据分片与map数的计算方法可以得出结论，一个map处理的分片最大不超过dfs.block.size * 1.1 ，默认情况下是70.4兆。但是有2个特例：

1.   Hive中合并小文件的map only JOB，此JOB只会有一个或很少的几个map。

2.   输入文件格式为压缩的Text File，因为压缩的文本格式不知道如何拆分，所以也只能用一个map。

考虑采用合适的压缩器（压缩速度vs性能）对输出进行压缩，提高HDFS的写入性能。
每个reduce不要输出多个文件，避免生成附属文件。我们一般用附属文件来记录统计信息，如果这些信息不多的话，可以使用计数器。
为输出文件选择合适的格式。对于下游消费者程序来说，用zlib/gzip/lzo等算法来对大量文本数据进行压缩往往事与愿违。因为zlib/gzip/lzo文件是不能分割的，只能整个进行处理。这会引起恶劣的负载均衡和故障恢复问题。作为改善，可以使用SequenceFile和TFile格式，它们不但是压缩的，而且是可以分割的。
如果每个输出文件都很大（若干GB），请考虑使用更大的输出块（dfs.block.size）。
CREATE TABLE apachelog (
  host STRING,
  identity STRING,
  user STRING,
  time STRING,
  request STRING,
  status STRING,
  size STRING,
  referer STRING,
  agent STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
  "input.regex" = "([^]*) ([^]*) ([^]*) (-|\\[^\\]*\\]) ([^ \"]*|\"[^\"]*\") (-|[0-9]*) (-|[0-9]*)(?: ([^ \"]*|\".*\") ([^ \"]*|\".*\"))?"
)
STORED AS TEXTFILE;

map数目调整

public static void setMinInputSplitSize(Job job,long size) {
     job.getConfiguration().setLong("mapred.min.split.size", size);
   }
   public static long getMinSplitSize(JobContext job) {
     return job.getConfiguration().getLong("mapred.min.split.size", 1L);
   }
    
 public static void setMaxInputSplitSize(Job job,long size) {
     job.getConfiguration().setLong("mapred.max.split.size", size);
   }
   public static long getMaxSplitSize(JobContext context) {
     return context.getConfiguration().getLong("mapred.max.split.size",Long.MAX_VALUE);
   }
如上我们可以看到，Hadoop在这里实现了对mapred.min.split.size和mapred.max.split.size的定义，且默认值分别为1和Long的最大。因此，我们在程序只需重新赋值给这两个值就可以控制InputSplit分片的大小了。
3.假如我们想要设置的分片大小为10MB
 则我们可以在MapReduce程序的驱动部分添加如下代码：

TextInputFormat.setMinInputSplitSize(job,1024L);//设置最小分片大小
 TextInputFormat.setMaxInputSplitSize(job,1024×1024×10L);//设置最大分片大小

UDF永久生效
hive根据不同的参数匹配evaluate();
udf比较简单一行生成一行，对一行进行计算，
udaf聚合函数比较麻烦，有merge函数等，在不同的map函数中和reduce中进行。
udtf有一行生成多行或者多列，initialize和progress以及close等，在建表的时候运用，可以用序列化和正则

来做到相同的效果     见http://www.uroot.com/archives/1059

UDF永久生效现在是只能修改源代码.不过可以变通一下处理也可以实现.
在HIVE_HOME的bin目录下新建一个.hiverc的文件,把写好的udf的注册语句写在这里就可以类似HIVE内置的用法一样用
原理是，在运行./hive命令时，同时会加载HIVE_HOME/bin/.hiverc and $HOME/.hiverc作为初始化所需要的文件 
在.hiverc文件中加入以下内容：
add jar /run/jar/Avg_test.jar
create temporary function avg_test 'hive.udaf.Avg';