Datax寫入parquet類型的hive表時處理timestamp類型字段的方法

一、概述

  1、  hive中的Timestamp

    Hive在0.8的版本后開始支持Timestamp的格式。Hive在儲存時間戳的時候會先把時間轉成UTC的時間，然后再把轉換后的時間存儲到Parquet文件中。在讀取Parquet文件的時候Hive會把時間從UTC時間再轉化回成本地的時間。這樣的話，如果存和讀取都是用Hive的話，時間不會有任何的問題。上述說的是用Parquet文件來存取時間格式流程，如果是存成普通的文本文件的話，存取都不會進行任何時間的轉換。

  2、Parquet中的Timestamp

    Parquet文件格式是當前Hadoop生態中最流行的列式存儲格式。Parquet支持的類型有BOOLEAN、INT32、INT64、INT96、FLOAT、DOUBLE、BYTE_ARRAY，所以Timestamp其實是一種邏輯類型。由於Impala存儲的時間精度達到納秒的級別，所以在Parquet文件中用INT96來存儲時間。其他的數據處理引擎也跟進該精度，所以也用INT96來存儲，但是在時區兼容性方面做得並不好。

二、parquet格式的hive中timestamp字段問題

   在datax中如果我們直接用group.append(columns.get(i).getString(Key.NAME),dataFormat.format(column.asDate()),在使用xshell讀取時會報如下錯誤：

 

   對於parquet類型文件的時間戳邏輯類型（注釋為int96），這種時間戳編碼（int96）似乎很少見，而且不受支持。再了解Parquet的timestamp存儲原理后，這個問題就好解決了，保存為Int96的時間戳由一天中的納秒組成。

  明確地：messagetype模式中的列使用哪種 Parquet類型？我們應該使用基元類型primitivetypename.int96。

Types.MessageTypeBuilder mtb = Types.buildMessage();
for(Configuration eachColumnConf : columns) {
  SupportHiveDataType columnType = SupportHiveDataType.valueOf(eachColumnConf.getString(Key.Type).toUpperCase());
  switch(columnType) {

   ...
   case TIMESTAMP:
   case DATE:
      mtb.optional(PrimitiveType.PrimitiveTypeName.INT96).named(eachColumnConf.getString(Key.NAME))
   break;
...}

 

 在存入時，需要將值進行轉換

group.append(columns.get(i).getString(Key.NAME),dataFormat.format(column.asDate())

 用spark sql中的這段代碼作為參考，終於找到了答案

String value = "2019-02-13 13:35:05";
 
final long NANOS_PER_HOUR = TimeUnit.HOURS.toNanos(1);
final long NANOS_PER_MINUTE = TimeUnit.MINUTES.toNanos(1);
final long NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1);
// Parse date
SimpleDateFormat parser = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
cal.setTime(parser.parse(value));
// Calculate Julian days and nanoseconds in the day
LocalDate dt = LocalDate.of(cal.get(Calendar.YEAR), cal.get(Calendar.MONTH)+1, cal.get(Calendar.DAY_OF_MONTH));
int julianDays = (int) JulianFields.JULIAN_DAY.getFrom(dt);
long nanos = (cal.get(Calendar.HOUR_OF_DAY) * NANOS_PER_HOUR)
        + (cal.get(Calendar.MINUTE) * NANOS_PER_MINUTE)
        + (cal.get(Calendar.SECOND) * NANOS_PER_SECOND);
// Write INT96 timestamp
byte[] timestampBuffer = new byte[12];
ByteBuffer buf = ByteBuffer.wrap(timestampBuffer);
buf.order(ByteOrder.LITTLE_ENDIAN).putLong(nanos).putInt(julianDays);
// This is the properly encoded INT96 timestamp
Binary tsValue = Binary.fromReusedByteArray(timestampBuffer);

   即將需要存儲的時間先轉換為Binary類型再存儲到parquet，hive讀取時再自動轉換為Timestamp，問題解決。