More update to the hive blur code.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/3a41d16c Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/3a41d16c Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/3a41d16c Branch: refs/heads/blur-384-random-port-cleanup Commit: 3a41d16c3e28fa02c2892d382c865802101da081 Parents: 3ffdd12 Author: Aaron McCurry <[email protected]> Authored: Thu Oct 2 22:23:07 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Thu Oct 2 22:23:07 2014 -0400 ---------------------------------------------------------------------- .../blur/hive/BlurObjectInspectorGenerator.java | 35 +++++++++------ .../org/apache/blur/hive/BlurSerializer.java | 46 +++++++++++++++++++- 2 files changed, 66 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3a41d16c/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java index 98ca4df..87ad250 100644 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java +++ b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java @@ -35,19 +35,21 @@ import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; public class BlurObjectInspectorGenerator { - private static final String STORED = "stored"; - private static final String STRING = "string"; - private static final String TEXT = "text"; - private static final String LONG = "long"; - private static final String FLOAT = "float"; - private static final String INT = "int"; - private static final String DOUBLE = "double"; - private static final String DATE = "date"; - private static final String GEO_TERMPREFIX = "geo-termprefix"; - private static final String GEO_POINTVECTOR = "geo-pointvector"; - private static final String GEO_RECURSIVEPREFIX = "geo-recursiveprefix"; - private static final String LATITUDE = "latitude"; - private static final String LONGITUDE = "longitude"; + public static final String RECORDID = "recordid"; + public static final String ROWID = "rowid"; + public static final String STORED = "stored"; + public static final String STRING = "string"; + public static final String TEXT = "text"; + public static final String LONG = "long"; + public static final String FLOAT = "float"; + public static final String INT = "int"; + public static final String DOUBLE = "double"; + public static final String DATE = "date"; + public static final String GEO_TERMPREFIX = "geo-termprefix"; + public static final String GEO_POINTVECTOR = "geo-pointvector"; + public static final String GEO_RECURSIVEPREFIX = "geo-recursiveprefix"; + public static final String LATITUDE = "latitude"; + public static final String LONGITUDE = "longitude"; private static final Comparator<ColumnDefinition> COMPARATOR = new Comparator<ColumnDefinition>() { @Override @@ -63,6 +65,13 @@ public class BlurObjectInspectorGenerator { public BlurObjectInspectorGenerator(Collection<ColumnDefinition> colDefs) throws SerDeException { List<ColumnDefinition> colDefList = new ArrayList<ColumnDefinition>(colDefs); Collections.sort(colDefList, COMPARATOR); + + _columnNames.add(ROWID); + _columnTypes.add(TypeInfoFactory.stringTypeInfo); + + _columnNames.add(RECORDID); + _columnTypes.add(TypeInfoFactory.stringTypeInfo); + for (ColumnDefinition columnDefinition : colDefList) { _columnNames.add(columnDefinition.getColumnName()); _columnTypes.add(getTypeInfo(columnDefinition)); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3a41d16c/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java index f656d85..0fcf392 100644 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java +++ b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java @@ -19,16 +19,58 @@ package org.apache.blur.hive; import java.util.List; import java.util.Map; +import org.apache.blur.mapreduce.lib.BlurRecord; import org.apache.blur.thrift.generated.ColumnDefinition; +import org.apache.hadoop.hive.serde2.SerDeException; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; +import org.apache.hadoop.hive.serde2.objectinspector.StructField; +import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; import org.apache.hadoop.io.Writable; public class BlurSerializer { public Writable serialize(Object o, ObjectInspector objectInspector, List<String> columnNames, - List<TypeInfo> columnTypes, Map<String, ColumnDefinition> schema, String family) { - throw new RuntimeException("Not Implemented"); + List<TypeInfo> columnTypes, Map<String, ColumnDefinition> schema, String family) throws SerDeException { + BlurRecord blurRecord = new BlurRecord(); + blurRecord.setFamily(family); + + StructObjectInspector soi = (StructObjectInspector) objectInspector; + + List<? extends StructField> outputFieldRefs = soi.getAllStructFieldRefs(); + int size = columnNames.size(); + if (outputFieldRefs.size() != size) { + throw new SerDeException("Number of input columns was different than output columns (in = " + size + " vs out = " + + outputFieldRefs.size()); + } + List<Object> structFieldsDataAsList = soi.getStructFieldsDataAsList(o); + for (int i = 0; i < size; i++) { + // StructField structFieldRef = outputFieldRefs.get(i); + Object structFieldData = structFieldsDataAsList.get(i); + if (structFieldData == null) { + continue; + } + // ObjectInspector fieldOI = structFieldRef.getFieldObjectInspector(); + String columnName = columnNames.get(i); + if (columnName.equals(BlurObjectInspectorGenerator.ROWID)) { + blurRecord.setRowId((String) structFieldData); + } else if (columnName.equals(BlurObjectInspectorGenerator.RECORDID)) { + blurRecord.setRecordId((String) structFieldData); + } else { + if (columnName.equals(BlurObjectInspectorGenerator.GEO_POINTVECTOR) + || columnName.equals(BlurObjectInspectorGenerator.GEO_RECURSIVEPREFIX) + || columnName.equals(BlurObjectInspectorGenerator.GEO_TERMPREFIX)) { + throw new SerDeException("Not supported yet."); + } else { + blurRecord.addColumn(columnName, toString(structFieldData)); + } + } + } + return blurRecord; + } + + private String toString(Object o) { + return o.toString(); } }
