ElasticSearch中的日期映射为Hive中的日期格式

た入场券 2022-05-23 12:28 329阅读 0赞

现在的场景是需要将ElasticSearch中的数据导入到Hive中，但是在导入的时候发现了日期映射的异常，ElasticSearch中日期字段定义的格式为：

"time" : {
                "type" : "date",
                "store" : true,
                "include_in_all" : true,
                "format" : "yyyy-MM-dd HH:mm:ss"
              }

在Hive中建立外部表如下：

CREATE EXTERNAL TABLE `test_table_es`(
      `meta_id` string COMMENT 'from deserializer', 
      ......
      `time` TIMESTAMP)
    ROW FORMAT SERDE 
      'org.elasticsearch.hadoop.hive.EsSerDe' 
    STORED BY 
      'org.elasticsearch.hadoop.hive.EsStorageHandler' 
    WITH SERDEPROPERTIES ( 
      'serialization.format'='1')
    TBLPROPERTIES (
      'COLUMN_STATS_ACCURATE'='false', 
      'es.index.auto.create'='false', 
      'es.mapping.names'='meta_id:_metadata._id,time:time', 
      'es.nodes'='127.0.0.1:9200', 
      'es.read.metadata'='true', 
      'es.resource'='test_index/test_type');

建立外部表的时候，不会报错，但是查询的时候会报日期不能够正常映射为Hive中的Timestamp格式。

elasticsearch-hadoop中用于将ES中的日期转换为Hive中的日期格式的类为org.elasticsearch.hadoop.hive.HiveValueReader，通过查看该类的源码，其实现的用户日期转换的方法为：

@Override
        protected Object parseDate(String value, boolean richDate) {
            return (richDate ? new TimestampWritable(new Timestamp(DatatypeConverter.parseDateTime(value).getTimeInMillis())) : parseString(value));
        }

可以看到它是通过javax.xml.bind.DatatypeConverter.parseDateTime(String)方法将对应的日期字符串转换为日期的，该方法不支持的日期字符串格式为“yyyy-MM-dd HH:mm:ss”的字符串，它支持的日期字符串的格式为“yyyy-MM-ddTHH:mm:ss”这样的。

因而为了支持这种转换，可以选择两种处理方式，一是修改原始数据，二是在转换的过程中做数据转换，考虑到第一种方式要处理非常多的数据，因而采用了第二种方式，实现自己的ValueReader，在实现ValueReader的时候，要考虑兼容其它的日期格式，并且只处理属性中指定了日期格式为“yyyy-MM-dd HH:mm:ss”的日期转换，其它日期格式还是采用默认的方式处理，这样才能够兼容其它的日期格式，否则会导致其它的日期格式出错，以下是一个实现的自定义的EsValueReader的原码：

package com.service.hadoop;
    
    import java.sql.Timestamp;
    import java.text.ParseException;
    import java.text.ParsePosition;
    import java.text.SimpleDateFormat;
    import java.util.Calendar;
    import java.util.Date;
    
    import javax.xml.bind.DatatypeConverter;
    
    import org.apache.hadoop.hive.serde2.io.TimestampWritable;
    import org.elasticsearch.hadoop.cfg.Settings;
    import org.elasticsearch.hadoop.hive.HiveValueReader;
    
    import com.sun.xml.bind.DatatypeConverterImpl;
    
    /**
     * 类EsValueReader.java的实现描述：用于转换ES中的日期类型，用于匹配Hive中的日期类型
     * 
     * @author fenglibin 2018年4月19日 下午3:54:00
     */
    public class EsValueReader extends HiveValueReader {
    
        private String                dateFormat;
        private static final String   DEFALUT_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    
        @Override
        public void setSettings(Settings settings) {
            super.setSettings(settings);
            dateFormat = settings.getProperty("es.date.format");
        }
    
        @Override
        protected Object parseDate(String value, boolean richDate) {
            if (value != null && value.trim().length() > 0 && DEFALUT_DATE_FORMAT.equalsIgnoreCase(dateFormat)) {
                return (richDate ? new TimestampWritable(new Timestamp(parseDate(value,
                                                                                 DEFALUT_DATE_FORMAT).getTime())) : parseString(value));
            }
            /**如果没有设置日期格式，通过默认的方式支持，以避免使用新的ValueReader后影响到其它的外部表**/
            return super.parseDate(value, richDate);
            
        }
    
        /**
         * 解析日期,根据指定的格式进行解析.<br>
         * 如果解析错误,则返回null
         * @param stringDate 日期字符串
         * @param format 日期格式
         * @return 日期类型
         */
        private static Date parseDate(String stringDate, String format) {
            if (stringDate == null) {
                return null;
            }
            try {
                return parseDate(stringDate, new String[] { format });
            } catch (ParseException e) {
                return null;
            }
        }
        
        public static Date parseDate(String str, String... parsePatterns) throws ParseException {
            return parseDateWithLeniency(str, parsePatterns, true);
        }
        
        private static Date parseDateWithLeniency(
                String str, String[] parsePatterns, boolean lenient) throws ParseException {
            if (str == null || parsePatterns == null) {
                throw new IllegalArgumentException("Date and Patterns must not be null");
            }
            
            SimpleDateFormat parser = new SimpleDateFormat();
            parser.setLenient(lenient);
            ParsePosition pos = new ParsePosition(0);
            for (String parsePattern : parsePatterns) {
    
                String pattern = parsePattern;
    
                // LANG-530 - need to make sure 'ZZ' output doesn't get passed to SimpleDateFormat
                if (parsePattern.endsWith("ZZ")) {
                    pattern = pattern.substring(0, pattern.length() - 1);
                }
                
                parser.applyPattern(pattern);
                pos.setIndex(0);
    
                String str2 = str;
                // LANG-530 - need to make sure 'ZZ' output doesn't hit SimpleDateFormat as it will ParseException
                if (parsePattern.endsWith("ZZ")) {
                    str2 = str.replaceAll("([-+][0-9][0-9]):([0-9][0-9])$", "$1$2"); 
                }
    
                Date date = parser.parse(str2, pos);
                if (date != null && pos.getIndex() == str2.length()) {
                    return date;
                }
            }
            throw new ParseException("Unable to parse the date: " + str, -1);
        }
    
    }

将这个类导出一个jar包并将其加到Hive的auxlib目录，重启Hive。然后删除外部表重新建立，此时参数中增加指定"es.data.format"的格式为"yyyy-MM-dd HH:mm:ss"和指定value reader参数"es.ser.reader.value.class"的处理类为自定义实现的EsValueReader，示例语句如下：

CREATE EXTERNAL TABLE `test_table_es`(
      `meta_id` string COMMENT 'from deserializer', 
      ......
      `time` TIMESTAMP)
    ROW FORMAT SERDE 
      'org.elasticsearch.hadoop.hive.EsSerDe' 
    STORED BY 
      'org.elasticsearch.hadoop.hive.EsStorageHandler' 
    WITH SERDEPROPERTIES ( 
      'serialization.format'='1')
    TBLPROPERTIES (
      'COLUMN_STATS_ACCURATE'='false', 
      'es.date.format'='yyyy-MM-dd HH:mm:ss', 
      'es.index.auto.create'='false', 
      'es.mapping.names'='meta_id:_metadata._id,time:time', 
      'es.nodes'='127.0.0.1:9200', 
      'es.read.metadata'='true', 
      'es.ser.reader.value.class'='xxx.hadoop.EsValueReader'
      'es.resource'='test_index/test_type');