Repository: incubator-blur Updated Branches: refs/heads/master f07b1a6be -> 1d3847019
Adding an actual unit test for the hive blur integration. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/1d384701 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/1d384701 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/1d384701 Branch: refs/heads/master Commit: 1d384701974554a52733c9c69c3a54faa055333a Parents: f07b1a6 Author: Aaron McCurry <[email protected]> Authored: Thu Jan 15 09:05:46 2015 -0500 Committer: Aaron McCurry <[email protected]> Committed: Thu Jan 15 09:05:46 2015 -0500 ---------------------------------------------------------------------- blur-hive/pom.xml | 13 + .../blur/hive/BlurObjectInspectorGenerator.java | 4 +- .../java/org/apache/blur/hive/BlurSerDe.java | 20 +- .../org/apache/blur/hive/BlurSerializer.java | 241 +++++++------------ .../org/apache/blur/hive/BlurSerDeTest.java | 171 +++++++++++++ 5 files changed, 289 insertions(+), 160 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/1d384701/blur-hive/pom.xml ---------------------------------------------------------------------- diff --git a/blur-hive/pom.xml b/blur-hive/pom.xml index cbcc7b7..1f47f83 100644 --- a/blur-hive/pom.xml +++ b/blur-hive/pom.xml @@ -40,6 +40,12 @@ </dependency> <dependency> <groupId>org.apache.hive</groupId> + <artifactId>hive-serde</artifactId> + <version>${hive.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.hive</groupId> <artifactId>hive-exec</artifactId> <version>${hive.version}</version> <scope>provided</scope> @@ -52,6 +58,13 @@ </dependency> <dependency> <groupId>org.apache.blur</groupId> + <artifactId>blur-core</artifactId> + <version>${project.version}</version> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.blur</groupId> <artifactId>blur-thrift</artifactId> <version>${project.version}</version> <exclusions> http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/1d384701/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java b/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java index 0905cc9..c915531 100644 --- a/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java @@ -118,13 +118,11 @@ public class BlurObjectInspectorGenerator { return TypeInfoFactory.doubleTypeInfo; } else if (fieldType.equals(DATE)) { return TypeInfoFactory.dateTypeInfo; - } else if (fieldType.equals(GEO_POINTVECTOR)) { - return TypeInfoFactory.dateTypeInfo; } else if (fieldType.equals(GEO_POINTVECTOR) || fieldType.equals(GEO_RECURSIVEPREFIX) || fieldType.equals(GEO_TERMPREFIX)) { List<TypeInfo> typeInfos = Arrays.asList((TypeInfo) TypeInfoFactory.floatTypeInfo, (TypeInfo) TypeInfoFactory.floatTypeInfo); - TypeInfoFactory.getStructTypeInfo(Arrays.asList(LONGITUDE, LATITUDE), typeInfos); + return TypeInfoFactory.getStructTypeInfo(Arrays.asList(LATITUDE, LONGITUDE), typeInfos); } throw new SerDeException("Blur Field Type [" + fieldType + "] is not supported."); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/1d384701/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java b/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java index b35feec..ea1a675 100644 --- a/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java @@ -44,8 +44,9 @@ import org.apache.hadoop.io.Writable; public class BlurSerDe extends AbstractSerDe { public static final String BLUR_CONTROLLER_CONNECTION_STR = "BLUR_CONTROLLER_CONNECTION_STR"; - private static final String FAMILY = "blur.family"; - private static final String TABLE = "blur.table"; + public static final String FAMILY = "blur.family"; + public static final String TABLE = "blur.table"; + public static final String ZK = BlurConstants.BLUR_ZOOKEEPER_CONNECTION; private String _family; private Map<String, ColumnDefinition> _schema; private ObjectInspector _objectInspector; @@ -56,12 +57,15 @@ public class BlurSerDe extends AbstractSerDe { @Override public void initialize(Configuration conf, Properties tbl) throws SerDeException { String table = tbl.getProperty(TABLE); + nullCheck(TABLE, table); _family = tbl.getProperty(FAMILY); + nullCheck(FAMILY, _family); BlurConfiguration configuration; try { configuration = new BlurConfiguration(); - configuration.set(BlurConstants.BLUR_ZOOKEEPER_CONNECTION, - tbl.getProperty(BlurConstants.BLUR_ZOOKEEPER_CONNECTION)); + String zkConnectionStr = tbl.getProperty(ZK); + nullCheck(ZK, zkConnectionStr); + configuration.set(ZK, zkConnectionStr); } catch (IOException e) { throw new SerDeException(e); } @@ -105,7 +109,13 @@ public class BlurSerDe extends AbstractSerDe { _columnNames = blurObjectInspectorGenerator.getColumnNames(); _columnTypes = blurObjectInspectorGenerator.getColumnTypes(); - _serializer = new BlurSerializer(); + _serializer = new BlurSerializer(_schema); + } + + private void nullCheck(String name, String value) throws SerDeException { + if (value == null) { + throw new SerDeException("Property [" + name + "] cannot be null."); + } } private String getControllerConnectionStr(Iface client) throws BlurException, TException { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/1d384701/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java b/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java index b65698f..5cd5b68 100644 --- a/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java @@ -16,42 +16,50 @@ */ package org.apache.blur.hive; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; 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.io.ByteWritable; -import org.apache.hadoop.hive.serde2.io.DoubleWritable; -import org.apache.hadoop.hive.serde2.io.ShortWritable; -import org.apache.hadoop.hive.serde2.lazy.LazyBoolean; -import org.apache.hadoop.hive.serde2.lazy.LazyByte; -import org.apache.hadoop.hive.serde2.lazy.LazyDate; -import org.apache.hadoop.hive.serde2.lazy.LazyDouble; -import org.apache.hadoop.hive.serde2.lazy.LazyFloat; -import org.apache.hadoop.hive.serde2.lazy.LazyHiveChar; -import org.apache.hadoop.hive.serde2.lazy.LazyHiveDecimal; -import org.apache.hadoop.hive.serde2.lazy.LazyHiveVarchar; -import org.apache.hadoop.hive.serde2.lazy.LazyInteger; -import org.apache.hadoop.hive.serde2.lazy.LazyLong; -import org.apache.hadoop.hive.serde2.lazy.LazyShort; -import org.apache.hadoop.hive.serde2.lazy.LazyString; -import org.apache.hadoop.hive.serde2.lazy.LazyTimestamp; -import org.apache.hadoop.hive.serde2.lazy.LazyVoid; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; +import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector; 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.BooleanWritable; -import org.apache.hadoop.io.FloatWritable; -import org.apache.hadoop.io.IntWritable; -import org.apache.hadoop.io.LongWritable; -import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; public class BlurSerializer { + private static final String DATE_FORMAT = "dateFormat"; + private static final String DATE = "date"; + private Map<String, ThreadLocal<SimpleDateFormat>> _dateFormat = new HashMap<String, ThreadLocal<SimpleDateFormat>>(); + + public BlurSerializer(Map<String, ColumnDefinition> colDefs) { + Set<Entry<String, ColumnDefinition>> entrySet = colDefs.entrySet(); + for (Entry<String, ColumnDefinition> e : entrySet) { + String columnName = e.getKey(); + ColumnDefinition columnDefinition = e.getValue(); + String fieldType = columnDefinition.getFieldType(); + if (fieldType.equals(DATE)) { + Map<String, String> properties = columnDefinition.getProperties(); + final String dateFormat = properties.get(DATE_FORMAT); + ThreadLocal<SimpleDateFormat> threadLocal = new ThreadLocal<SimpleDateFormat>() { + @Override + protected SimpleDateFormat initialValue() { + return new SimpleDateFormat(dateFormat); + } + }; + _dateFormat.put(columnName, threadLocal); + } + } + } + public Writable serialize(Object o, ObjectInspector objectInspector, List<String> columnNames, List<TypeInfo> columnTypes, Map<String, ColumnDefinition> schema, String family) throws SerDeException { BlurRecord blurRecord = new BlurRecord(); @@ -65,159 +73,88 @@ public class BlurSerializer { throw new SerDeException("Number of input columns was different than output columns (in = " + size + " vs out = " + outputFieldRefs.size()); } + List<Object> structFieldsDataAsList = soi.getStructFieldsDataAsList(o); + + if (structFieldsDataAsList.size() != size) { + throw new SerDeException("Number of input columns was different than output columns (in = " + + structFieldsDataAsList.size() + " vs out = " + size); + } + for (int i = 0; i < size; i++) { - // StructField structFieldRef = outputFieldRefs.get(i); + StructField structFieldRef = outputFieldRefs.get(i); Object structFieldData = structFieldsDataAsList.get(i); if (structFieldData == null) { continue; } - // ObjectInspector fieldOI = structFieldRef.getFieldObjectInspector(); + ObjectInspector fieldOI = structFieldRef.getFieldObjectInspector(); String columnName = columnNames.get(i); - String stringValue = toString(structFieldData); - if (stringValue == null) { - continue; - } - if (columnName.equals(BlurObjectInspectorGenerator.ROWID)) { - blurRecord.setRowId(stringValue); - } else if (columnName.equals(BlurObjectInspectorGenerator.RECORDID)) { - blurRecord.setRecordId(stringValue); - } else { - if (columnName.equals(BlurObjectInspectorGenerator.GEO_POINTVECTOR) - || columnName.equals(BlurObjectInspectorGenerator.GEO_RECURSIVEPREFIX) - || columnName.equals(BlurObjectInspectorGenerator.GEO_TERMPREFIX)) { - throw new SerDeException("Not supported yet."); + if (fieldOI instanceof PrimitiveObjectInspector) { + PrimitiveObjectInspector primitiveObjectInspector = (PrimitiveObjectInspector) fieldOI; + Object primitiveJavaObject = primitiveObjectInspector.getPrimitiveJavaObject(structFieldData); + String strValue = toString(columnName, primitiveJavaObject); + if (columnName.equals(BlurObjectInspectorGenerator.ROWID)) { + blurRecord.setRowId(strValue); + } else if (columnName.equals(BlurObjectInspectorGenerator.RECORDID)) { + blurRecord.setRecordId(strValue); } else { - blurRecord.addColumn(columnName, stringValue); + blurRecord.addColumn(columnName, strValue); } + } else if (fieldOI instanceof StructObjectInspector) { + StructObjectInspector structObjectInspector = (StructObjectInspector) fieldOI; + Map<String, StructField> allStructFieldRefs = toMap(structObjectInspector.getAllStructFieldRefs()); + StructField latStructField = allStructFieldRefs.get(BlurObjectInspectorGenerator.LATITUDE); + StructField longStructField = allStructFieldRefs.get(BlurObjectInspectorGenerator.LONGITUDE); + Object latStructFieldData = structObjectInspector.getStructFieldData(structFieldData, latStructField); + Object longStructFieldData = structObjectInspector.getStructFieldData(structFieldData, longStructField); + blurRecord.addColumn(columnName, toLatLong(latStructFieldData, longStructFieldData)); + } else { + throw new SerDeException("ObjectInspector [" + fieldOI + "] of type [" + + (fieldOI != null ? fieldOI.getClass() : null) + "] not supported."); } } return blurRecord; } - private String toString(Object o) { - if (o == null) { - return null; - } - if (o instanceof LazyBoolean) { - return lazyBoolean((LazyBoolean) o); - } else if (o instanceof LazyByte) { - return lazyByte((LazyByte) o); - } else if (o instanceof LazyDate) { - return lazyDate((LazyDate) o); - } else if (o instanceof LazyDouble) { - return lazyDouble((LazyDouble) o); - } else if (o instanceof LazyFloat) { - return lazyFloat((LazyFloat) o); - } else if (o instanceof LazyHiveChar) { - return lazyHiveChar((LazyHiveChar) o); - } else if (o instanceof LazyHiveDecimal) { - return lazyHiveDecimal((LazyHiveDecimal) o); - } else if (o instanceof LazyHiveVarchar) { - return lazyHiveVarchar((LazyHiveVarchar) o); - } else if (o instanceof LazyInteger) { - return lazyInteger((LazyInteger) o); - } else if (o instanceof LazyLong) { - return lazyLong((LazyLong) o); - } else if (o instanceof LazyShort) { - return lazyShort((LazyShort) o); - } else if (o instanceof LazyShort) { - return lazyString((LazyString) o); - } else if (o instanceof LazyTimestamp) { - return lazyTimestamp((LazyTimestamp) o); - } else if (o instanceof LazyVoid) { - return null; - } - return o.toString(); - } - - private String lazyInteger(LazyInteger o) { - IntWritable writableObject = o.getWritableObject(); - if (writableObject == null) { - return null; - } - int i = writableObject.get(); - return Integer.toString(i); - } - - private String lazyLong(LazyLong o) { - LongWritable writableObject = o.getWritableObject(); - if (writableObject == null) { - return null; - } - long l = writableObject.get(); - return Long.toString(l); - } - - private String lazyShort(LazyShort o) { - ShortWritable writableObject = o.getWritableObject(); - if (writableObject == null) { - return null; - } - short s = writableObject.get(); - return Short.toString(s); + private String toLatLong(Object latStructFieldData, Object longStructFieldData) throws SerDeException { + return toString(BlurObjectInspectorGenerator.LATITUDE, latStructFieldData) + "," + + toString(BlurObjectInspectorGenerator.LONGITUDE, longStructFieldData); } - private String lazyString(LazyString o) { - Text writableObject = o.getWritableObject(); - if (writableObject == null) { - return null; + private Map<String, StructField> toMap(List<? extends StructField> allStructFieldRefs) { + Map<String, StructField> map = new HashMap<String, StructField>(); + for (StructField structField : allStructFieldRefs) { + map.put(structField.getFieldName(), structField); } - return writableObject.toString(); - } - - private String lazyTimestamp(LazyTimestamp o) { - throw new RuntimeException("Not implemented."); - } - - private String lazyHiveVarchar(LazyHiveVarchar o) { - throw new RuntimeException("Not implemented."); - } - - private String lazyHiveDecimal(LazyHiveDecimal o) { - throw new RuntimeException("Not implemented."); - } - - private String lazyHiveChar(LazyHiveChar o) { - throw new RuntimeException("Not implemented."); + return map; } - private String lazyFloat(LazyFloat o) { - FloatWritable writableObject = o.getWritableObject(); - if (writableObject == null) { - return null; - } - float f = writableObject.get(); - return Float.toString(f); - } - - private String lazyDouble(LazyDouble o) { - DoubleWritable writableObject = o.getWritableObject(); - if (writableObject == null) { - return null; - } - double d = writableObject.get(); - return Double.toString(d); - } - - private String lazyDate(LazyDate o) { - throw new RuntimeException("Not implemented."); - } - - private String lazyByte(LazyByte o) { - ByteWritable writableObject = o.getWritableObject(); - if (writableObject == null) { + private String toString(String columnName, Object o) throws SerDeException { + if (o == null) { return null; + } else if (o instanceof String) { + return o.toString(); + } else if (o instanceof Long) { + return ((Long) o).toString(); + } else if (o instanceof Integer) { + return ((Integer) o).toString(); + } else if (o instanceof Float) { + return ((Float) o).toString(); + } else if (o instanceof Double) { + return ((Double) o).toString(); + } else if (o instanceof Date) { + SimpleDateFormat simpleDateFormat = getSimpleDateFormat(columnName); + return simpleDateFormat.format((Date) o); + } else { + throw new SerDeException("Unknown type [" + o + "]"); } - byte b = writableObject.get(); - return Integer.toString(b); } - private String lazyBoolean(LazyBoolean lazyBoolean) { - BooleanWritable writableObject = lazyBoolean.getWritableObject(); - if (writableObject == null) { - return null; + private SimpleDateFormat getSimpleDateFormat(String columnName) throws SerDeException { + ThreadLocal<SimpleDateFormat> threadLocal = _dateFormat.get(columnName); + if (threadLocal == null) { + throw new SerDeException("Date format missing for column [" + columnName + "]"); } - return Boolean.toString(writableObject.get()); + return threadLocal.get(); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/1d384701/blur-hive/src/test/java/org/apache/blur/hive/BlurSerDeTest.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/test/java/org/apache/blur/hive/BlurSerDeTest.java b/blur-hive/src/test/java/org/apache/blur/hive/BlurSerDeTest.java new file mode 100644 index 0000000..ad9f92e --- /dev/null +++ b/blur-hive/src/test/java/org/apache/blur/hive/BlurSerDeTest.java @@ -0,0 +1,171 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.blur.hive; + +import static org.junit.Assert.*; + +import java.io.File; +import java.io.IOException; +import java.sql.Date; +import java.text.SimpleDateFormat; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import org.apache.blur.MiniCluster; +import org.apache.blur.mapreduce.lib.BlurColumn; +import org.apache.blur.mapreduce.lib.BlurRecord; +import org.apache.blur.thirdparty.thrift_0_9_0.TException; +import org.apache.blur.thrift.BlurClient; +import org.apache.blur.thrift.generated.Blur.Iface; +import org.apache.blur.thrift.generated.BlurException; +import org.apache.blur.thrift.generated.ColumnDefinition; +import org.apache.blur.thrift.generated.TableDescriptor; +import org.apache.blur.utils.GCWatcher; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.LocalFileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.permission.FsAction; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.hive.serde2.SerDeException; +import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class BlurSerDeTest { + + private static final String YYYYMMDD = "yyyyMMdd"; + private static final String TEST = "test"; + private static final File TMPDIR = new File(System.getProperty("blur.tmp.dir", "./target/tmp_BlurSerDeTest")); + private static MiniCluster miniCluster; + private static boolean externalProcesses = true; + + @BeforeClass + public static void startCluster() throws IOException { + GCWatcher.init(0.60); + LocalFileSystem localFS = FileSystem.getLocal(new Configuration()); + File testDirectory = new File(TMPDIR, "blur-SerDe-test").getAbsoluteFile(); + testDirectory.mkdirs(); + + Path directory = new Path(testDirectory.getPath()); + FsPermission dirPermissions = localFS.getFileStatus(directory).getPermission(); + FsAction userAction = dirPermissions.getUserAction(); + FsAction groupAction = dirPermissions.getGroupAction(); + FsAction otherAction = dirPermissions.getOtherAction(); + + StringBuilder builder = new StringBuilder(); + builder.append(userAction.ordinal()); + builder.append(groupAction.ordinal()); + builder.append(otherAction.ordinal()); + String dirPermissionNum = builder.toString(); + System.setProperty("dfs.datanode.data.dir.perm", dirPermissionNum); + testDirectory.delete(); + miniCluster = new MiniCluster(); + miniCluster.startBlurCluster(new File(testDirectory, "cluster").getAbsolutePath(), 2, 3, true, externalProcesses); + } + + @AfterClass + public static void shutdownCluster() { + miniCluster.shutdownBlurCluster(); + } + + @Before + public void setup() throws BlurException, TException, IOException { + String controllerConnectionStr = miniCluster.getControllerConnectionStr(); + Iface client = BlurClient.getClient(controllerConnectionStr); + List<String> tableList = client.tableList(); + if (!tableList.contains(TEST)) { + TableDescriptor tableDescriptor = new TableDescriptor(); + tableDescriptor.setName(TEST); + tableDescriptor.setShardCount(1); + tableDescriptor.setTableUri(miniCluster.getFileSystemUri().toString() + "/blur/tables/test"); + client.createTable(tableDescriptor); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "string-col", null, false, "string", null, false)); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "text-col", null, false, "text", null, false)); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "stored-col", null, false, "stored", null, false)); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "double-col", null, false, "double", null, false)); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "float-col", null, false, "float", null, false)); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "long-col", null, false, "long", null, false)); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "int-col", null, false, "int", null, false)); + Map<String, String> props = new HashMap<String, String>(); + props.put("dateFormat", YYYYMMDD); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "date-col", null, false, "date", props, false)); + client.addColumnDefinition(TEST, new ColumnDefinition("fam0", "geo-col", null, false, "geo-pointvector", null, + false)); + + } + } + + @Test + public void test1() throws SerDeException { + long now = System.currentTimeMillis(); + Date date = new Date(now); + BlurSerDe blurSerDe = new BlurSerDe(); + + Configuration conf = new Configuration(); + Properties tbl = new Properties(); + tbl.put(BlurSerDe.TABLE, TEST); + tbl.put(BlurSerDe.FAMILY, "fam0"); + tbl.put(BlurSerDe.ZK, miniCluster.getZkConnectionString()); + + blurSerDe.initialize(conf, tbl); + + ObjectInspector objectInspector = blurSerDe.getObjectInspector(); + Object[] row = new Object[11]; + int c = 0; + row[c++] = "rowid"; + row[c++] = "recordid"; + row[c++] = date; + row[c++] = 1234.5678; + row[c++] = 1234.567f; + row[c++] = new Object[] { 1.0f, 2.0 }; + row[c++] = 12345678; + row[c++] = 12345678l; + row[c++] = "stored input"; + row[c++] = "string input"; + row[c++] = "text input"; + + BlurRecord blurRecord = (BlurRecord) blurSerDe.serialize(row, objectInspector); + assertEquals("rowid", blurRecord.getRowId()); + assertEquals("recordid", blurRecord.getRecordId()); + + Map<String, String> columns = toMap(blurRecord.getColumns()); + assertEquals("string input", columns.get("string-col")); + assertEquals("text input", columns.get("text-col")); + assertEquals("stored input", columns.get("stored-col")); + assertEquals("1234.5678", columns.get("double-col")); + assertEquals("1234.567", columns.get("float-col")); + assertEquals("12345678", columns.get("long-col")); + assertEquals("12345678", columns.get("int-col")); + assertEquals("1.0,2.0", columns.get("geo-col")); + SimpleDateFormat simpleDateFormat = new SimpleDateFormat(YYYYMMDD); + assertEquals(simpleDateFormat.format(date), columns.get("date-col")); + } + + private Map<String, String> toMap(List<BlurColumn> columns) { + Map<String, String> map = new HashMap<String, String>(); + for (BlurColumn blurColumn : columns) { + map.put(blurColumn.getName(), blurColumn.getValue()); + } + return map; + } + +}
