[HADOOP] 사용자 정의 하이브 SerDe 열을 선택할 수 없습니다하지만 SELECT 않을 때 작동 *

사용자 정의 하이브 SerDe 열을 선택할 수 없습니다하지만 SELECT 않을 때 작동 *

나는 사용자 정의 SerDe을 쓰는 사람 만이 직렬화하기 위해 사용됩니다. 기본 데이터는 이진 중고품, 각 행은 이벤트 로그이다. 각 이벤트는 내가 액세스 할 수있는 스키마를 가지고 있지만, 우리는의를 저장하기 전에 메시지를 부르 자, 다른 스키마에 이벤트를 포장. 로 언급 한 기본 이벤트가 메시지로 포장되어 있기 때문에 나는 ThriftDeserializer를 사용하는 대신 SerDe을 쓰고 있어요 이유입니다. 그래서 우리는 먼저 메시지의 스키마를 사용하여 직렬화 한 후 해당 이벤트에 대한 데이터를 역 직렬화 할 필요가있다.

SerDe 작동 (만) 나는 *는 SELECT을 예상대로 내가 데이터를 역 직렬화 할 수 있지만 테이블 대신 SELECT *에서 열을 선택할 때마다, 행이 모두 NULL 때. 관리자가 반환되는 객체는 ThriftStructObjectInspector하고 역 직렬화에 의해 반환되는 객체는 TBase이다.

무엇 * 하이브 우리가 열을 선택하면 NULL을 반환,하지만 난 SELECT를 수행 할 때 열 데이터를 반환 할 수 있을까?

여기에 SerDe 클래스 (일부 클래스 이름 변경은)입니다 :

public class MyThriftSerde extends AbstractSerDe {

  private static final Log LOG = LogFactory.getLog(MyThriftSerde.class);

  /* Abstracting away the deserialization of the underlying event which is wrapped in a message */
  private static final MessageDeserializer myMessageDeserializer =
      MessageDeserializer.getInstance();

  /* Underlying event class which is wrapped in a Message */
  private String schemaClassName;
  private Class<?> schemaClass;

  /* Used to read the input row */
  public static List<String> inputFieldNames;
  public static List<ObjectInspector> inputFieldOIs;
  public static List<Integer> notSkipIDs;
  public static ObjectInspector inputRowObjectInspector;

  /* Output Object Inspector */
  public static ObjectInspector thriftStructObjectInspector;

  @Override
  public void initialize(Configuration conf, Properties tbl) throws SerDeException {
    try {

      logHeading("INITIALIZE MyThriftSerde");

      schemaClassName = tbl.getProperty(SERIALIZATION_CLASS);
      schemaClass = conf.getClassByName(schemaClassName);

      LOG.info(String.format("Building DDL for event: %s", schemaClass.getName()));

      inputFieldNames = new ArrayList<>();
      inputFieldOIs = new ArrayList<>();
      notSkipIDs = new ArrayList<>();

      /* Initialize the Input fields */

      // The underlying data is stored in RCFile format, and only has 1 column, event_binary
      // So we create a ColumnarStructBase for each row we deserialize.
      // This ColumnasStruct only has 1 column: event_binary
      inputFieldNames.add("event_binary");
      notSkipIDs.add(0);
      inputFieldOIs.add(LazyPrimitiveObjectInspectorFactory.LAZY_BINARY_OBJECT_INSPECTOR);
      inputRowObjectInspector =
          ObjectInspectorFactory.getColumnarStructObjectInspector(inputFieldNames, inputFieldOIs);

      /* Output Object Inspector*/

      // This is what the SerDe will return, it is a ThriftStructObjectInspector
      thriftStructObjectInspector =
          ObjectInspectorFactory.getReflectionObjectInspector(
              schemaClass, ObjectInspectorFactory.ObjectInspectorOptions.THRIFT);

      // Only for debugging
      logHeading("THRIFT OBJECT INSPECTOR");
      LOG.info("Output OI Class Name: " + thriftStructObjectInspector.getClass().getName());
      LOG.info(
          "OI Details: "
              + ObjectInspectorUtils.getObjectInspectorName(thriftStructObjectInspector));

    } catch (Exception e) {
      LOG.info("Exception while initializing SerDe", e);
    }
  }

  @Override
  public Object deserialize(Writable rowWritable) throws SerDeException {

    logHeading("START DESERIALIZATION");

    ColumnarStructBase inputLazyStruct =
        new ColumnarStruct(inputRowObjectInspector, notSkipIDs, null);
    LazyBinary eventBinary;
    Message rowAsMessage;
    TBase deserializedRow = null;

    try {
      inputLazyStruct.init((BytesRefArrayWritable) rowWritable);
      eventBinary = (LazyBinary) inputLazyStruct.getField(0);
      rowAsMessage =
          myMessageDeserializer.fromBytes(eventBinary.getWritableObject().copyBytes(), null);
      deserializedRow = rowAsMessage.getEvent();

      LOG.info("deserializedRow.getClass(): " + deserializedRow.getClass());
      LOG.info("deserializedRow.toString(): " + deserializedRow.toString());

    } catch (Exception e) {
      e.printStackTrace();
    }

    logHeading("END DESERIALIZATION");

    return deserializedRow;
  }

  private void logHeading(String s) {
    LOG.info(String.format("-------------------  %s  -------------------", s));
  }

  @Override
  public ObjectInspector getObjectInspector() {
    return thriftStructObjectInspector;
  }
}

코드에 대한 컨텍스트 :

내가 사용하는 드리프트 데이터를 가리키는 외부 테이블을 생성

ADD JAR hdfs://my-jar.jar;

CREATE EXTERNAL TABLE dev_db.thrift_event_data_deserialized
ROW FORMAT SERDE 'com.test.only.MyThriftSerde'
WITH SERDEPROPERTIES (
  "serialization.class"="com.test.only.TestEvent"
) STORED AS RCFILE
LOCATION 'location/of/thrift/data';

MSCK REPAIR TABLE thrift_event_data_deserialized;

이어서 dev_db.thrift_event_data_deserialized LIMIT (10)로부터 SELECT *; 예상대로 작동 그러나, SELECT column1_name는 dev_db.thrift_event_data_deserialized LIMIT 10 column2_name; 작동하지 않습니다.

내가 여기에 누락 어떤 생각? 이에 도움이 싶어요!

해결법

from https://stackoverflow.com/questions/52305545/custom-hive-serde-unable-to-select-column-but-works-when-i-do-select by cc-by-sa and MIT license