More hive integration updates.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/3ffdd123 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/3ffdd123 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/3ffdd123 Branch: refs/heads/blur-384-random-port-cleanup Commit: 3ffdd1239a22624ec100c04d62e8b759fa288ec6 Parents: 3c41ef1 Author: Aaron McCurry <[email protected]> Authored: Thu Oct 2 21:54:55 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Thu Oct 2 21:54:55 2014 -0400 ---------------------------------------------------------------------- .../apache/blur/hive/BlurHiveOutputFormat.java | 63 ++++++--- .../blur/hive/BlurHiveStorageHandler.java | 3 +- .../blur/hive/BlurObjectInspectorGenerator.java | 136 +++++++++++++++---- .../java/org/apache/blur/hive/BlurSerDe.java | 36 +++-- .../org/apache/blur/hive/BlurSerializer.java | 19 ++- .../apache/blur/hive/NullHiveInputFormat.java | 71 +++++++--- 6 files changed, 232 insertions(+), 96 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3ffdd123/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java index 21a01b0..2a7368a 100644 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java +++ b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java @@ -1,9 +1,28 @@ +/** + * 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 java.io.IOException; import java.util.Properties; +import org.apache.blur.mapreduce.lib.BlurMutate; +import org.apache.blur.mapreduce.lib.BlurMutate.MUTATE_TYPE; import org.apache.blur.mapreduce.lib.BlurRecord; +import org.apache.blur.mapreduce.lib.GenericBlurRecordWriter; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.ql.io.HiveOutputFormat; @@ -11,43 +30,43 @@ import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.RecordWriter; +import org.apache.hadoop.mapreduce.TaskAttemptID; import org.apache.hadoop.util.Progressable; -/** - * 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. - */ - public class BlurHiveOutputFormat implements HiveOutputFormat<Text, BlurRecord> { @Override - public void checkOutputSpecs(FileSystem arg0, JobConf arg1) throws IOException { + public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOException { throw new RuntimeException("Not Implemented"); } @Override - public RecordWriter<Text, BlurRecord> getRecordWriter(FileSystem arg0, JobConf arg1, String arg2, Progressable arg3) - throws IOException { - throw new RuntimeException("Not Implemented"); + public RecordWriter<Text, BlurRecord> getRecordWriter(FileSystem fileSystem, JobConf jobConf, String name, + Progressable progressable) throws IOException { + throw new RuntimeException("Should never be called."); } @Override public org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter getHiveRecordWriter(JobConf jc, Path finalOutPath, Class<? extends Writable> valueClass, boolean isCompressed, Properties tableProperties, Progressable progress) throws IOException { - throw new RuntimeException("Not Implemented"); + TaskAttemptID taskAttemptID = TaskAttemptID.forName(jc.get("mapred.task.id")); + int id = taskAttemptID.getTaskID().getId(); + final GenericBlurRecordWriter writer = new GenericBlurRecordWriter(jc, id, taskAttemptID.toString() + ".tmp"); + return new org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter() { + + @Override + public void write(Writable w) throws IOException { + BlurRecord blurRecord = (BlurRecord) w; + BlurMutate blurMutate = new BlurMutate(MUTATE_TYPE.REPLACE, blurRecord); + writer.write(new Text(blurRecord.getRowId()), blurMutate); + } + + @Override + public void close(boolean abort) throws IOException { + writer.close(); + } + }; } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3ffdd123/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java index 727276f..5053ff9 100644 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java +++ b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java @@ -23,6 +23,7 @@ import org.apache.hadoop.mapred.InputFormat; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.OutputFormat; +@SuppressWarnings({ "rawtypes", "deprecation" }) public class BlurHiveStorageHandler extends DefaultStorageHandler { @Override @@ -45,6 +46,4 @@ public class BlurHiveStorageHandler extends DefaultStorageHandler { // Will set setup Table Descriptor and Output Committer. } - - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3ffdd123/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 9fbea50..98ca4df 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 @@ -1,45 +1,135 @@ +/** + * 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 java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; import java.util.List; -import java.util.Map; 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.ObjectInspectorFactory; +import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory; +import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo; +import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; - -/** - * 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. - */ +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; public class BlurObjectInspectorGenerator { - public BlurObjectInspectorGenerator(Map<String, ColumnDefinition> schema) { - // TODO Auto-generated constructor stub + 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"; + + private static final Comparator<ColumnDefinition> COMPARATOR = new Comparator<ColumnDefinition>() { + @Override + public int compare(ColumnDefinition o1, ColumnDefinition o2) { + return o1.getColumnName().compareTo(o2.getColumnName()); + } + }; + + private ObjectInspector _objectInspector; + private List<String> _columnNames; + private List<TypeInfo> _columnTypes; + + public BlurObjectInspectorGenerator(Collection<ColumnDefinition> colDefs) throws SerDeException { + List<ColumnDefinition> colDefList = new ArrayList<ColumnDefinition>(colDefs); + Collections.sort(colDefList, COMPARATOR); + for (ColumnDefinition columnDefinition : colDefList) { + _columnNames.add(columnDefinition.getColumnName()); + _columnTypes.add(getTypeInfo(columnDefinition)); + } + _objectInspector = createObjectInspector(); + } + + private ObjectInspector createObjectInspector() throws SerDeException { + List<ObjectInspector> columnObjectInspectorList = new ArrayList<ObjectInspector>(_columnNames.size()); + for (int i = 0; i < _columnNames.size(); i++) { + columnObjectInspectorList.add(i, createObjectInspectorWorker(_columnTypes.get(i))); + } + return ObjectInspectorFactory.getStandardStructObjectInspector(_columnNames, columnObjectInspectorList); + } + + private ObjectInspector createObjectInspectorWorker(TypeInfo ti) throws SerDeException { + switch (ti.getCategory()) { + case PRIMITIVE: + PrimitiveTypeInfo pti = (PrimitiveTypeInfo) ti; + return PrimitiveObjectInspectorFactory.getPrimitiveJavaObjectInspector(pti); + case STRUCT: + StructTypeInfo sti = (StructTypeInfo) ti; + ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>(sti.getAllStructFieldTypeInfos().size()); + for (TypeInfo typeInfo : sti.getAllStructFieldTypeInfos()) { + ois.add(createObjectInspectorWorker(typeInfo)); + } + return ObjectInspectorFactory.getStandardStructObjectInspector(sti.getAllStructFieldNames(), ois); + default: + throw new SerDeException("No Hive categories matched for [" + ti + "]"); + } + } + + private TypeInfo getTypeInfo(ColumnDefinition columnDefinition) throws SerDeException { + String fieldType = columnDefinition.getFieldType(); + if (fieldType.equals(TEXT) || fieldType.equals(STRING) || fieldType.equals(STORED)) { + return TypeInfoFactory.stringTypeInfo; + } else if (fieldType.equals(LONG)) { + return TypeInfoFactory.longTypeInfo; + } else if (fieldType.equals(INT)) { + return TypeInfoFactory.intTypeInfo; + } else if (fieldType.equals(FLOAT)) { + return TypeInfoFactory.floatTypeInfo; + } else if (fieldType.equals(DOUBLE)) { + 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); + } + throw new SerDeException("Blur Field Type [" + fieldType + "] is not supported."); } public ObjectInspector getObjectInspector() { - throw new RuntimeException("Not Implemented"); + return _objectInspector; } public List<String> getColumnNames() { - throw new RuntimeException("Not Implemented"); + return _columnNames; } public List<TypeInfo> getColumnTypes() { - throw new RuntimeException("Not Implemented"); + return _columnTypes; } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3ffdd123/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java index 5581a61..cc8e760 100644 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java +++ b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java @@ -1,3 +1,19 @@ +/** + * 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 java.io.IOException; @@ -14,7 +30,6 @@ 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.Schema; -import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.utils.BlurConstants; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hive.serde2.AbstractSerDe; @@ -24,23 +39,6 @@ import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; import org.apache.hadoop.io.Writable; -/** - * 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. - */ - public class BlurSerDe extends AbstractSerDe { private static final String FAMILY = "blur.family"; @@ -94,7 +92,7 @@ public class BlurSerDe extends AbstractSerDe { } } - BlurObjectInspectorGenerator blurObjectInspectorGenerator = new BlurObjectInspectorGenerator(_schema); + BlurObjectInspectorGenerator blurObjectInspectorGenerator = new BlurObjectInspectorGenerator(_schema.values()); _objectInspector = blurObjectInspectorGenerator.getObjectInspector(); _columnNames = blurObjectInspectorGenerator.getColumnNames(); _columnTypes = blurObjectInspectorGenerator.getColumnTypes(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3ffdd123/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 2a3199c..f656d85 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 @@ -1,13 +1,3 @@ -package org.apache.blur.hive; - -import java.util.List; -import java.util.Map; - -import org.apache.blur.thrift.generated.ColumnDefinition; -import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; -import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; -import org.apache.hadoop.io.Writable; - /** * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with this @@ -24,6 +14,15 @@ import org.apache.hadoop.io.Writable; * License for the specific language governing permissions and limitations under * the License. */ +package org.apache.blur.hive; + +import java.util.List; +import java.util.Map; + +import org.apache.blur.thrift.generated.ColumnDefinition; +import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; +import org.apache.hadoop.io.Writable; public class BlurSerializer { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3ffdd123/contrib/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java b/contrib/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java index feb02e4..62eb656 100644 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java +++ b/contrib/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java @@ -1,3 +1,19 @@ +/** + * 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 java.io.IOException; @@ -10,34 +26,49 @@ import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.RecordReader; import org.apache.hadoop.mapred.Reporter; -/** - * 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. - */ - public class NullHiveInputFormat implements InputFormat<Writable, Writable> { @Override public RecordReader<Writable, Writable> getRecordReader(InputSplit arg0, JobConf arg1, Reporter arg2) throws IOException { - throw new RuntimeException("Not Implemented"); + return new RecordReader<Writable, Writable>() { + + @Override + public void close() throws IOException { + + } + + @Override + public Writable createKey() { + return null; + } + + @Override + public Writable createValue() { + return null; + } + + @Override + public long getPos() throws IOException { + return 0l; + } + + @Override + public float getProgress() throws IOException { + return 0.0f; + } + + @Override + public boolean next(Writable key, Writable value) throws IOException { + return false; + } + + }; } @Override - public InputSplit[] getSplits(JobConf arg0, int arg1) throws IOException { - throw new RuntimeException("Not Implemented"); + public InputSplit[] getSplits(JobConf jobConf, int state) throws IOException { + return new InputSplit[] {}; } }
