【大数据Hive】hive 多字段分隔符使用详解

���录

一、前言

二、hive默认分隔符规则以及限制

2.1 正常示例：单字节分隔符数据加载示例

2.2 特殊格式的文本数据，分隔符为特殊字符

2.2.1 文本数据的字段中包含了分隔符

三、突破默认限制规则约束

3.1  数据加载不匹配情况 1

3.2  数据加载不匹配情况 2

3.3  解决方案一：替换分隔符

3.4  解决方案二：RegexSerDe正则加载

问题一处理过程：

问题二处理过程：

3.5  解决方案三：自定义InputFormat

3.5.1 操作流程

四、URL解析函数

4.1 URL基本组成

4.1.1 parse_url

4.1.2 问题分析

4.1.3 parse_url_tuple

4.1.4 案例操作演示

---

一、前言

分隔符是hive在建表的时候要考虑的一个重要因素，根据要加载的原始数据的格式不同，通常数据文件中的分隔符也有差异，因此可以在建表的时候指定分隔符，从而映射到hive的数据表。

二、hive默认分隔符规则以及限制

Hive默认序列化类是LazySimpleSerDe，其只支持使用单字节分隔符（char）来加载文本数据，例如逗号、制表符、空格等等，默认的分隔符为”\001”。

根据不同文件的不同分隔符，我们可以通过在创建表时使用 row format delimited 来指定文件中的分割符，确保正确将表中的每一列与文件中的每一列实现一一对应的关系。

如下是hive建表语法树中的一部分

在这个语法树中，大家熟知的分隔符即 DELIMITED 关键字，从语法中看出来默认情况下，其分割的都是单字节的数据，可现实情况下，实际要处理的文本数据内容可能要复杂很多，比如下面这些情况：

2.1 正常示例：单字节分隔符数据加载示例

下面这种文本格式的原始数据，可以直接使用没问题；

 

2.2 特殊格式的文本数据，分隔符为特殊字符

每一行数据的分隔符是多字节分隔符，例如：”||”、“--”等，如下面这样的数据；

2.2.1 文本数据的字段中包含了分隔符

每列的分隔符为空格，但是数据中包含了分割符，时间字段中也有空格；

三、突破默认限制规则约束

3.1  数据加载不匹配情况 1

文本内容数据格式如下

 建表sql，这里字段分隔符采用 || 与文本对应；

drop table singer;create table singer(    id string,    name string,    country string,    province string,    gender string,    works string)row format delimited fields terminated by '||';load data local inpath '/usr/local/soft/selectdata/test01.txt' into table singer;

执行建表并加载数据

从数据来看，字段并没有解析完全，并且某些字段解析失败，和预期的不太一样，这是怎么回事呢？

3.2  数据加载不匹配情况 2

原始文本数据内容格式如下

建表并加载数据，这里采用空格作为分隔符；

drop table apachelog;create table apachelog( ip string,stime string,mothed string,url string,policy string,stat string,body string)row format delimited fields terminated by ' ';load data local inpath '/usr/local/soft/selectdata/apache_web_access.log' into table apachelog;

执行完成后检查数据

从数据来看，某些字段的解析不仅错误，而且字段也出现了错位；

从上面两个简单的示例来看，如果要解析的原始文本数据中的某些字段自身包含了分隔符，这时候再使用默认的LazySimpleSerDe序列化加载数据时，将得不到预期的结果，出现数据解析错误的情况。

关于上述问题，下面提几种常用的解决办法。

3.3  解决方案一：替换分隔符

在第一个示例中的数据，要想使用默认分隔符，可以考虑对原始数据进行预处理，将双|转换为单个|后再导入；

至于转换的过程，可以人工处理，也可以使用MR程序处理，使用MR程序处理的话可以参考下面的伪代码，

package bigdata.itcast.cn.hbase.mr;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;import java.io.IOException;/** * @ClassName ChangeSplitCharMR * @Description TODO MapReduce实现将多字节分隔符转换为单字节符 * @Create By  itcast */public class ChangeSplitCharMR extends Configured implements Tool {    public int run(String[] arg) throws Exception {        /**         * 构建Job         */        Job job = Job.getInstance(this.getConf(),"changeSplit");        job.setJarByClass(ChangeSplitCharMR.class);        /**         * 配置Job         */        //input：读取需要转换的文件        job.setInputFormatClass(TextInputFormat.class);        Path inputPath = new Path("datas/split/test01.txt");        FileInputFormat.setInputPaths(job,inputPath);        //map：调用Mapper        job.setMapperClass(ChangeSplitMapper.class);        job.setMapOutputKeyClass(Text.class);        job.setMapOutputValueClass(NullWritable.class);        //reduce：不需要Reduce过程        job.setNumReduceTasks(0);        //output        job.setOutputFormatClass(TextOutputFormat.class);        Path outputPath = new Path("datas/output/changeSplit");        TextOutputFormat.setOutputPath(job,outputPath);        /**         * 提交Job         */        return job.waitForCompletion(true) ? 0 : -1;    }    //程序入口    public static void main(String[] args) throws Exception {        //调用run        Configuration conf = new Configuration();        int status = ToolRunner.run(conf, new ChangeSplitCharMR(), args);        System.exit(status);    }    public static class ChangeSplitMapper extends Mapper{        //定义输出的Key        private Text outputKey = new Text();        //定义输出的Value        private NullWritable outputValue = NullWritable.get();        @Override        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {            //获取每条数据            String line = value.toString();            //将里面的||转换为|            String newLine = line.replaceAll("\|\|", "|");            //替换后的内容作为Key            this.outputKey.set(newLine);            //输出结果            context.write(this.outputKey,this.outputValue);        }    }}

3.4  解决方案二：RegexSerDe正则加载

顾名思义就是使用hive提供的相关正则的语法来处理这个问题，为什么呢？因为hive内置了很多SerDe类；

Hive内置的SerDe

• 除了使用最多的LazySimpleSerDe，Hive该内置了很多SerDe类；
• 官网地址：https://cwiki.apache.org/confluence/display/Hive/SerDe；
• 多种SerDe用于解析和加载不同类型的数据文件，常用的有ORCSerDe 、RegexSerDe、JsonSerDe等；

  1、RegexSerDe用来加载特殊数据的问题，使用正则匹配来加载数据；

  2、根据正则表达式匹配每一列数据；

  官网参考文档

  针对上面演示时的问题，来看看如何使用这种方式来解决，比如第一份数据，针对这份数据，只需要写一个正则，能够识别到其中的分隔符双 || ，将建表时的字段分割符使用这个正则，然后加载数据的时候就可以把hive解析出预期的数据格式了；

  使用正则Regex处理这两个问题，下面看具体的操作演示

  问题一处理过程：

  建表并加载数据

  --如果表已存在就删除表drop table if exists singer;--创建表create table singer(id string,--歌手id                    name string,--歌手名称                    country string,--国家                    province string,--省份                    gender string,--性别                    works string)--作品--指定使用RegexSerde加载数据ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'WITH SERDEPROPERTIES ("input.regex" = "([0-9]*)\|\|(.*)\|\|(.*)\|\|(.*)\|\|(.*)\|\|(.*)");--加载数据load data local inpath '/usr/local/soft/selectdata/test01.txt' into table singer;

  执行过程

  检查数据发现，通过这种方式数据就能正确的加载了；

  问题二处理过程：

  创建表并加载数据，使用正则处理

  --如果表存在，就删除表drop table if exists apachelog;--创建表create table apachelog(                          ip string,      --IP地址                          stime string,    --时间                          mothed string,  --请求方式                          url string,     --请求地址                          policy string,  --请求协议                          stat string,    --请求状态                          body string     --字节大小)--指定使用RegexSerde加载数据    ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'--指定正则表达式        WITH SERDEPROPERTIES (        "input.regex" = "([^ ]*) ([^}]*) ([^ ]*) ([^ ]*) ([^ ]*) ([0-9]*) ([^ ]*)"        ) stored as textfile ;load data local inpath '/usr/local/soft/selectdata/apache_web_access.log' into table apachelog;

  执行过程

  检查数据发现，通过这种方式数据就能正确的加载了；

  3.5  解决方案三：自定义InputFormat

  Hive中也允许使用自定义InputFormat来解决以上问题，通过在自定义InputFormat，来自定义解析逻辑实现读取每一行的数据。

  下面是官方文档关于该方案的说明； 

  3.5.1 操作流程

  自定义InputFormat，与MapReudce中自定义InputFormat一致，继承TextInputFormat，下面是完整的代码；

  自定义UserInputFormat

  import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapred.*;import java.io.IOException;/** * @ClassName UserInputFormat * @Description TODO 用于实现自定义InputFormat，读取每行数据 */public class UserInputFormat extends TextInputFormat {    @Override    public RecordReader getRecordReader(InputSplit genericSplit, JobConf job,             Reporter reporter) throws IOException {        reporter.setStatus(genericSplit.toString());        UserRecordReader reader = new UserRecordReader(job,(FileSplit)genericSplit);        return reader;    }}

  UserRecordReader

  用于自定义读取器，在自定义InputFormat中使用，将读取到的每行数据中的||替换为|

  代码如下

  import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FSDataInputStream;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.fs.Seekable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.io.compress.*;import org.apache.hadoop.mapred.FileSplit;import org.apache.hadoop.mapred.LineRecordReader;import org.apache.hadoop.mapred.RecordReader;import java.io.IOException;import java.io.InputStream;/** * @ClassName UserRecordReader * @Description TODO 用于自定义读取器，在自定义InputFormat中使用，将读取到的每行数据中的||替换为| */public class UserRecordReader implements RecordReader {    private static final Log LOG = LogFactory.getLog(LineRecordReader.class.getName());    int maxLineLength;    private CompressionCodecFactory compressionCodecs = null;    private long start;    private long pos;    private long end;    private LineReader in;    private Seekable filePosition;    private CompressionCodec codec;    private Decompressor decompressor;    public UserRecordReader(Configuration job, FileSplit split) throws IOException {        this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength", Integer.MAX_VALUE);        start = split.getStart();        end = start + split.getLength();        final Path file = split.getPath();        compressionCodecs = new CompressionCodecFactory(job);        codec = compressionCodecs.getCodec(file);        FileSystem fs = file.getFileSystem(job);        FSDataInputStream fileIn = fs.open(split.getPath());        if (isCompressedInput()) {            decompressor = CodecPool.getDecompressor(codec);            if (codec instanceof SplittableCompressionCodec) {                final SplitCompressionInputStream cIn = ((SplittableCompressionCodec) codec)                        .createInputStream(fileIn, decompressor, start, end,                                SplittableCompressionCodec.READ_MODE.BYBLOCK);                in = new LineReader(cIn, job);                start = cIn.getAdjustedStart();                end = cIn.getAdjustedEnd();                filePosition = cIn; // take pos from compressed stream            } else {                in = new LineReader(codec.createInputStream(fileIn, decompressor), job);                filePosition = fileIn;            }        } else {            fileIn.seek(start);            in = new LineReader(fileIn, job);            filePosition = fileIn;        }        if (start != 0) {            start += in.readLine(new Text(), 0, maxBytesToConsume(start));        }        this.pos = start;    }    private boolean isCompressedInput() {        return (codec != null);    }    private int maxBytesToConsume(long pos) {        return isCompressedInput() ? Integer.MAX_VALUE : (int) Math.min(Integer.MAX_VALUE, end - pos);    }    private long getFilePosition() throws IOException {        long retVal;        if (isCompressedInput() && null != filePosition) {            retVal = filePosition.getPos();        } else {            retVal = pos;        }        return retVal;    }    public LongWritable createKey() {        return new LongWritable();    }    public Text createValue() {        return new Text();    }    /**     * Read a line.     */    public synchronized boolean next(LongWritable key, Text value) throws IOException {        while (getFilePosition() =1) parts from a URL；  
  It takes a URL and one or multiple partnames, and returns a tuple；
     
  4.1.4 案例操作演示
    
  创建一张表并加载数据
   
  drop table if exists tb_url;--建表create table tb_url(    id int,    url string)row format delimitedfields terminated by '\t';--加载数据load data local inpath '/usr/local/soft/selectdata/url.txt' into table tb_url;
   
  执行过程
     
  检查数据是否加载成功
    
  接下来体验下parse_url_tuple函数的使用
   
  解析host和path
     
  select parse_url_tuple(url,"HOST","PATH") as (host,path) from tb_url;
     
  解析出 PROTOCOL,HOST和PATH
     
  select parse_url_tuple(url,"PROTOCOL","HOST","PATH") as (protocol,host,path) from tb_url;
     
  解析查询参数
     
  select parse_url_tuple(url,"HOST","PATH","QUERY") as (host,path,query) from tb_url;