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

private HCatSchema schema;

private HCatRecordObjectInspector cachedObjectInspector;

private org.apache.hadoop.hive.serde2.JsonSerDe jsonSerde = new org.apache.hadoop.hive.serde2.JsonSerDe();

@Override

public void initialize(Configuration configuration, Properties tableProperties, Properties partitionProperties)

throws SerDeException {

super.initialize(configuration, tableProperties, partitionProperties);

jsonSerde.initialize(configuration, tableProperties, partitionProperties, false);

StructTypeInfo rowTypeInfo = jsonSerde.getTypeInfo();

cachedObjectInspector = HCatRecordObjectInspectorFactory.getHCatRecordObjectInspector(rowTypeInfo);

try {

schema = HCatSchemaUtils.getHCatSchema(rowTypeInfo).get(0).getStructSubSchema();

LOG.debug("schema : {}", schema);

LOG.debug("fields : {}", schema.getFieldNames());

} catch (HCatException e) {

throw new SerDeException(e);

}

}

/**

* Takes JSON string in Text form, and has to return an object representation above

* it that's readable by the corresponding object inspector.

*

* For this implementation, since we're using the jackson parser, we can construct

* our own object implementation, and we use HCatRecord for it

*/

@Override

public Object doDeserialize(Writable blob) throws SerDeException {

try {

List<?> row = (List<?>) jsonSerde.deserialize(blob);

List<Object> fatRow = fatLand(row);

return new DefaultHCatRecord(fatRow);

} catch (Exception e) {

throw new SerDeException(e);

}

}

@SuppressWarnings({"rawtypes", "unchecked" })

private static List<Object> fatLand(final List<?> arr) {

final List ret = new ArrayList<>();

for (final Object o : arr) {

if (o == null) {

ret.add(null);

} else if (o instanceof Map<?, ?>) {

ret.add(fatMap(((Map) o)));

} else if (o instanceof List<?>) {

ret.add(fatLand((List) o));

} else if (o.getClass().isArray()

&& o.getClass().getComponentType() != byte.class) {

Class<?> ct = o.getClass().getComponentType();

if (ct.isPrimitive()) {

ret.add(primitiveArrayToList(o));

} else {

ret.add(fatLand(Arrays.asList((Object[]) o)));

}

} else {

ret.add(o);

}

}

return ret;

}

private static Object fatMap(Map<Object, Object> map) {

Map<Object, Object> ret = new LinkedHashMap<>();

Set<Entry<Object, Object>> es = map.entrySet();

for (Entry<Object, Object> e : es) {

final Object oldV = e.getValue();

final Object newV;

if (oldV != null && oldV.getClass().isArray()) {

newV = fatLand(Arrays.asList((Object[]) oldV));

} else {

newV = oldV;

}

ret.put(e.getKey(), newV);

}

return ret;

}

private static Object primitiveArrayToList(Object arr) {

Class<?> ct = arr.getClass().getComponentType();

if (int.class.equals(ct)) {

return Arrays.asList(ArrayUtils.toObject((int[]) arr));

}

if (long.class.equals(ct)) {

return Arrays.asList(ArrayUtils.toObject((long[]) arr));

}

if (char.class.equals(ct)) {

return Arrays.asList(ArrayUtils.toObject((char[]) arr));

}

if (byte.class.equals(ct)) {

return Arrays.asList(ArrayUtils.toObject((byte[]) arr));

}

if (short.class.equals(ct)) {

return Arrays.asList(ArrayUtils.toObject((short[]) arr));

}

if (float.class.equals(ct)) {

return Arrays.asList(ArrayUtils.toObject((float[]) arr));

}

if (double.class.equals(ct)) {

return Arrays.asList(ArrayUtils.toObject((double[]) arr));

}

throw new RuntimeException("Unhandled primitiveArrayToList for type: " + ct);

}

public String getHiveInternalColumnName(int fpos) {

return HiveConf.getColumnInternalName(fpos);

}

/**

* Given an object and object inspector pair, traverse the object

* and generate a Text representation of the object.

*/

@Override

public Writable doSerialize(Object obj, ObjectInspector objInspector)

throws SerDeException {

return jsonSerde.serialize(obj, objInspector);

}

/**

* Returns an object inspector for the specified schema that

* is capable of reading in the object representation of the JSON string

*/

@Override

public ObjectInspector getObjectInspector() throws SerDeException {

return cachedObjectInspector;

}

@Override

public Class<? extends Writable> getSerializedClass() {

return Text.class;

}

/**

* Transform Writable data from UTF-8 to charset before serialize.

* @param blob

* @return

*/

@Override

protected Writable transformFromUTF8(Writable blob) {

Text text = (Text)blob;

return SerDeUtils.transformTextFromUTF8(text, this.charset);

}

@Override

protected Writable transformToUTF8(Writable blob) {

Text text = (Text)blob;

return SerDeUtils.transformTextToUTF8(text, this.charset);

}