本文介紹了如何在Hive里新建一個外部分區(qū)表并加載數(shù)據(jù)

1.建表

# 使用數(shù)據(jù)庫
use blog;

# 創(chuàng)建外部分區(qū)表
create external table external_blog_record(
    host string comment "主機(jī)",
    app string comment "應(yīng)用",
    source string comment "來源",
    remote_addr string comment "訪問IP",
    time_iso6401 string comment "訪問時間",
    http_host string comment "域名",
    request_method string comment "請求方式",
    request_url string comment "請求地址",
    request_protocol string comment "請求協(xié)議",
    request_time string comment "請求耗時",
    status string comment "請求狀態(tài)",
    body_byte_sents string comment "內(nèi)容體大小",
    upstream_addr string comment "轉(zhuǎn)發(fā)服務(wù)器地址",
    upstream_response_time string comment "轉(zhuǎn)發(fā)響應(yīng)耗時",
    upstream_status string comment "轉(zhuǎn)發(fā)狀態(tài)",
    http_referer string comment "來源地址",
    http_user_agent string comment "瀏覽器類型",
    res_type string comment "資源類型：首頁树碱、文章浅妆、類別、其他"
) 
comment "日志原始記錄外部分區(qū)表"
partitioned by (day string) 
row format delimited fields terminated by '\t' 
location '/log/blog';

新建一個名為external_blog_record的數(shù)據(jù)庫表并制定分區(qū)參數(shù)day壳嚎，數(shù)據(jù)的格式用'\t'分隔桐智，數(shù)據(jù)的目錄存放在HDFS的'/log/blog'目錄下。

2.查看分區(qū)

# 查看表分區(qū)
show partitions external_blog_record;

分區(qū)列表

可以看到目前表里面已經(jīng)存在很多分區(qū)了烟馅，查看HDFS的目錄

hdfs中的分區(qū)

每個分區(qū)下對應(yīng)存放這日志文件说庭。

分區(qū)中的日志文件

3.新增分區(qū)

只需要在 /log/blog 下 新建day=XXX 即可，但是這樣新建的分區(qū)并沒有和Hive關(guān)聯(lián)起來焙糟，必須運行如下命令口渔，使分區(qū)與Hive關(guān)聯(lián)起來样屠。

msck repair table external_blog_record;

就可以用上面的查看分區(qū)的命令查看是否新建成功穿撮。

4.查詢分區(qū)下的記錄

hive> select count(*) from external_blog_record where day=20181122;
Query ID = hadoop_20181123144713_2b8b197a-c09b-4bf6-8ad3-b88cbd1ee4ca
Total jobs = 1
Launching Job 1 out of 1
Number of reduce tasks determined at compile time: 1
In order to change the average load for a reducer (in bytes):
  set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
  set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
  set mapreduce.job.reduces=<number>
Starting Job = job_1542348923310_0146, Tracking URL = http://hadoop1:8088/proxy/application_1542348923310_0146/
Kill Command = /opt/soft/hadoop-2.7.3/bin/hadoop job  -kill job_1542348923310_0146
Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 1
2018-11-23 14:47:25,697 Stage-1 map = 0%,  reduce = 0%
2018-11-23 14:47:32,237 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 1.49 sec
2018-11-23 14:47:39,923 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 2.62 sec
MapReduce Total cumulative CPU time: 2 seconds 620 msec
Ended Job = job_1542348923310_0146
MapReduce Jobs Launched:
Stage-Stage-1: Map: 1  Reduce: 1   Cumulative CPU: 2.62 sec   HDFS Read: 414396 HDFS Write: 5 SUCCESS
Total MapReduce CPU Time Spent: 2 seconds 620 msec
OK
1561
Time taken: 27.642 seconds, Fetched: 1 row(s)

5.附錄

利用MapperReduce來定時合并小文件并加載到Hive分區(qū)表里

/**
 * 合并日志文件并加載到Hive分區(qū)表
 */
public class MergeSmallFileAndLoadIntoHive {

    private static final Logger LOG = LoggerFactory.getLogger(MergeSmallFileAndLoadIntoHive.class);

    static class SmallFileCombinerMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        NullWritable v = NullWritable.get();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            context.write(value, v);
        }
    }

    public static void main(String[] args) throws Exception {

        boolean test = false;
        String logPath;
        String patition;

        if (test) {
            patition = "day=20181114";

            // Linux
            logPath = "/log/blog";

            // Windows
            logPath = "D:" + File.separator + "hadoop" + File.separator + "blog";
        } else {
            if (args == null || args.length < 2) {
                throw new RuntimeException("\"參數(shù)的長度不正確,參考：[java -jar xxxx.jar me.jinkun.mr.merge.MergeSmallFileAndLoadIntoHive /log/blog day=20181116]\"");
            }

            logPath = args[0];
            patition = args[1];
        }


        String tempInPath = logPath + File.separator + "temp" + File.separator + patition + File.separator + "in";
        String tempOutPath = logPath + File.separator + "temp" + File.separator + patition + File.separator + "out";

        //權(quán)限問題
        System.setProperty("HADOOP_USER_NAME", "hadoop");

        Configuration conf = new Configuration();
        if (!test) {
            conf.set("fs.defaultFS", "hdfs://hadoop1:9000");
        }

        // 1.獲取當(dāng)天臨時保存的日志
        List<Path> paths = new ArrayList<>();
        long currentTimeMillis = System.currentTimeMillis();
        FileSystem fs = FileSystem.get(conf);
        FileStatus[] fileStatuses = fs.listStatus(new Path(tempInPath));
        for (FileStatus fileStatus : fileStatuses) {
            if (fileStatus.isDirectory()) {
                Path path = fileStatus.getPath();
                String name = fileStatus.getPath().getName();
                if (!name.startsWith("delete") &&
                        name.compareTo(String.valueOf(currentTimeMillis)) < 0) {
                    paths.add(path);
                }
                LOG.info("文件夾名為：" + name);
            }
        }

        if (paths.size() == 0) {
            LOG.info("暫無可以合并的文件夾!不提交JOB");
            System.exit(0);
        }

        Job job = Job.getInstance(conf);
        job.setJarByClass(MergeSmallFileAndLoadIntoHive.class);
        job.setMapperClass(SmallFileCombinerMapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        // 2.合并小文件到臨時文件夾
        job.setInputFormatClass(CombineTextInputFormat.class);
        CombineTextInputFormat.setMaxInputSplitSize(job, 1024 * 1024 * 128);//128M
        CombineTextInputFormat.setInputPaths(job, paths.toArray(new Path[paths.size()]));
        Path tempResultPath = new Path(tempOutPath);
        FileOutputFormat.setOutputPath(job, tempResultPath);

        job.setNumReduceTasks(0);

        boolean flag = job.waitForCompletion(true);

        // 如果成功
        if (flag) {

            // 3.將合并后的文件移動到Hive的分區(qū)表
            int index = 0;
            FileStatus[] resultStatus = fs.listStatus(tempResultPath);
            for (FileStatus fileStatus : resultStatus) {
                Path path = fileStatus.getPath();
                if (path.getName().startsWith("part")) {
                    fs.rename(path, new Path(logPath + File.separator + patition + File.separator + currentTimeMillis + "." + index + ".log"));
                    index++;
                }
            }
            fs.delete(tempResultPath, true);

            // 4.標(biāo)記合并過的文件夾為已經(jīng)刪除
            for (Path path : paths) {
                fs.rename(path, new Path(path.getParent(), "delete_" + path.getName()));
            }

            fs.close();
        }
    }
}

執(zhí)行腳本

#!/bin/sh
day=`date '+%Y%m%d'`
echo "提交合并任務(wù) $day"
nohup /opt/soft/hadoop-2.7.3/bin/hadoop jar /opt/soft-install/schedule/mapreduce-1.0.jar me.jinkun.mr.merge.MergeSmallFileAndLoadIntoHive /log/blog day=$day > nohup.log 2>&1 &