Moving blur hive project out of contrib to a main project.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/4784a6fa Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/4784a6fa Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/4784a6fa Branch: refs/heads/master Commit: 4784a6fafd3f24b913cab5b04899c6b01d95e3f7 Parents: 72799d8 Author: Aaron McCurry <[email protected]> Authored: Fri Jan 2 14:34:22 2015 -0500 Committer: Aaron McCurry <[email protected]> Committed: Fri Jan 2 14:34:22 2015 -0500 ---------------------------------------------------------------------- blur-hive/pom.xml | 222 ++++++++++++++++++ blur-hive/src/assemble/bin.xml | 16 ++ .../blur/hive/BlurHiveOutputCommitter.java | 86 +++++++ .../apache/blur/hive/BlurHiveOutputFormat.java | 171 ++++++++++++++ .../blur/hive/BlurHiveStorageHandler.java | 78 +++++++ .../blur/hive/BlurObjectInspectorGenerator.java | 144 ++++++++++++ .../java/org/apache/blur/hive/BlurSerDe.java | 148 ++++++++++++ .../org/apache/blur/hive/BlurSerializer.java | 223 ++++++++++++++++++ .../apache/blur/hive/NullHiveInputFormat.java | 74 ++++++ .../java/org/apache/blur/hive/CreateData.java | 52 +++++ .../java/org/apache/blur/hive/RunHiveTest.java | 27 +++ blur-hive/src/test/java/test.hive | 55 +++++ contrib/blur-hive/pom.xml | 232 ------------------- contrib/blur-hive/src/assemble/bin.xml | 16 -- .../blur/hive/BlurHiveOutputCommitter.java | 86 ------- .../apache/blur/hive/BlurHiveOutputFormat.java | 171 -------------- .../blur/hive/BlurHiveStorageHandler.java | 78 ------- .../blur/hive/BlurObjectInspectorGenerator.java | 144 ------------ .../java/org/apache/blur/hive/BlurSerDe.java | 148 ------------ .../org/apache/blur/hive/BlurSerializer.java | 223 ------------------ .../apache/blur/hive/NullHiveInputFormat.java | 74 ------ .../java/org/apache/blur/hive/CreateData.java | 52 ----- .../java/org/apache/blur/hive/RunHiveTest.java | 27 --- contrib/blur-hive/src/test/java/test.hive | 55 ----- pom.xml | 8 +- 25 files changed, 1302 insertions(+), 1308 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/pom.xml ---------------------------------------------------------------------- diff --git a/blur-hive/pom.xml b/blur-hive/pom.xml new file mode 100644 index 0000000..44d4a7f --- /dev/null +++ b/blur-hive/pom.xml @@ -0,0 +1,222 @@ +<?xml version="1.0" encoding="UTF-8" ?> +<!-- 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. --> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <groupId>org.apache.blur</groupId> + <artifactId>blur</artifactId> + <version>0.2.4-incubating-SNAPSHOT</version> + <relativePath>../pom.xml</relativePath> + </parent> + <modelVersion>4.0.0</modelVersion> + <groupId>org.apache.blur</groupId> + <artifactId>blur-hive</artifactId> + <version>${projectVersion}</version> + <packaging>jar</packaging> + <name>Blur Hive</name> + <description>The Blur hive module contains a storage handler and serde for Hive so that it can integrate with Hive.</description> + + <dependencies> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + <version>${slf4j.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + <version>${slf4j.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.hive</groupId> + <artifactId>hive-exec</artifactId> + <version>${hive.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.hive</groupId> + <artifactId>hive-cli</artifactId> + <version>${hive.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.blur</groupId> + <artifactId>blur-thrift</artifactId> + <version>${project.version}</version> + <exclusions> + <exclusion> + <groupId>org.apache.httpcomponents</groupId> + <artifactId>httpclient</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + <version>${log4j.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.blur</groupId> + <artifactId>blur-util</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + <version>${log4j.version}</version> + <scope>provided</scope> + <exclusions> + <exclusion> + <groupId>javax.mail</groupId> + <artifactId>mail</artifactId> + </exclusion> + <exclusion> + <groupId>javax.jms</groupId> + <artifactId>jms</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jdmk</groupId> + <artifactId>jmxtools</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jmx</groupId> + <artifactId>jmxri</artifactId> + </exclusion> + </exclusions> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-compiler-plugin</artifactId> + <configuration> + <source>1.6</source> + <target>1.6</target> + </configuration> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <version>2.2</version> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <createDependencyReducedPom>false</createDependencyReducedPom> + </configuration> + </execution> + </executions> + </plugin> + </plugins> + </build> + <profiles> + <profile> + <id>hadoop1</id> + <activation> + <property> + <name>hadoop1</name> + </property> + </activation> + <dependencies> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-core</artifactId> + <version>${hadoop.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.blur</groupId> + <artifactId>blur-mapred-hadoop1</artifactId> + <version>${project.version}</version> + <exclusions> + <exclusion> + <groupId>org.apache.blur</groupId> + <artifactId>blur-store</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.blur</groupId> + <artifactId>blur-query</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.blur</groupId> + <artifactId>blur-gui</artifactId> + </exclusion> + <exclusion> + <groupId>org.eclipse.jetty</groupId> + <artifactId>jetty-webapp</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-test</artifactId> + <version>${hadoop.version}</version> + <scope>test</scope> + </dependency> + </dependencies> + </profile> + <profile> + <id>hadoop2-mr1</id> + <activation> + <property> + <name>hadoop2-mr1</name> + </property> + </activation> + <dependencies> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-client</artifactId> + <version>${hadoop.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-minicluster</artifactId> + <version>${hadoop.version}</version> + <scope>test</scope> + </dependency> + </dependencies> + </profile> + <profile> + <id>hadoop2</id> + <activation> + <property> + <name>hadoop2</name> + </property> + </activation> + <dependencies> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-client</artifactId> + <version>${hadoop.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-minicluster</artifactId> + <version>${hadoop.version}</version> + <scope>test</scope> + </dependency> + </dependencies> + </profile> + </profiles> + + +</project> http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/assemble/bin.xml ---------------------------------------------------------------------- diff --git a/blur-hive/src/assemble/bin.xml b/blur-hive/src/assemble/bin.xml new file mode 100644 index 0000000..1b65be3 --- /dev/null +++ b/blur-hive/src/assemble/bin.xml @@ -0,0 +1,16 @@ +<!-- 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. --> +<assembly + xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd";> + +</assembly> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java b/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java new file mode 100644 index 0000000..979cabd --- /dev/null +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java @@ -0,0 +1,86 @@ +/** + * 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 org.apache.blur.mapreduce.lib.BlurOutputFormat; +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.TableDescriptor; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapred.JobContext; +import org.apache.hadoop.mapred.OutputCommitter; +import org.apache.hadoop.mapred.TaskAttemptContext; + +public class BlurHiveOutputCommitter extends OutputCommitter { + + @Override + public void setupJob(JobContext jobContext) throws IOException { + } + + @Override + public void setupTask(TaskAttemptContext taskContext) throws IOException { + } + + @Override + public boolean needsTaskCommit(TaskAttemptContext taskContext) throws IOException { + return false; + } + + @Override + public void commitTask(TaskAttemptContext taskContext) throws IOException { + } + + @Override + public void abortTask(TaskAttemptContext taskContext) throws IOException { + + } + + @Override + public void abortJob(JobContext context, int status) throws IOException { + finishBulkJob(context, false); + } + + @Override + public void cleanupJob(JobContext context) throws IOException { + + } + + @Override + public void commitJob(JobContext context) throws IOException { + finishBulkJob(context, true); + } + + private void finishBulkJob(JobContext context, boolean apply) throws IOException { + Configuration configuration = context.getConfiguration(); + String connectionStr = configuration.get(BlurSerDe.BLUR_CONTROLLER_CONNECTION_STR); + Iface client = BlurClient.getClient(connectionStr); + TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(configuration); + String bulkId = BlurHiveOutputFormat.getBulkId(configuration); + try { + client.bulkMutateFinish(tableDescriptor.getName(), bulkId, apply, false); + } catch (BlurException e) { + throw new IOException(e); + } catch (TException e) { + throw new IOException(e); + } + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java b/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java new file mode 100644 index 0000000..5c1187e --- /dev/null +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.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 java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.blur.manager.BlurPartitioner; +import org.apache.blur.mapreduce.lib.BlurColumn; +import org.apache.blur.mapreduce.lib.BlurOutputFormat; +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.Column; +import org.apache.blur.thrift.generated.Record; +import org.apache.blur.thrift.generated.RecordMutation; +import org.apache.blur.thrift.generated.RecordMutationType; +import org.apache.blur.thrift.generated.RowMutation; +import org.apache.blur.thrift.generated.RowMutationType; +import org.apache.blur.thrift.generated.TableDescriptor; +import org.apache.blur.utils.ShardUtil; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.ql.io.HiveOutputFormat; +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.util.Progressable; + +public class BlurHiveOutputFormat implements HiveOutputFormat<Text, BlurRecord> { + + private static final String BLUR_BULK_MUTATE_ID = "blur.bulk.mutate.id"; + + public static String getBulkId(Configuration conf) { + return conf.get(BLUR_BULK_MUTATE_ID); + } + + public static void setBulkId(Configuration conf, String bulkId) { + conf.set(BLUR_BULK_MUTATE_ID, bulkId); + } + + @Override + public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOException { + + } + + @Override + 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 { + + TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(jc); + String conStr = jc.get(BlurSerDe.BLUR_CONTROLLER_CONNECTION_STR); + final Iface controllerClient = BlurClient.getClient(conStr); + final String table = tableDescriptor.getName(); + final int numberOfShardsInTable = tableDescriptor.getShardCount(); + final String bulkId = getBulkId(jc); + return new org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter() { + + private BlurPartitioner _blurPartitioner = new BlurPartitioner(); + private Map<String, List<RowMutation>> _serverBatches = new ConcurrentHashMap<String, List<RowMutation>>(); + private int _capacity = 100; + private Map<String, String> _shardToServerLayout; + + @Override + public void write(Writable w) throws IOException { + BlurRecord blurRecord = (BlurRecord) w; + String rowId = blurRecord.getRowId(); + RowMutation rowMutation = new RowMutation(); + rowMutation.setTable(table); + rowMutation.setRowId(rowId); + rowMutation.setRowMutationType(RowMutationType.UPDATE_ROW); + rowMutation.addToRecordMutations(new RecordMutation(RecordMutationType.REPLACE_ENTIRE_RECORD, + toRecord(blurRecord))); + + try { + String server = getServer(rowId); + List<RowMutation> batch = _serverBatches.get(server); + if (batch == null) { + _serverBatches.put(server, batch = new ArrayList<RowMutation>(_capacity)); + } + batch.add(rowMutation); + checkForFlush(_capacity); + } catch (BlurException e) { + throw new IOException(e); + } catch (TException e) { + throw new IOException(e); + } + } + + @Override + public void close(boolean abort) throws IOException { + try { + checkForFlush(1); + } catch (BlurException e) { + throw new IOException(e); + } catch (TException e) { + throw new IOException(e); + } + } + + private void checkForFlush(int max) throws BlurException, TException { + for (Entry<String, List<RowMutation>> e : _serverBatches.entrySet()) { + String server = e.getKey(); + List<RowMutation> batch = e.getValue(); + if (batch.size() >= max) { + Iface client = BlurClient.getClient(server); + client.bulkMutateAddMultiple(table, bulkId, batch); + batch.clear(); + } + } + } + + private String getServer(String rowId) throws BlurException, TException { + int shard = _blurPartitioner.getShard(rowId, numberOfShardsInTable); + String shardId = ShardUtil.getShardName(shard); + return getServerFromShardId(table, shardId); + } + + private String getServerFromShardId(String table, String shardId) throws BlurException, TException { + if (_shardToServerLayout == null) { + _shardToServerLayout = controllerClient.shardServerLayout(table); + } + return _shardToServerLayout.get(shardId); + } + + }; + } + + protected Record toRecord(BlurRecord blurRecord) { + return new Record(blurRecord.getRecordId(), blurRecord.getFamily(), toColumns(blurRecord.getColumns())); + } + + private List<Column> toColumns(List<BlurColumn> columns) { + List<Column> result = new ArrayList<Column>(); + for (BlurColumn blurColumn : columns) { + result.add(new Column(blurColumn.getName(), blurColumn.getValue())); + } + return result; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java b/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java new file mode 100644 index 0000000..f9ff7a7 --- /dev/null +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java @@ -0,0 +1,78 @@ +/** + * 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.Map; +import java.util.UUID; + +import org.apache.blur.mapreduce.lib.BlurOutputFormat; +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.TableDescriptor; +import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler; +import org.apache.hadoop.hive.ql.plan.TableDesc; +import org.apache.hadoop.hive.serde2.SerDe; +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 + public Class<? extends InputFormat> getInputFormatClass() { + return NullHiveInputFormat.class; + } + + @Override + public Class<? extends OutputFormat> getOutputFormatClass() { + return BlurHiveOutputFormat.class; + } + + @Override + public Class<? extends SerDe> getSerDeClass() { + return BlurSerDe.class; + } + + @Override + public void configureJobConf(TableDesc tableDesc, JobConf jobConf) { + try { + String bulkId = UUID.randomUUID().toString(); + String connectionStr = jobConf.get(BlurSerDe.BLUR_CONTROLLER_CONNECTION_STR); + Iface client = BlurClient.getClient(connectionStr); + TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(jobConf); + client.bulkMutateStart(tableDescriptor.getName(), bulkId); + BlurHiveOutputFormat.setBulkId(jobConf, bulkId); + jobConf.setOutputCommitter(BlurHiveOutputCommitter.class); + } catch (IOException e) { + throw new RuntimeException(e); + } catch (BlurException e) { + throw new RuntimeException(e); + } catch (TException e) { + throw new RuntimeException(e); + } + } + + @Override + public void configureOutputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties) { + + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/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 new file mode 100644 index 0000000..0905cc9 --- /dev/null +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java @@ -0,0 +1,144 @@ +/** + * 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 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; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; + +public class BlurObjectInspectorGenerator { + + 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 + public int compare(ColumnDefinition o1, ColumnDefinition o2) { + return o1.getColumnName().compareTo(o2.getColumnName()); + } + }; + + private ObjectInspector _objectInspector; + private List<String> _columnNames = new ArrayList<String>(); + private List<TypeInfo> _columnTypes = new ArrayList<TypeInfo>(); + + 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)); + } + _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() { + return _objectInspector; + } + + public List<String> getColumnNames() { + return _columnNames; + } + + public List<TypeInfo> getColumnTypes() { + return _columnTypes; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/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 new file mode 100644 index 0000000..b35feec --- /dev/null +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurSerDe.java @@ -0,0 +1,148 @@ +/** + * 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.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.mapreduce.lib.BlurOutputFormat; +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.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; +import org.apache.hadoop.hive.serde2.SerDeException; +import org.apache.hadoop.hive.serde2.SerDeStats; +import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; +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"; + private String _family; + private Map<String, ColumnDefinition> _schema; + private ObjectInspector _objectInspector; + private List<String> _columnNames; + private List<TypeInfo> _columnTypes; + private BlurSerializer _serializer; + + @Override + public void initialize(Configuration conf, Properties tbl) throws SerDeException { + String table = tbl.getProperty(TABLE); + _family = tbl.getProperty(FAMILY); + BlurConfiguration configuration; + try { + configuration = new BlurConfiguration(); + configuration.set(BlurConstants.BLUR_ZOOKEEPER_CONNECTION, + tbl.getProperty(BlurConstants.BLUR_ZOOKEEPER_CONNECTION)); + } catch (IOException e) { + throw new SerDeException(e); + } + + Iface client = BlurClient.getClient(configuration); + Schema schema; + try { + List<String> tableList = client.tableList(); + if (!tableList.contains(table)) { + throw new SerDeException("Table [" + table + "] does not exist."); + } + if (conf != null) { + TableDescriptor tableDescriptor = client.describe(table); + BlurOutputFormat.setTableDescriptor(conf, tableDescriptor); + conf.set(BLUR_CONTROLLER_CONNECTION_STR, getControllerConnectionStr(client)); + } + schema = client.schema(table); + } catch (BlurException e) { + throw new SerDeException(e); + } catch (TException e) { + throw new SerDeException(e); + } catch (IOException e) { + throw new SerDeException(e); + } + + Map<String, ColumnDefinition> columns = schema.getFamilies().get(_family); + if (columns == null) { + throw new SerDeException("Family [" + _family + "] does not exist in table [" + table + "]"); + } + + _schema = new HashMap<String, ColumnDefinition>(); + for (ColumnDefinition columnDefinition : columns.values()) { + String subColumnName = columnDefinition.getSubColumnName(); + if (subColumnName == null) { + _schema.put(columnDefinition.getColumnName(), columnDefinition); + } + } + + BlurObjectInspectorGenerator blurObjectInspectorGenerator = new BlurObjectInspectorGenerator(_schema.values()); + _objectInspector = blurObjectInspectorGenerator.getObjectInspector(); + _columnNames = blurObjectInspectorGenerator.getColumnNames(); + _columnTypes = blurObjectInspectorGenerator.getColumnTypes(); + + _serializer = new BlurSerializer(); + } + + private String getControllerConnectionStr(Iface client) throws BlurException, TException { + List<String> controllerServerList = client.controllerServerList(); + StringBuilder builder = new StringBuilder(); + for (String c : controllerServerList) { + if (builder.length() != 0) { + builder.append(','); + } + builder.append(c); + } + return builder.toString(); + } + + @Override + public Writable serialize(Object o, ObjectInspector oi) throws SerDeException { + return _serializer.serialize(o, oi, _columnNames, _columnTypes, _schema, _family); + } + + @Override + public Object deserialize(Writable writable) throws SerDeException { + throw new RuntimeException("Not Implemented"); + } + + @Override + public ObjectInspector getObjectInspector() throws SerDeException { + return _objectInspector; + } + + @Override + public SerDeStats getSerDeStats() { + return null; + } + + @Override + public Class<? extends Writable> getSerializedClass() { + return BlurRecord.class; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/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 new file mode 100644 index 0000000..b65698f --- /dev/null +++ b/blur-hive/src/main/java/org/apache/blur/hive/BlurSerializer.java @@ -0,0 +1,223 @@ +/** + * 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.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.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.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 { + + 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(); + 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); + 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."); + } else { + blurRecord.addColumn(columnName, stringValue); + } + } + } + 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 lazyString(LazyString o) { + Text writableObject = o.getWritableObject(); + if (writableObject == null) { + return null; + } + 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."); + } + + 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) { + return null; + } + byte b = writableObject.get(); + return Integer.toString(b); + } + + private String lazyBoolean(LazyBoolean lazyBoolean) { + BooleanWritable writableObject = lazyBoolean.getWritableObject(); + if (writableObject == null) { + return null; + } + return Boolean.toString(writableObject.get()); + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java b/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java new file mode 100644 index 0000000..62eb656 --- /dev/null +++ b/blur-hive/src/main/java/org/apache/blur/hive/NullHiveInputFormat.java @@ -0,0 +1,74 @@ +/** + * 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 groovy.lang.Writable; + +import org.apache.hadoop.mapred.InputFormat; +import org.apache.hadoop.mapred.InputSplit; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.RecordReader; +import org.apache.hadoop.mapred.Reporter; + +public class NullHiveInputFormat implements InputFormat<Writable, Writable> { + + @Override + public RecordReader<Writable, Writable> getRecordReader(InputSplit arg0, JobConf arg1, Reporter arg2) + throws IOException { + 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 jobConf, int state) throws IOException { + return new InputSplit[] {}; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/test/java/org/apache/blur/hive/CreateData.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/test/java/org/apache/blur/hive/CreateData.java b/blur-hive/src/test/java/org/apache/blur/hive/CreateData.java new file mode 100644 index 0000000..63bf568 --- /dev/null +++ b/blur-hive/src/test/java/org/apache/blur/hive/CreateData.java @@ -0,0 +1,52 @@ +/** + * 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.io.PrintWriter; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; + +public class CreateData { + + private static final String SEP = new String(new char[] { 1 }); + + public static void main(String[] args) throws IOException { + Path path = new Path("hdfs://localhost:9000/user/hive/warehouse/test.db/input_data/data"); + Configuration configuration = new Configuration(); + FileSystem fileSystem = path.getFileSystem(configuration); + FSDataOutputStream outputStream = fileSystem.create(path); + PrintWriter print = new PrintWriter(outputStream); + int rows = 100000; + for (int i = 0; i < rows; i++) { + String s = Integer.toString(i); + print.print(s); + print.print(SEP); + print.print(s + "-" + System.currentTimeMillis()); + for (int c = 0; c < 10; c++) { + print.print(SEP); + print.print(s); + } + print.println(); + } + print.close(); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/test/java/org/apache/blur/hive/RunHiveTest.java ---------------------------------------------------------------------- diff --git a/blur-hive/src/test/java/org/apache/blur/hive/RunHiveTest.java b/blur-hive/src/test/java/org/apache/blur/hive/RunHiveTest.java new file mode 100644 index 0000000..3451c1c --- /dev/null +++ b/blur-hive/src/test/java/org/apache/blur/hive/RunHiveTest.java @@ -0,0 +1,27 @@ +/** + * 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 org.apache.hadoop.hive.cli.CliDriver; + +public class RunHiveTest { + + public static void main(String[] args) throws Exception { + CliDriver.main(new String[] { "-f", "./src/test/java/test.hive" }); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/blur-hive/src/test/java/test.hive ---------------------------------------------------------------------- diff --git a/blur-hive/src/test/java/test.hive b/blur-hive/src/test/java/test.hive new file mode 100644 index 0000000..5f950a8 --- /dev/null +++ b/blur-hive/src/test/java/test.hive @@ -0,0 +1,55 @@ +-- 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. + +set mapred.job.tracker=localhost:9001; +set hive.metastore.warehouse.dir=hdfs://localhost:9000/user/hive/warehouse; + +add jar file:///Users/amccurry/Development/incubator-blur/contrib/blur-hive/target/blur-hive-0.2.4-incubating-SNAPSHOT-hadoop1.jar; + +create database if not exists test; +use test; + +CREATE TABLE if not exists test +ROW FORMAT SERDE 'org.apache.blur.hive.BlurSerDe' +WITH SERDEPROPERTIES ( + 'blur.zookeeper.connection'='localhost', + 'blur.table'='test_hdfs', + 'blur.family'='fam0' +) +STORED BY 'org.apache.blur.hive.BlurHiveStorageHandler'; + +desc test; + +create table if not exists input_data ( + rowid string, + recordid string, + col0 string, + col1 string, + col2 string, + col3 string, + col4 string, + col5 string, + col6 string, + col7 string, + col8 string, + col9 string +); + +select * from input_data; + +insert overwrite table test select * from input_data; + + + http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/contrib/blur-hive/pom.xml ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/pom.xml b/contrib/blur-hive/pom.xml deleted file mode 100644 index 7a09502..0000000 --- a/contrib/blur-hive/pom.xml +++ /dev/null @@ -1,232 +0,0 @@ -<?xml version="1.0" encoding="UTF-8" ?> -<!-- 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. --> -<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; - xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> - <parent> - <groupId>org.apache.blur</groupId> - <artifactId>blur</artifactId> - <version>0.2.4-incubating-SNAPSHOT</version> - <relativePath>../../pom.xml</relativePath> - </parent> - <modelVersion>4.0.0</modelVersion> - <groupId>org.apache.blur</groupId> - <artifactId>blur-hive</artifactId> - <version>${projectVersion}</version> - <packaging>jar</packaging> - <name>Blur Hive</name> - <description>The Blur hive module contains a storage handler and serde for Hive so that it can integrate with Hive.</description> - - <dependencies> - <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-api</artifactId> - <version>${slf4j.version}</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - <version>${slf4j.version}</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.hive</groupId> - <artifactId>hive-exec</artifactId> - <version>0.13.1</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.hive</groupId> - <artifactId>hive-cli</artifactId> - <version>0.13.1</version> - <scope>test</scope> - </dependency> - <dependency> - <groupId>org.apache.blur</groupId> - <artifactId>blur-thrift</artifactId> - <version>${project.version}</version> - <exclusions> - <exclusion> - <groupId>org.apache.httpcomponents</groupId> - <artifactId>httpclient</artifactId> - </exclusion> - </exclusions> - </dependency> - <dependency> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - <version>${log4j.version}</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.blur</groupId> - <artifactId>blur-util</artifactId> - <version>${project.version}</version> - </dependency> - <dependency> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - <version>${log4j.version}</version> - <scope>provided</scope> - <exclusions> - <exclusion> - <groupId>javax.mail</groupId> - <artifactId>mail</artifactId> - </exclusion> - <exclusion> - <groupId>javax.jms</groupId> - <artifactId>jms</artifactId> - </exclusion> - <exclusion> - <groupId>com.sun.jdmk</groupId> - <artifactId>jmxtools</artifactId> - </exclusion> - <exclusion> - <groupId>com.sun.jmx</groupId> - <artifactId>jmxri</artifactId> - </exclusion> - </exclusions> - </dependency> - </dependencies> - - <build> - <plugins> - <plugin> - <groupId>org.apache.maven.plugins</groupId> - <artifactId>maven-compiler-plugin</artifactId> - <configuration> - <source>1.6</source> - <target>1.6</target> - </configuration> - </plugin> - <plugin> - <groupId>org.apache.maven.plugins</groupId> - <artifactId>maven-shade-plugin</artifactId> - <version>2.2</version> - <executions> - <execution> - <phase>package</phase> - <goals> - <goal>shade</goal> - </goals> - <configuration> - <createDependencyReducedPom>false</createDependencyReducedPom> - </configuration> - </execution> - </executions> - </plugin> - </plugins> - </build> - <profiles> - <profile> - <id>hadoop1</id> - <activation> - <property> - <name>hadoop1</name> - </property> - </activation> - <properties> - <projectVersion>${project.parent.version}-hadoop1</projectVersion> - </properties> - <dependencies> - <dependency> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-core</artifactId> - <version>${hadoop.version}</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.blur</groupId> - <artifactId>blur-mapred-hadoop1</artifactId> - <version>${project.version}</version> - <exclusions> - <exclusion> - <groupId>org.apache.blur</groupId> - <artifactId>blur-store</artifactId> - </exclusion> - <exclusion> - <groupId>org.apache.blur</groupId> - <artifactId>blur-query</artifactId> - </exclusion> - <exclusion> - <groupId>org.apache.blur</groupId> - <artifactId>blur-gui</artifactId> - </exclusion> - <exclusion> - <groupId>org.eclipse.jetty</groupId> - <artifactId>jetty-webapp</artifactId> - </exclusion> - </exclusions> - </dependency> - <dependency> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-test</artifactId> - <version>${hadoop.version}</version> - <scope>test</scope> - </dependency> - </dependencies> - </profile> - <profile> - <id>hadoop2-mr1</id> - <activation> - <property> - <name>hadoop2-mr1</name> - </property> - </activation> - <properties> - <projectVersion>${project.parent.version}-hadoop2-mr1</projectVersion> - </properties> - - <dependencies> - <dependency> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-client</artifactId> - <version>${hadoop.version}</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-minicluster</artifactId> - <version>${hadoop.version}</version> - <scope>test</scope> - </dependency> - </dependencies> - </profile> - <profile> - <id>hadoop2</id> - <activation> - <property> - <name>hadoop2</name> - </property> - </activation> - <properties> - <projectVersion>${project.parent.version}-hadoop2</projectVersion> - </properties> - <dependencies> - <dependency> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-client</artifactId> - <version>${hadoop.version}</version> - <scope>provided</scope> - </dependency> - <dependency> - <groupId>org.apache.hadoop</groupId> - <artifactId>hadoop-minicluster</artifactId> - <version>${hadoop.version}</version> - <scope>test</scope> - </dependency> - </dependencies> - </profile> - </profiles> - - -</project> http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/contrib/blur-hive/src/assemble/bin.xml ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/assemble/bin.xml b/contrib/blur-hive/src/assemble/bin.xml deleted file mode 100644 index 1b65be3..0000000 --- a/contrib/blur-hive/src/assemble/bin.xml +++ /dev/null @@ -1,16 +0,0 @@ -<!-- 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. --> -<assembly - xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"; - xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; - xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd";> - -</assembly> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java ---------------------------------------------------------------------- diff --git a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java b/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java deleted file mode 100644 index 979cabd..0000000 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputCommitter.java +++ /dev/null @@ -1,86 +0,0 @@ -/** - * 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 org.apache.blur.mapreduce.lib.BlurOutputFormat; -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.TableDescriptor; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapred.JobContext; -import org.apache.hadoop.mapred.OutputCommitter; -import org.apache.hadoop.mapred.TaskAttemptContext; - -public class BlurHiveOutputCommitter extends OutputCommitter { - - @Override - public void setupJob(JobContext jobContext) throws IOException { - } - - @Override - public void setupTask(TaskAttemptContext taskContext) throws IOException { - } - - @Override - public boolean needsTaskCommit(TaskAttemptContext taskContext) throws IOException { - return false; - } - - @Override - public void commitTask(TaskAttemptContext taskContext) throws IOException { - } - - @Override - public void abortTask(TaskAttemptContext taskContext) throws IOException { - - } - - @Override - public void abortJob(JobContext context, int status) throws IOException { - finishBulkJob(context, false); - } - - @Override - public void cleanupJob(JobContext context) throws IOException { - - } - - @Override - public void commitJob(JobContext context) throws IOException { - finishBulkJob(context, true); - } - - private void finishBulkJob(JobContext context, boolean apply) throws IOException { - Configuration configuration = context.getConfiguration(); - String connectionStr = configuration.get(BlurSerDe.BLUR_CONTROLLER_CONNECTION_STR); - Iface client = BlurClient.getClient(connectionStr); - TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(configuration); - String bulkId = BlurHiveOutputFormat.getBulkId(configuration); - try { - client.bulkMutateFinish(tableDescriptor.getName(), bulkId, apply, false); - } catch (BlurException e) { - throw new IOException(e); - } catch (TException e) { - throw new IOException(e); - } - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/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 deleted file mode 100644 index 5c1187e..0000000 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveOutputFormat.java +++ /dev/null @@ -1,171 +0,0 @@ -/** - * 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.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; -import java.util.Properties; -import java.util.concurrent.ConcurrentHashMap; - -import org.apache.blur.manager.BlurPartitioner; -import org.apache.blur.mapreduce.lib.BlurColumn; -import org.apache.blur.mapreduce.lib.BlurOutputFormat; -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.Column; -import org.apache.blur.thrift.generated.Record; -import org.apache.blur.thrift.generated.RecordMutation; -import org.apache.blur.thrift.generated.RecordMutationType; -import org.apache.blur.thrift.generated.RowMutation; -import org.apache.blur.thrift.generated.RowMutationType; -import org.apache.blur.thrift.generated.TableDescriptor; -import org.apache.blur.utils.ShardUtil; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hive.ql.io.HiveOutputFormat; -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.util.Progressable; - -public class BlurHiveOutputFormat implements HiveOutputFormat<Text, BlurRecord> { - - private static final String BLUR_BULK_MUTATE_ID = "blur.bulk.mutate.id"; - - public static String getBulkId(Configuration conf) { - return conf.get(BLUR_BULK_MUTATE_ID); - } - - public static void setBulkId(Configuration conf, String bulkId) { - conf.set(BLUR_BULK_MUTATE_ID, bulkId); - } - - @Override - public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOException { - - } - - @Override - 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 { - - TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(jc); - String conStr = jc.get(BlurSerDe.BLUR_CONTROLLER_CONNECTION_STR); - final Iface controllerClient = BlurClient.getClient(conStr); - final String table = tableDescriptor.getName(); - final int numberOfShardsInTable = tableDescriptor.getShardCount(); - final String bulkId = getBulkId(jc); - return new org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter() { - - private BlurPartitioner _blurPartitioner = new BlurPartitioner(); - private Map<String, List<RowMutation>> _serverBatches = new ConcurrentHashMap<String, List<RowMutation>>(); - private int _capacity = 100; - private Map<String, String> _shardToServerLayout; - - @Override - public void write(Writable w) throws IOException { - BlurRecord blurRecord = (BlurRecord) w; - String rowId = blurRecord.getRowId(); - RowMutation rowMutation = new RowMutation(); - rowMutation.setTable(table); - rowMutation.setRowId(rowId); - rowMutation.setRowMutationType(RowMutationType.UPDATE_ROW); - rowMutation.addToRecordMutations(new RecordMutation(RecordMutationType.REPLACE_ENTIRE_RECORD, - toRecord(blurRecord))); - - try { - String server = getServer(rowId); - List<RowMutation> batch = _serverBatches.get(server); - if (batch == null) { - _serverBatches.put(server, batch = new ArrayList<RowMutation>(_capacity)); - } - batch.add(rowMutation); - checkForFlush(_capacity); - } catch (BlurException e) { - throw new IOException(e); - } catch (TException e) { - throw new IOException(e); - } - } - - @Override - public void close(boolean abort) throws IOException { - try { - checkForFlush(1); - } catch (BlurException e) { - throw new IOException(e); - } catch (TException e) { - throw new IOException(e); - } - } - - private void checkForFlush(int max) throws BlurException, TException { - for (Entry<String, List<RowMutation>> e : _serverBatches.entrySet()) { - String server = e.getKey(); - List<RowMutation> batch = e.getValue(); - if (batch.size() >= max) { - Iface client = BlurClient.getClient(server); - client.bulkMutateAddMultiple(table, bulkId, batch); - batch.clear(); - } - } - } - - private String getServer(String rowId) throws BlurException, TException { - int shard = _blurPartitioner.getShard(rowId, numberOfShardsInTable); - String shardId = ShardUtil.getShardName(shard); - return getServerFromShardId(table, shardId); - } - - private String getServerFromShardId(String table, String shardId) throws BlurException, TException { - if (_shardToServerLayout == null) { - _shardToServerLayout = controllerClient.shardServerLayout(table); - } - return _shardToServerLayout.get(shardId); - } - - }; - } - - protected Record toRecord(BlurRecord blurRecord) { - return new Record(blurRecord.getRecordId(), blurRecord.getFamily(), toColumns(blurRecord.getColumns())); - } - - private List<Column> toColumns(List<BlurColumn> columns) { - List<Column> result = new ArrayList<Column>(); - for (BlurColumn blurColumn : columns) { - result.add(new Column(blurColumn.getName(), blurColumn.getValue())); - } - return result; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/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 deleted file mode 100644 index f9ff7a7..0000000 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurHiveStorageHandler.java +++ /dev/null @@ -1,78 +0,0 @@ -/** - * 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.Map; -import java.util.UUID; - -import org.apache.blur.mapreduce.lib.BlurOutputFormat; -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.TableDescriptor; -import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler; -import org.apache.hadoop.hive.ql.plan.TableDesc; -import org.apache.hadoop.hive.serde2.SerDe; -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 - public Class<? extends InputFormat> getInputFormatClass() { - return NullHiveInputFormat.class; - } - - @Override - public Class<? extends OutputFormat> getOutputFormatClass() { - return BlurHiveOutputFormat.class; - } - - @Override - public Class<? extends SerDe> getSerDeClass() { - return BlurSerDe.class; - } - - @Override - public void configureJobConf(TableDesc tableDesc, JobConf jobConf) { - try { - String bulkId = UUID.randomUUID().toString(); - String connectionStr = jobConf.get(BlurSerDe.BLUR_CONTROLLER_CONNECTION_STR); - Iface client = BlurClient.getClient(connectionStr); - TableDescriptor tableDescriptor = BlurOutputFormat.getTableDescriptor(jobConf); - client.bulkMutateStart(tableDescriptor.getName(), bulkId); - BlurHiveOutputFormat.setBulkId(jobConf, bulkId); - jobConf.setOutputCommitter(BlurHiveOutputCommitter.class); - } catch (IOException e) { - throw new RuntimeException(e); - } catch (BlurException e) { - throw new RuntimeException(e); - } catch (TException e) { - throw new RuntimeException(e); - } - } - - @Override - public void configureOutputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties) { - - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4784a6fa/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 deleted file mode 100644 index 0905cc9..0000000 --- a/contrib/blur-hive/src/main/java/org/apache/blur/hive/BlurObjectInspectorGenerator.java +++ /dev/null @@ -1,144 +0,0 @@ -/** - * 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 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; -import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; - -public class BlurObjectInspectorGenerator { - - 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 - public int compare(ColumnDefinition o1, ColumnDefinition o2) { - return o1.getColumnName().compareTo(o2.getColumnName()); - } - }; - - private ObjectInspector _objectInspector; - private List<String> _columnNames = new ArrayList<String>(); - private List<TypeInfo> _columnTypes = new ArrayList<TypeInfo>(); - - 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)); - } - _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() { - return _objectInspector; - } - - public List<String> getColumnNames() { - return _columnNames; - } - - public List<TypeInfo> getColumnTypes() { - return _columnTypes; - } - -}
