You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by di...@apache.org on 2020/05/08 22:18:44 UTC
[giraph] branch trunk updated: GIRAPH-1234
This is an automated email from the ASF dual-hosted git repository.
dionysios pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/giraph.git
The following commit(s) were added to refs/heads/trunk by this push:
new f600ccf GIRAPH-1234
f600ccf is described below
commit f600ccf00f70049dc3ffa7e71ec25cb81d2c25a3
Author: Dionysios Logothetis <dl...@gmail.com>
AuthorDate: Fri May 8 15:18:25 2020 -0700
GIRAPH-1234
closes #120
---
giraph-dist/pom.xml | 4 -
giraph-hcatalog/pom.xml | 211 ---------
giraph-hcatalog/src/main/assembly/compile.xml | 39 --
.../giraph/io/hcatalog/GiraphHCatInputFormat.java | 427 ------------------
.../giraph/io/hcatalog/HCatGiraphRunner.java | 490 ---------------------
.../io/hcatalog/HCatalogEdgeInputFormat.java | 254 -----------
.../io/hcatalog/HCatalogVertexInputFormat.java | 380 ----------------
.../io/hcatalog/HCatalogVertexOutputFormat.java | 229 ----------
.../hcatalog/HCatalogVertexValueInputFormat.java | 166 -------
.../org/apache/giraph/io/hcatalog/HiveUtils.java | 64 ---
.../apache/giraph/io/hcatalog/package-info.java | 23 -
.../org/apache/hcatalog/mapreduce/HCatUtils.java | 189 --------
.../apache/hcatalog/mapreduce/package-info.java | 21 -
.../apache/giraph/io/hcatalog/TestHiveUtils.java | 36 --
pom.xml | 57 ---
15 files changed, 2590 deletions(-)
diff --git a/giraph-dist/pom.xml b/giraph-dist/pom.xml
index c156117..ecdd8a4 100644
--- a/giraph-dist/pom.xml
+++ b/giraph-dist/pom.xml
@@ -103,10 +103,6 @@
</dependency>
<dependency>
<groupId>org.apache.giraph</groupId>
- <artifactId>giraph-hcatalog</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.giraph</groupId>
<artifactId>giraph-examples</artifactId>
</dependency>
<dependency>
diff --git a/giraph-hcatalog/pom.xml b/giraph-hcatalog/pom.xml
deleted file mode 100644
index 77554a8..0000000
--- a/giraph-hcatalog/pom.xml
+++ /dev/null
@@ -1,211 +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.
--->
-
-<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/maven-v4_0_0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.giraph</groupId>
- <artifactId>giraph-parent</artifactId>
- <version>1.3.0-SNAPSHOT</version>
- </parent>
- <artifactId>giraph-hcatalog</artifactId>
- <packaging>jar</packaging>
-
- <name>Apache Giraph HCatalog I/O</name>
- <url>http://giraph.apache.org/giraph-hcatalog/</url>
- <description>Giraph HCatalog input/output classes</description>
-
- <properties>
- <top.dir>${project.basedir}/..</top.dir>
- </properties>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-assembly-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-checkstyle-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-javadoc-plugin</artifactId>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-site-plugin</artifactId>
- <configuration>
- <siteDirectory>${project.basedir}/src/site</siteDirectory>
- </configuration>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <version>2.6</version>
- <configuration>
- <systemProperties>
- <property>
- <name>prop.jarLocation</name>
- <value>${top.dir}/giraph-core/target/giraph-${project.version}-${forHadoop}-jar-with-dependencies.jar</value>
- </property>
- </systemProperties>
- </configuration>
- </plugin>
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>findbugs-maven-plugin</artifactId>
- </plugin>
- </plugins>
- </build>
-
- <profiles>
- <profile>
- <id>hadoop_0.20.203</id>
- <dependencies>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-test</artifactId>
- <version>${hadoop.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- </profile>
-
- <profile>
- <id>hadoop_1</id>
- <activation>
- <activeByDefault>true</activeByDefault>
- </activation>
- <dependencies>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-test</artifactId>
- <version>${hadoop.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- </profile>
-
- <profile>
- <id>hadoop_2</id>
- <dependencies>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-minicluster</artifactId>
- <version>${hadoop.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- </profile>
-
- <profile>
- <id>hadoop_non_secure</id>
- <dependencies>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-test</artifactId>
- <version>${hadoop.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- </profile>
-
- <profile>
- <id>hadoop_facebook</id>
- <dependencies>
- <dependency>
- <groupId>com.facebook.hadoop</groupId>
- <artifactId>hadoop-test</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.jboss.netty</groupId>
- <artifactId>netty</artifactId>
- <version>${dep.oldnetty.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- </profile>
- </profiles>
-
- <dependencies>
- <!-- compile dependencies. sorted lexicographically. -->
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </dependency>
- <dependency>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </dependency>
- <dependency>
- <groupId>commons-cli</groupId>
- <artifactId>commons-cli</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.giraph</groupId>
- <artifactId>giraph-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.hcatalog</groupId>
- <artifactId>hcatalog-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-exec</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-metastore</artifactId>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-serde</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.thrift</groupId>
- <artifactId>libthrift</artifactId>
- <scope>runtime</scope>
- </dependency>
-
- <!-- runtime dependency -->
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- <scope>runtime</scope>
- </dependency>
-
- <!-- test dependencies. sorted lexicographically. -->
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
-</project>
diff --git a/giraph-hcatalog/src/main/assembly/compile.xml b/giraph-hcatalog/src/main/assembly/compile.xml
deleted file mode 100644
index 6acf679..0000000
--- a/giraph-hcatalog/src/main/assembly/compile.xml
+++ /dev/null
@@ -1,39 +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.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
- <id>jar-with-dependencies</id>
- <formats>
- <format>jar</format>
- </formats>
- <includeBaseDirectory>false</includeBaseDirectory>
-
- <dependencySets>
- <dependencySet>
- <useProjectArtifact>true</useProjectArtifact>
- <outputDirectory>/</outputDirectory>
- <unpackOptions>
- <excludes>
- <exclude>META-INF/LICENSE</exclude>
- </excludes>
- </unpackOptions>
- <unpack>true</unpack>
- <scope>runtime</scope>
- </dependencySet>
- </dependencySets>
-</assembly>
\ No newline at end of file
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/GiraphHCatInputFormat.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/GiraphHCatInputFormat.java
deleted file mode 100644
index 2e91cba..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/GiraphHCatInputFormat.java
+++ /dev/null
@@ -1,427 +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.giraph.io.hcatalog;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.util.StringUtils;
-import org.apache.hcatalog.common.HCatConstants;
-import org.apache.hcatalog.common.HCatUtil;
-import org.apache.hcatalog.data.HCatRecord;
-import org.apache.hcatalog.data.schema.HCatFieldSchema;
-import org.apache.hcatalog.data.schema.HCatSchema;
-import org.apache.hcatalog.mapreduce.HCatBaseInputFormat;
-import org.apache.hcatalog.mapreduce.HCatSplit;
-import org.apache.hcatalog.mapreduce.HCatStorageHandler;
-import org.apache.hcatalog.mapreduce.HCatUtils;
-import org.apache.hcatalog.mapreduce.InputJobInfo;
-import org.apache.hcatalog.mapreduce.PartInfo;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Provides functionality similar to
- * {@link org.apache.hcatalog.mapreduce.HCatInputFormat},
- * but allows for different data sources (vertex and edge data).
- */
-public class GiraphHCatInputFormat extends HCatBaseInputFormat {
- /** Vertex input job info for HCatalog. */
- public static final String VERTEX_INPUT_JOB_INFO =
- "giraph.hcat.vertex.input.job.info";
- /** Edge input job info for HCatalog. */
- public static final String EDGE_INPUT_JOB_INFO =
- "giraph.hcat.edge.input.job.info";
-
- /**
- * Set vertex {@link InputJobInfo}.
- *
- * @param job The job
- * @param inputJobInfo Vertex input job info
- * @throws IOException
- */
- public static void setVertexInput(Job job,
- InputJobInfo inputJobInfo)
- throws IOException {
- InputJobInfo vertexInputJobInfo = InputJobInfo.create(
- inputJobInfo.getDatabaseName(),
- inputJobInfo.getTableName(),
- inputJobInfo.getFilter());
- vertexInputJobInfo.getProperties().putAll(inputJobInfo.getProperties());
- Configuration conf = job.getConfiguration();
- conf.set(VERTEX_INPUT_JOB_INFO, HCatUtil.serialize(
- HCatUtils.getInputJobInfo(conf, vertexInputJobInfo)));
- }
-
- /**
- * Set edge {@link InputJobInfo}.
- *
- * @param job The job
- * @param inputJobInfo Edge input job info
- * @throws IOException
- */
- public static void setEdgeInput(Job job,
- InputJobInfo inputJobInfo)
- throws IOException {
- InputJobInfo edgeInputJobInfo = InputJobInfo.create(
- inputJobInfo.getDatabaseName(),
- inputJobInfo.getTableName(),
- inputJobInfo.getFilter());
- edgeInputJobInfo.getProperties().putAll(inputJobInfo.getProperties());
- Configuration conf = job.getConfiguration();
- conf.set(EDGE_INPUT_JOB_INFO, HCatUtil.serialize(
- HCatUtils.getInputJobInfo(conf, edgeInputJobInfo)));
- }
-
- /**
- * Get table schema from input job info.
- *
- * @param inputJobInfo Input job info
- * @return Input table schema
- * @throws IOException
- */
- private static HCatSchema getTableSchema(InputJobInfo inputJobInfo)
- throws IOException {
- HCatSchema allCols = new HCatSchema(new LinkedList<HCatFieldSchema>());
- for (HCatFieldSchema field :
- inputJobInfo.getTableInfo().getDataColumns().getFields()) {
- allCols.append(field);
- }
- for (HCatFieldSchema field :
- inputJobInfo.getTableInfo().getPartitionColumns().getFields()) {
- allCols.append(field);
- }
- return allCols;
- }
-
- /**
- * Get vertex input table schema.
- *
- * @param conf Job configuration
- * @return Vertex input table schema
- * @throws IOException
- */
- public static HCatSchema getVertexTableSchema(Configuration conf)
- throws IOException {
- return getTableSchema(getVertexJobInfo(conf));
- }
-
- /**
- * Get edge input table schema.
- *
- * @param conf Job configuration
- * @return Edge input table schema
- * @throws IOException
- */
- public static HCatSchema getEdgeTableSchema(Configuration conf)
- throws IOException {
- return getTableSchema(getEdgeJobInfo(conf));
- }
-
- /**
- * Set input path for job.
- *
- * @param jobConf Job configuration
- * @param location Location of input files
- * @throws IOException
- */
- private void setInputPath(JobConf jobConf, String location)
- throws IOException {
- int length = location.length();
- int curlyOpen = 0;
- int pathStart = 0;
- boolean globPattern = false;
- List<String> pathStrings = new ArrayList<String>();
-
- for (int i = 0; i < length; i++) {
- char ch = location.charAt(i);
- switch (ch) {
- case '{':
- curlyOpen++;
- if (!globPattern) {
- globPattern = true;
- }
- break;
- case '}':
- curlyOpen--;
- if (curlyOpen == 0 && globPattern) {
- globPattern = false;
- }
- break;
- case ',':
- if (!globPattern) {
- pathStrings.add(location.substring(pathStart, i));
- pathStart = i + 1;
- }
- break;
- default:
- }
- }
- pathStrings.add(location.substring(pathStart, length));
-
- Path[] paths = StringUtils.stringToPath(pathStrings.toArray(new String[0]));
-
- FileSystem fs = FileSystem.get(jobConf);
- Path path = paths[0].makeQualified(fs);
- StringBuilder str = new StringBuilder(StringUtils.escapeString(
- path.toString()));
- for (int i = 1; i < paths.length; i++) {
- str.append(StringUtils.COMMA_STR);
- path = paths[i].makeQualified(fs);
- str.append(StringUtils.escapeString(path.toString()));
- }
-
- jobConf.set("mapred.input.dir", str.toString());
- }
-
- /**
- * Get input splits for job.
- *
- * @param jobContext Job context
- * @param inputJobInfo Input job info
- * @return MapReduce setting for file input directory
- * @throws IOException
- * @throws InterruptedException
- */
- private List<InputSplit> getSplits(JobContext jobContext,
- InputJobInfo inputJobInfo)
- throws IOException, InterruptedException {
- Configuration conf = jobContext.getConfiguration();
-
- List<InputSplit> splits = new ArrayList<InputSplit>();
- List<PartInfo> partitionInfoList = inputJobInfo.getPartitions();
- if (partitionInfoList == null) {
- //No partitions match the specified partition filter
- return splits;
- }
-
- HCatStorageHandler storageHandler;
- JobConf jobConf;
- //For each matching partition, call getSplits on the underlying InputFormat
- for (PartInfo partitionInfo : partitionInfoList) {
- jobConf = HCatUtil.getJobConfFromContext(jobContext);
- setInputPath(jobConf, partitionInfo.getLocation());
- Map<String, String> jobProperties = partitionInfo.getJobProperties();
-
- HCatSchema allCols = new HCatSchema(new LinkedList<HCatFieldSchema>());
- for (HCatFieldSchema field :
- inputJobInfo.getTableInfo().getDataColumns().getFields()) {
- allCols.append(field);
- }
- for (HCatFieldSchema field :
- inputJobInfo.getTableInfo().getPartitionColumns().getFields()) {
- allCols.append(field);
- }
-
- HCatUtil.copyJobPropertiesToJobConf(jobProperties, jobConf);
-
- storageHandler = HCatUtil.getStorageHandler(
- jobConf, partitionInfo);
-
- //Get the input format
- Class inputFormatClass = storageHandler.getInputFormatClass();
- org.apache.hadoop.mapred.InputFormat inputFormat =
- getMapRedInputFormat(jobConf, inputFormatClass);
-
- //Call getSplit on the InputFormat, create an HCatSplit for each
- //underlying split. When the desired number of input splits is missing,
- //use a default number (denoted by zero).
- //TODO: Currently each partition is split independently into
- //a desired number. However, we want the union of all partitions to be
- //split into a desired number while maintaining balanced sizes of input
- //splits.
- int desiredNumSplits =
- conf.getInt(HCatConstants.HCAT_DESIRED_PARTITION_NUM_SPLITS, 0);
- org.apache.hadoop.mapred.InputSplit[] baseSplits =
- inputFormat.getSplits(jobConf, desiredNumSplits);
-
- for (org.apache.hadoop.mapred.InputSplit split : baseSplits) {
- splits.add(new HCatSplit(partitionInfo, split, allCols));
- }
- }
-
- return splits;
- }
-
- /**
- * Get vertex {@link InputJobInfo}.
- *
- * @param conf Configuration
- * @return Vertex input job info
- * @throws IOException
- */
- private static InputJobInfo getVertexJobInfo(Configuration conf)
- throws IOException {
- String jobString = conf.get(VERTEX_INPUT_JOB_INFO);
- if (jobString == null) {
- throw new IOException("Vertex job information not found in JobContext." +
- " GiraphHCatInputFormat.setVertexInput() not called?");
- }
- return (InputJobInfo) HCatUtil.deserialize(jobString);
- }
-
- /**
- * Get edge {@link InputJobInfo}.
- *
- * @param conf Configuration
- * @return Edge input job info
- * @throws IOException
- */
- private static InputJobInfo getEdgeJobInfo(Configuration conf)
- throws IOException {
- String jobString = conf.get(EDGE_INPUT_JOB_INFO);
- if (jobString == null) {
- throw new IOException("Edge job information not found in JobContext." +
- " GiraphHCatInputFormat.setEdgeInput() not called?");
- }
- return (InputJobInfo) HCatUtil.deserialize(jobString);
- }
-
- /**
- * Get vertex input splits.
- *
- * @param jobContext Job context
- * @return List of vertex {@link InputSplit}s
- * @throws IOException
- * @throws InterruptedException
- */
- public List<InputSplit> getVertexSplits(JobContext jobContext)
- throws IOException, InterruptedException {
- return getSplits(jobContext,
- getVertexJobInfo(jobContext.getConfiguration()));
- }
-
- /**
- * Get edge input splits.
- *
- * @param jobContext Job context
- * @return List of edge {@link InputSplit}s
- * @throws IOException
- * @throws InterruptedException
- */
- public List<InputSplit> getEdgeSplits(JobContext jobContext)
- throws IOException, InterruptedException {
- return getSplits(jobContext,
- getEdgeJobInfo(jobContext.getConfiguration()));
- }
-
- /**
- * Create an {@link org.apache.hcatalog.mapreduce.HCatRecordReader}.
- *
- * @param split Input split
- * @param schema Table schema
- * @param taskContext Context
- * @return Record reader
- * @throws IOException
- * @throws InterruptedException
- */
- private RecordReader<WritableComparable, HCatRecord>
- createRecordReader(InputSplit split,
- HCatSchema schema,
- TaskAttemptContext taskContext)
- throws IOException, InterruptedException {
- HCatSplit hcatSplit = HCatUtils.castToHCatSplit(split);
- PartInfo partitionInfo = hcatSplit.getPartitionInfo();
- JobContext jobContext = taskContext;
- Configuration conf = jobContext.getConfiguration();
-
- HCatStorageHandler storageHandler = HCatUtil.getStorageHandler(
- conf, partitionInfo);
-
- JobConf jobConf = HCatUtil.getJobConfFromContext(jobContext);
- Map<String, String> jobProperties = partitionInfo.getJobProperties();
- HCatUtil.copyJobPropertiesToJobConf(jobProperties, jobConf);
-
- Map<String, String> valuesNotInDataCols = getColValsNotInDataColumns(
- schema, partitionInfo);
-
- return HCatUtils.newHCatReader(storageHandler, valuesNotInDataCols);
- }
-
- /**
- * Create a {@link RecordReader} for vertices.
- *
- * @param split Input split
- * @param taskContext Context
- * @return Record reader
- * @throws IOException
- * @throws InterruptedException
- */
- public RecordReader<WritableComparable, HCatRecord>
- createVertexRecordReader(InputSplit split, TaskAttemptContext taskContext)
- throws IOException, InterruptedException {
- return createRecordReader(split, getVertexTableSchema(
- taskContext.getConfiguration()), taskContext);
- }
-
- /**
- * Create a {@link RecordReader} for edges.
- *
- * @param split Input split
- * @param taskContext Context
- * @return Record reader
- * @throws IOException
- * @throws InterruptedException
- */
- public RecordReader<WritableComparable, HCatRecord>
- createEdgeRecordReader(InputSplit split, TaskAttemptContext taskContext)
- throws IOException, InterruptedException {
- return createRecordReader(split, getEdgeTableSchema(
- taskContext.getConfiguration()), taskContext);
- }
-
- /**
- * Get values for fields requested by output schema which will not be in the
- * data.
- *
- * @param outputSchema Output schema
- * @param partInfo Partition info
- * @return Values not in data columns
- */
- private static Map<String, String> getColValsNotInDataColumns(
- HCatSchema outputSchema,
- PartInfo partInfo) {
- HCatSchema dataSchema = partInfo.getPartitionSchema();
- Map<String, String> vals = new HashMap<String, String>();
- for (String fieldName : outputSchema.getFieldNames()) {
- if (dataSchema.getPosition(fieldName) == null) {
- // this entry of output is not present in the output schema
- // so, we first check the table schema to see if it is a part col
- if (partInfo.getPartitionValues().containsKey(fieldName)) {
- vals.put(fieldName, partInfo.getPartitionValues().get(fieldName));
- } else {
- vals.put(fieldName, null);
- }
- }
- }
- return vals;
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatGiraphRunner.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatGiraphRunner.java
deleted file mode 100644
index 25d100b..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatGiraphRunner.java
+++ /dev/null
@@ -1,490 +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.giraph.io.hcatalog;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.giraph.graph.Computation;
-import org.apache.giraph.io.EdgeInputFormat;
-import org.apache.giraph.io.VertexInputFormat;
-import org.apache.giraph.io.VertexOutputFormat;
-import org.apache.giraph.job.GiraphJob;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.util.Tool;
-import org.apache.hadoop.util.ToolRunner;
-import org.apache.hcatalog.mapreduce.HCatOutputFormat;
-import org.apache.hcatalog.mapreduce.InputJobInfo;
-import org.apache.hcatalog.mapreduce.OutputJobInfo;
-import org.apache.log4j.Logger;
-
-import com.google.common.collect.Lists;
-
-import java.io.File;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Hive Giraph Runner
- */
-public class HCatGiraphRunner implements Tool {
- /**
- * logger
- */
- private static final Logger LOG = Logger.getLogger(HCatGiraphRunner.class);
- /**
- * workers
- */
- protected int workers;
- /**
- * is verbose
- */
- protected boolean isVerbose;
- /**
- * output table partitions
- */
- protected Map<String, String> outputTablePartitionValues;
- /**
- * dbName
- */
- protected String dbName;
- /**
- * vertex input table name
- */
- protected String vertexInputTableName;
- /**
- * vertex input table filter
- */
- protected String vertexInputTableFilterExpr;
- /**
- * edge input table name
- */
- protected String edgeInputTableName;
- /**
- * edge input table filter
- */
- protected String edgeInputTableFilterExpr;
- /**
- * output table name
- */
- protected String outputTableName;
- /** Configuration */
- private Configuration conf;
- /** Skip output? (Useful for testing without writing) */
- private boolean skipOutput = false;
-
- /**
- * computation class.
- */
- private Class<? extends Computation> computationClass;
- /**
- * vertex input format internal.
- */
- private Class<? extends VertexInputFormat> vertexInputFormatClass;
- /**
- * edge input format internal.
- */
- private Class<? extends EdgeInputFormat> edgeInputFormatClass;
- /**
- * vertex output format internal.
- */
- private Class<? extends VertexOutputFormat> vertexOutputFormatClass;
-
- /**
- * Giraph runner class.
- *
- * @param computationClass Computation class
- * @param vertexInputFormatClass Vertex input format
- * @param edgeInputFormatClass Edge input format
- * @param vertexOutputFormatClass Output format
- */
- protected HCatGiraphRunner(
- Class<? extends Computation> computationClass,
- Class<? extends VertexInputFormat> vertexInputFormatClass,
- Class<? extends EdgeInputFormat> edgeInputFormatClass,
- Class<? extends VertexOutputFormat> vertexOutputFormatClass) {
- this.computationClass = computationClass;
- this.vertexInputFormatClass = vertexInputFormatClass;
- this.edgeInputFormatClass = edgeInputFormatClass;
- this.vertexOutputFormatClass = vertexOutputFormatClass;
- this.conf = new HiveConf(getClass());
- }
-
- /**
- * main method
- * @param args system arguments
- * @throws Exception any errors from Hive Giraph Runner
- */
- public static void main(String[] args) throws Exception {
- System.exit(ToolRunner.run(
- new HCatGiraphRunner(null, null, null, null), args));
- }
-
- @Override
- public final int run(String[] args) throws Exception {
- // process args
- try {
- processArguments(args);
- } catch (InterruptedException e) {
- return 0;
- } catch (IllegalArgumentException e) {
- System.err.println(e.getMessage());
- return -1;
- }
-
- // additional configuration for Hive
- adjustConfigurationForHive(getConf());
-
- // setup GiraphJob
- GiraphJob job = new GiraphJob(getConf(), getClass().getName());
- job.getConfiguration().setComputationClass(computationClass);
-
- // setup input from Hive
- if (vertexInputFormatClass != null) {
- InputJobInfo vertexInputJobInfo = InputJobInfo.create(dbName,
- vertexInputTableName, vertexInputTableFilterExpr);
- GiraphHCatInputFormat.setVertexInput(job.getInternalJob(),
- vertexInputJobInfo);
- job.getConfiguration().setVertexInputFormatClass(vertexInputFormatClass);
- }
- if (edgeInputFormatClass != null) {
- InputJobInfo edgeInputJobInfo = InputJobInfo.create(dbName,
- edgeInputTableName, edgeInputTableFilterExpr);
- GiraphHCatInputFormat.setEdgeInput(job.getInternalJob(),
- edgeInputJobInfo);
- job.getConfiguration().setEdgeInputFormatClass(edgeInputFormatClass);
- }
-
- // setup output to Hive
- HCatOutputFormat.setOutput(job.getInternalJob(), OutputJobInfo.create(
- dbName, outputTableName, outputTablePartitionValues));
- HCatOutputFormat.setSchema(job.getInternalJob(),
- HCatOutputFormat.getTableSchema(job.getInternalJob()));
- if (skipOutput) {
- LOG.warn("run: Warning - Output will be skipped!");
- } else {
- job.getConfiguration().setVertexOutputFormatClass(
- vertexOutputFormatClass);
- }
-
- job.getConfiguration().setWorkerConfiguration(workers, workers, 100.0f);
- initGiraphJob(job);
-
- return job.run(isVerbose) ? 0 : -1;
- }
-
- /**
- * set hive configuration
- * @param conf Configuration argument
- */
- private static void adjustConfigurationForHive(Configuration conf) {
- // when output partitions are used, workers register them to the
- // metastore at cleanup stage, and on HiveConf's initialization, it
- // looks for hive-site.xml from.
- addToStringCollection(conf, "tmpfiles", conf.getClassLoader()
- .getResource("hive-site.xml").toString());
-
- // Also, you need hive.aux.jars as well
- // addToStringCollection(conf, "tmpjars",
- // conf.getStringCollection("hive.aux.jars.path"));
-
- // Or, more effectively, we can provide all the jars client needed to
- // the workers as well
- String[] hadoopJars = System.getenv("HADOOP_CLASSPATH").split(
- File.pathSeparator);
- List<String> hadoopJarURLs = Lists.newArrayList();
- for (String jarPath : hadoopJars) {
- File file = new File(jarPath);
- if (file.exists() && file.isFile()) {
- String jarURL = file.toURI().toString();
- hadoopJarURLs.add(jarURL);
- }
- }
- addToStringCollection(conf, "tmpjars", hadoopJarURLs);
- }
-
- /**
- * process arguments
- * @param args to process
- * @return CommandLine instance
- * @throws ParseException error parsing arguments
- * @throws InterruptedException interrupted
- */
- private CommandLine processArguments(String[] args) throws ParseException,
- InterruptedException {
- Options options = new Options();
- options.addOption("h", "help", false, "Help");
- options.addOption("v", "verbose", false, "Verbose");
- options.addOption("D", "hiveconf", true,
- "property=value for Hive/Hadoop configuration");
- options.addOption("w", "workers", true, "Number of workers");
- if (computationClass == null) {
- options.addOption(null, "computationClass", true,
- "Giraph Computation class to use");
- }
- if (vertexInputFormatClass == null) {
- options.addOption(null, "vertexInputFormatClass", true,
- "Giraph HCatalogVertexInputFormat class to use");
- }
- if (edgeInputFormatClass == null) {
- options.addOption(null, "edgeInputFormatClass", true,
- "Giraph HCatalogEdgeInputFormat class to use");
- }
-
- if (vertexOutputFormatClass == null) {
- options.addOption(null, "vertexOutputFormatClass", true,
- "Giraph HCatalogVertexOutputFormat class to use");
- }
-
- options.addOption("db", "dbName", true, "Hive database name");
- options.addOption("vi", "vertexInputTable", true,
- "Vertex input table name");
- options.addOption("VI", "vertexInputFilter", true,
- "Vertex input table filter expression (e.g., \"a<2 AND b='two'\"");
- options.addOption("ei", "edgeInputTable", true,
- "Edge input table name");
- options.addOption("EI", "edgeInputFilter", true,
- "Edge input table filter expression (e.g., \"a<2 AND b='two'\"");
- options.addOption("o", "outputTable", true, "Output table name");
- options.addOption("O", "outputPartition", true,
- "Output table partition values (e.g., \"a=1,b=two\")");
- options.addOption("s", "skipOutput", false, "Skip output?");
-
- addMoreOptions(options);
-
- CommandLineParser parser = new GnuParser();
- final CommandLine cmdln = parser.parse(options, args);
- if (args.length == 0 || cmdln.hasOption("help")) {
- new HelpFormatter().printHelp(getClass().getName(), options, true);
- throw new InterruptedException();
- }
-
- // Giraph classes
- if (cmdln.hasOption("computationClass")) {
- computationClass = findClass(cmdln.getOptionValue("computationClass"),
- Computation.class);
- }
- if (cmdln.hasOption("vertexInputFormatClass")) {
- vertexInputFormatClass = findClass(
- cmdln.getOptionValue("vertexInputFormatClass"),
- HCatalogVertexInputFormat.class);
- }
- if (cmdln.hasOption("edgeInputFormatClass")) {
- edgeInputFormatClass = findClass(
- cmdln.getOptionValue("edgeInputFormatClass"),
- HCatalogEdgeInputFormat.class);
- }
-
- if (cmdln.hasOption("vertexOutputFormatClass")) {
- vertexOutputFormatClass = findClass(
- cmdln.getOptionValue("vertexOutputFormatClass"),
- HCatalogVertexOutputFormat.class);
- }
-
- if (cmdln.hasOption("skipOutput")) {
- skipOutput = true;
- }
-
- if (computationClass == null) {
- throw new IllegalArgumentException(
- "Need the Giraph Computation class name (-computationClass) to use");
- }
- if (vertexInputFormatClass == null && edgeInputFormatClass == null) {
- throw new IllegalArgumentException(
- "Need at least one of Giraph VertexInputFormat " +
- "class name (-vertexInputFormatClass) and " +
- "EdgeInputFormat class name (-edgeInputFormatClass)");
- }
- if (vertexOutputFormatClass == null) {
- throw new IllegalArgumentException(
- "Need the Giraph VertexOutputFormat " +
- "class name (-vertexOutputFormatClass) to use");
- }
- if (!cmdln.hasOption("workers")) {
- throw new IllegalArgumentException(
- "Need to choose the number of workers (-w)");
- }
- if (!cmdln.hasOption("vertexInputTable") &&
- vertexInputFormatClass != null) {
- throw new IllegalArgumentException(
- "Need to set the vertex input table name (-vi)");
- }
- if (!cmdln.hasOption("edgeInputTable") &&
- edgeInputFormatClass != null) {
- throw new IllegalArgumentException(
- "Need to set the edge input table name (-ei)");
- }
- if (!cmdln.hasOption("outputTable")) {
- throw new IllegalArgumentException(
- "Need to set the output table name (-o)");
- }
- dbName = cmdln.getOptionValue("dbName", "default");
- vertexInputTableName = cmdln.getOptionValue("vertexInputTable");
- vertexInputTableFilterExpr = cmdln.getOptionValue("vertexInputFilter");
- edgeInputTableName = cmdln.getOptionValue("edgeInputTable");
- edgeInputTableFilterExpr = cmdln.getOptionValue("edgeInputFilter");
- outputTableName = cmdln.getOptionValue("outputTable");
- outputTablePartitionValues = HiveUtils.parsePartitionValues(cmdln
- .getOptionValue("outputPartition"));
- workers = Integer.parseInt(cmdln.getOptionValue("workers"));
- isVerbose = cmdln.hasOption("verbose");
-
- // pick up -hiveconf arguments
- for (String hiveconf : cmdln.getOptionValues("hiveconf")) {
- String[] keyval = hiveconf.split("=", 2);
- if (keyval.length == 2) {
- String name = keyval[0];
- String value = keyval[1];
- if (name.equals("tmpjars") || name.equals("tmpfiles")) {
- addToStringCollection(
- conf, name, value);
- } else {
- conf.set(name, value);
- }
- }
- }
-
- processMoreArguments(cmdln);
-
- return cmdln;
- }
-
- /**
- * add string to collection
- * @param conf Configuration
- * @param name name to add
- * @param values values for collection
- */
- private static void addToStringCollection(Configuration conf, String name,
- String... values) {
- addToStringCollection(conf, name, Arrays.asList(values));
- }
-
- /**
- * add string to collection
- * @param conf Configuration
- * @param name to add
- * @param values values for collection
- */
- private static void addToStringCollection(
- Configuration conf, String name, Collection
- <? extends String> values) {
- Collection<String> tmpfiles = conf.getStringCollection(name);
- tmpfiles.addAll(values);
- conf.setStrings(name, tmpfiles.toArray(new String[tmpfiles.size()]));
- }
-
- /**
- *
- * @param className to find
- * @param base base class
- * @param <T> class type found
- * @return type found
- */
- private <T> Class<? extends T> findClass(String className, Class<T> base) {
- try {
- Class<?> cls = Class.forName(className);
- if (base.isAssignableFrom(cls)) {
- return cls.asSubclass(base);
- }
- return null;
- } catch (ClassNotFoundException e) {
- throw new IllegalArgumentException(className + ": Invalid class name");
- }
- }
-
- @Override
- public final Configuration getConf() {
- return conf;
- }
-
- @Override
- public final void setConf(Configuration conf) {
- this.conf = conf;
- }
-
- /**
- * Override this method to add more command-line options. You can process
- * them by also overriding {@link #processMoreArguments(CommandLine)}.
- *
- * @param options Options
- */
- protected void addMoreOptions(Options options) {
- }
-
- /**
- * Override this method to process additional command-line arguments. You
- * may want to declare additional options by also overriding
- * {@link #addMoreOptions(Options)}.
- *
- * @param cmd Command
- */
- protected void processMoreArguments(CommandLine cmd) {
- }
-
- /**
- * Override this method to do additional setup with the GiraphJob that will
- * run.
- *
- * @param job
- * GiraphJob that is going to run
- */
- protected void initGiraphJob(GiraphJob job) {
- LOG.info(getClass().getSimpleName() + " with");
- String prefix = "\t";
- LOG.info(prefix + "-computationClass=" +
- computationClass.getCanonicalName());
- if (vertexInputFormatClass != null) {
- LOG.info(prefix + "-vertexInputFormatClass=" +
- vertexInputFormatClass.getCanonicalName());
- }
- if (edgeInputFormatClass != null) {
- LOG.info(prefix + "-edgeInputFormatClass=" +
- edgeInputFormatClass.getCanonicalName());
- }
- LOG.info(prefix + "-vertexOutputFormatClass=" +
- vertexOutputFormatClass.getCanonicalName());
- if (vertexInputTableName != null) {
- LOG.info(prefix + "-vertexInputTable=" + vertexInputTableName);
- }
- if (vertexInputTableFilterExpr != null) {
- LOG.info(prefix + "-vertexInputFilter=\"" +
- vertexInputTableFilterExpr + "\"");
- }
- if (edgeInputTableName != null) {
- LOG.info(prefix + "-edgeInputTable=" + edgeInputTableName);
- }
- if (edgeInputTableFilterExpr != null) {
- LOG.info(prefix + "-edgeInputFilter=\"" +
- edgeInputTableFilterExpr + "\"");
- }
- LOG.info(prefix + "-outputTable=" + outputTableName);
- if (outputTablePartitionValues != null) {
- LOG.info(prefix + "-outputPartition=\"" +
- outputTablePartitionValues + "\"");
- }
- LOG.info(prefix + "-workers=" + workers);
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogEdgeInputFormat.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogEdgeInputFormat.java
deleted file mode 100644
index 256076c..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogEdgeInputFormat.java
+++ /dev/null
@@ -1,254 +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.giraph.io.hcatalog;
-
-import java.io.IOException;
-import java.util.List;
-import org.apache.giraph.edge.Edge;
-import org.apache.giraph.edge.EdgeFactory;
-import org.apache.giraph.io.EdgeInputFormat;
-import org.apache.giraph.io.EdgeReader;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hcatalog.data.HCatRecord;
-
-/**
- * HCatalog {@link EdgeInputFormat} for reading edges from Hive/Pig.
- *
- * @param <I> Vertex id
- * @param <E> Edge value
- */
-public abstract class HCatalogEdgeInputFormat<
- I extends WritableComparable,
- E extends Writable>
- extends EdgeInputFormat<I, E> {
- /**
- * HCatalog input format.
- */
- private GiraphHCatInputFormat hCatInputFormat = new GiraphHCatInputFormat();
-
- @Override
- public final List<InputSplit> getSplits(JobContext context,
- int minSplitCountHint)
- throws IOException, InterruptedException {
- return hCatInputFormat.getEdgeSplits(context);
- }
-
- /**
- * Get underlying HCatalog input format. Used for creating readers.
- *
- * @return GiraphHCatInputFormat stored.
- */
- protected GiraphHCatInputFormat getHCatInputFormat() {
- return hCatInputFormat;
- }
-
- /**
- * {@link EdgeReader} for {@link HCatalogEdgeInputFormat}.
- */
- protected abstract static class HCatalogEdgeReader<
- I extends WritableComparable, E extends Writable>
- extends EdgeReader<I, E> {
- /** HCatalog input format to use */
- private final GiraphHCatInputFormat hCatInputFormat;
- /** Internal {@link RecordReader}. */
- private RecordReader<WritableComparable, HCatRecord> hCatRecordReader;
- /** Context passed to initialize. */
- private TaskAttemptContext context;
-
- /**
- * Constructor taking hcat input format to use.
- *
- * @param hCatInputFormat HCatalog input format
- */
- public HCatalogEdgeReader(GiraphHCatInputFormat hCatInputFormat) {
- this.hCatInputFormat = hCatInputFormat;
- }
-
- @Override
- public final void initialize(InputSplit inputSplit,
- TaskAttemptContext context)
- throws IOException, InterruptedException {
- hCatRecordReader =
- hCatInputFormat.createEdgeRecordReader(inputSplit, context);
- hCatRecordReader.initialize(inputSplit, context);
- this.context = context;
- }
-
- @Override
- public boolean nextEdge() throws IOException, InterruptedException {
- return hCatRecordReader.nextKeyValue();
- }
-
- @Override
- public final void close() throws IOException {
- hCatRecordReader.close();
- }
-
- @Override
- public final float getProgress() throws IOException, InterruptedException {
- return hCatRecordReader.getProgress();
- }
-
- /**
- * Get the record reader.
- *
- * @return Record reader to be used for reading.
- */
- protected final RecordReader<WritableComparable, HCatRecord>
- getRecordReader() {
- return hCatRecordReader;
- }
-
- /**
- * Get the context.
- *
- * @return Context passed to initialize.
- */
- protected final TaskAttemptContext getContext() {
- return context;
- }
- }
-
- /**
- * Create {@link EdgeReader}.
- *
- * @return {@link HCatalogEdgeReader} instance.
- */
- protected abstract HCatalogEdgeReader<I, E> createEdgeReader();
-
- @Override
- public EdgeReader<I, E>
- createEdgeReader(InputSplit split, TaskAttemptContext context)
- throws IOException {
- try {
- HCatalogEdgeReader reader = createEdgeReader();
- reader.initialize(split, context);
- return reader;
- } catch (InterruptedException e) {
- throw new IllegalStateException(
- "createEdgeReader: Interrupted creating reader.", e);
- }
- }
-
- /**
- * {@link HCatalogEdgeReader} for tables holding a complete edge
- * in each row.
- */
- protected abstract static class SingleRowHCatalogEdgeReader<
- I extends WritableComparable, E extends Writable>
- extends HCatalogEdgeReader<I, E> {
- /**
- * Constructor
- * @param hCatInputFormat giraph input format to use
- */
- public SingleRowHCatalogEdgeReader(GiraphHCatInputFormat hCatInputFormat) {
- super(hCatInputFormat);
- }
-
- /**
- * Get source vertex id from a record.
- *
- * @param record Input record
- * @return I Source vertex id
- */
- protected abstract I getSourceVertexId(HCatRecord record);
-
- /**
- * Get target vertex id from a record.
- *
- * @param record Input record
- * @return I Target vertex id
- */
- protected abstract I getTargetVertexId(HCatRecord record);
-
- /**
- * Get edge value from a record.
- *
- * @param record Input record
- * @return E Edge value
- */
- protected abstract E getEdgeValue(HCatRecord record);
-
- @Override
- public I getCurrentSourceId() throws IOException, InterruptedException {
- HCatRecord record = getRecordReader().getCurrentValue();
- return getSourceVertexId(record);
- }
-
- @Override
- public Edge<I, E> getCurrentEdge() throws IOException,
- InterruptedException {
- HCatRecord record = getRecordReader().getCurrentValue();
- return EdgeFactory.create(getTargetVertexId(record),
- getEdgeValue(record));
- }
- }
-
- /**
- * {@link HCatalogEdgeReader} for tables holding a complete edge
- * in each row where the edges contain no data other than IDs they point to.
- */
- protected abstract static class SingleRowHCatalogEdgeNoValueReader<
- I extends WritableComparable>
- extends HCatalogEdgeReader<I, NullWritable> {
- /**
- * Constructor
- * @param hCatInputFormat giraph input format to use
- */
- public SingleRowHCatalogEdgeNoValueReader(
- GiraphHCatInputFormat hCatInputFormat) {
- super(hCatInputFormat);
- }
-
- /**
- * Get source vertex id from a record.
- *
- * @param record Input record
- * @return I Source vertex id
- */
- protected abstract I getSourceVertexId(HCatRecord record);
-
- /**
- * Get target vertex id from a record.
- *
- * @param record Input record
- * @return I Target vertex id
- */
- protected abstract I getTargetVertexId(HCatRecord record);
-
- @Override
- public I getCurrentSourceId() throws IOException, InterruptedException {
- HCatRecord record = getRecordReader().getCurrentValue();
- return getSourceVertexId(record);
- }
-
- @Override
- public Edge<I, NullWritable> getCurrentEdge() throws IOException,
- InterruptedException {
- HCatRecord record = getRecordReader().getCurrentValue();
- return EdgeFactory.create(getTargetVertexId(record));
- }
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexInputFormat.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexInputFormat.java
deleted file mode 100644
index edecbff..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexInputFormat.java
+++ /dev/null
@@ -1,380 +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.giraph.io.hcatalog;
-
-import org.apache.giraph.edge.Edge;
-import org.apache.giraph.edge.EdgeFactory;
-import org.apache.giraph.graph.Vertex;
-import org.apache.giraph.io.VertexInputFormat;
-import org.apache.giraph.io.VertexReader;
-import org.apache.giraph.utils.TimedLogger;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hcatalog.data.HCatRecord;
-import org.apache.log4j.Logger;
-
-import com.google.common.collect.Lists;
-
-import java.io.IOException;
-import java.util.List;
-
-/**
- * Abstract class that users should subclass to load data from a Hive or Pig
- * table. You can easily implement a {@link HCatalogVertexReader} by extending
- * either {@link SingleRowHCatalogVertexReader} or
- * {@link MultiRowHCatalogVertexReader} depending on how data for each vertex is
- * stored in the input table.
- * <p>
- * The desired database and table name to load from can be specified via
- * {@link GiraphHCatInputFormat#setVertexInput(org.apache.hadoop.mapreduce.Job,
- * org.apache.hcatalog.mapreduce.InputJobInfo)}
- * as you setup your vertex input format with {@link
- * org.apache.giraph.conf.GiraphConfiguration#setVertexInputFormatClass(Class)
- * }.
- *
- * @param <I> Vertex id
- * @param <V> Vertex value
- * @param <E> Edge value
- */
-
-@SuppressWarnings("rawtypes")
-public abstract class HCatalogVertexInputFormat<
- I extends WritableComparable,
- V extends Writable,
- E extends Writable>
- extends VertexInputFormat<I, V, E> {
- /**
- * HCatalog input format.
- */
- private GiraphHCatInputFormat hCatInputFormat = new GiraphHCatInputFormat();
-
- @Override
- public final List<InputSplit> getSplits(
- final JobContext context, final int minSplitCountHint)
- throws IOException, InterruptedException {
- return hCatInputFormat.getVertexSplits(context);
- }
-
- /**
- * Abstract class that users should subclass
- * based on their specific vertex
- * input. HCatRecord can be parsed to get the
- * required data for implementing
- * getCurrentVertex(). If the vertex spans more
- * than one HCatRecord,
- * nextVertex() should be overwritten to handle that logic as well.
- */
- protected abstract class HCatalogVertexReader
- extends VertexReader<I, V, E> {
- /** Internal HCatRecordReader. */
- private RecordReader<WritableComparable,
- HCatRecord> hCatRecordReader;
- /** Context passed to initialize. */
- private TaskAttemptContext context;
-
- /**
- * Initialize with the HCatRecordReader.
- *
- * @param recordReader internal reader
- */
- private void initialize(
- final RecordReader<
- WritableComparable, HCatRecord>
- recordReader) {
- this.hCatRecordReader = recordReader;
- }
-
- @Override
- public final void initialize(
- final InputSplit inputSplit,
- final TaskAttemptContext ctxt)
- throws IOException, InterruptedException {
- hCatRecordReader.initialize(inputSplit, ctxt);
- this.context = ctxt;
- }
-
- @Override
- public boolean nextVertex() throws IOException, InterruptedException {
- // Users can override this if desired,
- // and a vertex is bigger than
- // a single row.
- return hCatRecordReader.nextKeyValue();
- }
-
- @Override
- public final void close() throws IOException {
- hCatRecordReader.close();
- }
-
- @Override
- public final float getProgress() throws IOException, InterruptedException {
- return hCatRecordReader.getProgress();
- }
-
- /**
- * Get the record reader.
- * @return Record reader to be used for reading.
- */
- protected final RecordReader<WritableComparable, HCatRecord>
- getRecordReader() {
- return hCatRecordReader;
- }
-
- /**
- * Get the context.
- *
- *
- *
- * @return Context passed to initialize.
- */
- protected final TaskAttemptContext getContext() {
- return context;
- }
- }
-
- /**
- * create vertex reader instance.
- * @return HCatalogVertexReader
- */
- protected abstract HCatalogVertexReader createVertexReader();
-
- @Override
- public final VertexReader<I, V, E>
- createVertexReader(final InputSplit split,
- final TaskAttemptContext context)
- throws IOException {
- try {
- HCatalogVertexReader reader = createVertexReader();
- reader.initialize(hCatInputFormat.
- createVertexRecordReader(split, context));
- return reader;
- } catch (InterruptedException e) {
- throw new IllegalStateException(
- "createVertexReader: " +
- "Interrupted creating reader.", e);
- }
- }
-
- /**
- * HCatalogVertexReader for tables holding
- * complete vertex info within each
- * row.
- */
- protected abstract class SingleRowHCatalogVertexReader
- extends HCatalogVertexReader {
- /**
- * 1024 const.
- */
- private static final int BYTE_CONST = 1024;
- /**
- * logger
- */
- private final Logger log =
- Logger.getLogger(SingleRowHCatalogVertexReader.class);
- /**
- * record count.
- */
- private int recordCount = 0;
- /**
- * modulus check counter.
- */
- private final int recordModLimit = 1000;
- /**
- * Timed logger to print every 30 seconds
- */
- private final TimedLogger timedLogger = new TimedLogger(30 * 1000,
- log);
-
- /**
- * get vertex id.
- * @param record hcat record
- * @return I id
- */
- protected abstract I getVertexId(HCatRecord record);
-
- /**
- * get vertex value.
- * @param record hcat record
- * @return V value
- */
- protected abstract V getVertexValue(HCatRecord record);
-
- /**
- * get edges.
- * @param record hcat record
- * @return Edges
- */
- protected abstract Iterable<Edge<I, E>> getEdges(HCatRecord record);
-
- @Override
- public final Vertex<I, V, E> getCurrentVertex()
- throws IOException, InterruptedException {
- HCatRecord record = getRecordReader().getCurrentValue();
- Vertex<I, V, E> vertex = getConf().createVertex();
- vertex.initialize(getVertexId(record), getVertexValue(record),
- getEdges(record));
- ++recordCount;
- if (log.isInfoEnabled() &&
- ((recordCount % recordModLimit) == 0)) {
- // memory usage
- Runtime runtime = Runtime.getRuntime();
- double gb = BYTE_CONST *
- BYTE_CONST *
- BYTE_CONST;
- timedLogger.info(
- "read " + recordCount + " records. Memory: " +
- (runtime.totalMemory() / gb) +
- "GB total = " +
- ((runtime.totalMemory() - runtime.freeMemory()) / gb) +
- "GB used + " + (runtime.freeMemory() / gb) +
- "GB free, " + (runtime.maxMemory() / gb) + "GB max");
- }
- return vertex;
- }
- }
- /**
- * HCatalogVertexReader for tables
- * holding vertex info across multiple rows
- * sorted by vertex id column,
- * so that they appear consecutively to the
- * RecordReader.
- */
- protected abstract class MultiRowHCatalogVertexReader extends
- HCatalogVertexReader {
- /**
- * modulus check counter.
- */
- private static final int RECORD_MOD_LIMIT = 1000;
- /**
- * logger
- */
- private final Logger log =
- Logger.getLogger(MultiRowHCatalogVertexReader.class);
- /**
- * current vertex id.
- */
- private I currentVertexId = null;
- /**
- * current vertex edges.
- */
- private List<Edge<I, E>> currentEdges = Lists.newLinkedList();
- /**
- * record for vertex.
- */
- private List<HCatRecord> recordsForVertex = Lists.newArrayList();
- /**
- * record count.
- */
- private int recordCount = 0;
- /**
- * vertex.
- */
- private Vertex<I, V, E> vertex = null;
- /**
- * Timed logger to print every 30 seconds
- */
- private final TimedLogger timedLogger = new TimedLogger(30 * 1000,
- log);
-
-
- /**
- * get vertex id from record.
- *
- * @param record hcat
- * @return I vertex id
- */
- protected abstract I getVertexId(HCatRecord record);
-
- /**
- * get vertex value from record.
- * @param records all vertex values
- * @return V iterable of record values
- */
- protected abstract V getVertexValue(
- Iterable<HCatRecord> records);
-
- /**
- * get target vertex id from record.
- *
- * @param record hcat
- * @return I vertex id of target.
- */
- protected abstract I getTargetVertexId(HCatRecord record);
-
- /**
- * get edge value from record.
- *
- * @param record hcat.
- * @return E edge value.
- */
- protected abstract E getEdgeValue(HCatRecord record);
-
- @Override
- public final Vertex<I, V, E>
- getCurrentVertex() throws IOException, InterruptedException {
- return vertex;
- }
-
- @Override
- public boolean nextVertex() throws IOException, InterruptedException {
- while (getRecordReader().nextKeyValue()) {
- HCatRecord record = getRecordReader().getCurrentValue();
- if (currentVertexId == null) {
- currentVertexId = getVertexId(record);
- }
- if (currentVertexId.equals(getVertexId(record))) {
- currentEdges.add(EdgeFactory.create(getTargetVertexId(record),
- getEdgeValue(record)));
- recordsForVertex.add(record);
- } else {
- createCurrentVertex();
- if (log.isInfoEnabled() && (recordCount % RECORD_MOD_LIMIT) == 0) {
- timedLogger.info("read " + recordCount);
- }
- currentVertexId = getVertexId(record);
- recordsForVertex.add(record);
- return true;
- }
- }
-
- if (currentEdges.isEmpty()) {
- return false;
- } else {
- createCurrentVertex();
- return true;
- }
- }
-
- /**
- * create current vertex.
- */
- private void createCurrentVertex() {
- vertex = getConf().createVertex();
- vertex.initialize(currentVertexId, getVertexValue(recordsForVertex),
- currentEdges);
- currentEdges.clear();
- recordsForVertex.clear();
- ++recordCount;
- }
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexOutputFormat.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexOutputFormat.java
deleted file mode 100644
index 1d234d1..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexOutputFormat.java
+++ /dev/null
@@ -1,229 +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.giraph.io.hcatalog;
-
-import org.apache.giraph.graph.Vertex;
-import org.apache.giraph.io.VertexOutputFormat;
-import org.apache.giraph.io.VertexWriter;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.OutputCommitter;
-import org.apache.hadoop.mapreduce.RecordWriter;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hcatalog.data.DefaultHCatRecord;
-import org.apache.hcatalog.data.HCatRecord;
-import org.apache.hcatalog.mapreduce.HCatOutputFormat;
-
-import java.io.IOException;
-
-/**
- * Abstract class that users should subclass to store data to Hive or Pig table.
- * You can easily implement a {@link HCatalogVertexWriter} by extending
- * {@link SingleRowHCatalogVertexWriter} or {@link MultiRowHCatalogVertexWriter}
- * depending on how you want to fit your vertices into the output table.
- * <p>
- * The desired database and table name to store to can be specified via
- * {@link HCatOutputFormat#setOutput(org.apache.hadoop.mapreduce.Job,
- * org.apache.hcatalog.mapreduce.OutputJobInfo)}
- * as you setup your vertex output format with
- * {@link org.apache.giraph.conf.GiraphConfiguration}
- * setVertexOutputFormatClass(Class)}. You must create the output table.
- *
- * @param <I> Vertex id
- * @param <V> Vertex value
- * @param <E> Edge value
- */
-@SuppressWarnings("rawtypes")
-public abstract class HCatalogVertexOutputFormat<
- I extends WritableComparable,
- V extends Writable,
- E extends Writable>
- extends VertexOutputFormat<I, V, E> {
- /**
- * hcat output format
- */
- protected HCatOutputFormat hCatOutputFormat = new HCatOutputFormat();
-
- @Override
- public final void checkOutputSpecs(JobContext context) throws IOException,
- InterruptedException {
- hCatOutputFormat.checkOutputSpecs(context);
- }
-
- @Override
- public final OutputCommitter getOutputCommitter(TaskAttemptContext context)
- throws IOException, InterruptedException {
- return hCatOutputFormat.getOutputCommitter(context);
- }
-
- /**
- * Abstract class that users should
- * subclass based on their specific vertex
- * output. Users should implement
- * writeVertex to create a HCatRecord that is
- * valid to for writing by HCatalogRecordWriter.
- *
- * @param <I> Vertex id
- * @param <V> Vertex value
- * @param <E> Edge value
- */
- protected abstract static class HCatalogVertexWriter<
- I extends WritableComparable,
- V extends Writable,
- E extends Writable>
- extends VertexWriter<I, V, E> {
-
- /** Internal HCatRecordWriter */
- private RecordWriter<WritableComparable<?>, HCatRecord> hCatRecordWriter;
- /** Context passed to initialize */
- private TaskAttemptContext context;
-
- /**
- * Initialize with the HCatRecordWriter
- * @param hCatRecordWriter
- * Internal writer
- */
- protected void initialize(
- RecordWriter<WritableComparable<?>,
- HCatRecord> hCatRecordWriter) {
- this.hCatRecordWriter = hCatRecordWriter;
- }
-
- /**
- * Get the record reader.
- * @return Record reader to be used for reading.
- */
- protected RecordWriter<WritableComparable<?>,
- HCatRecord> getRecordWriter() {
- return hCatRecordWriter;
- }
-
- /**
- * Get the context.
- *
- * @return Context passed to initialize.
- */
- protected TaskAttemptContext getContext() {
- return context;
- }
-
- @Override
- public void initialize(TaskAttemptContext context) throws IOException {
- this.context = context;
- }
-
- @Override
- public void close(TaskAttemptContext context) throws IOException,
- InterruptedException {
- hCatRecordWriter.close(context);
- }
-
- }
-
- /**
- * create vertex writer.
- * @return HCatalogVertexWriter
- */
- protected abstract HCatalogVertexWriter<I, V, E> createVertexWriter();
-
- @Override
- public final VertexWriter<I, V, E> createVertexWriter(
- TaskAttemptContext context) throws IOException,
- InterruptedException {
- HCatalogVertexWriter<I, V, E> writer = createVertexWriter();
- writer.initialize(hCatOutputFormat.getRecordWriter(context));
- return writer;
- }
-
- /**
- * HCatalogVertexWriter to write each vertex in each row.
- *
- * @param <I> Vertex id
- * @param <V> Vertex value
- * @param <E> Edge value
- */
- protected abstract static class SingleRowHCatalogVertexWriter<
- I extends WritableComparable,
- V extends Writable,
- E extends Writable>
- extends HCatalogVertexWriter<I, V, E> {
- /**
- * get num columns
- * @return intcolumns
- */
- protected abstract int getNumColumns();
-
- /**
- * fill record
- * @param record to fill
- * @param vertex to populate record
- */
- protected abstract void fillRecord(HCatRecord record,
- Vertex<I, V, E> vertex);
-
- /**
- * create record
- * @param vertex to populate record
- * @return HCatRecord newly created
- */
- protected HCatRecord createRecord(Vertex<I, V, E> vertex) {
- HCatRecord record = new DefaultHCatRecord(getNumColumns());
- fillRecord(record, vertex);
- return record;
- }
-
- @Override
- public final void writeVertex(Vertex<I, V, E> vertex) throws IOException,
- InterruptedException {
- getRecordWriter().write(null, createRecord(vertex));
- }
-
- }
-
- /**
- * HCatalogVertexWriter to write each vertex in multiple rows.
- *
- * @param <I> Vertex id
- * @param <V> Vertex value
- * @param <E> Edge value
- */
- public abstract static class MultiRowHCatalogVertexWriter<
- I extends WritableComparable,
- V extends Writable,
- E extends Writable>
- extends HCatalogVertexWriter<I, V, E> {
- /**
- * create records
- * @param vertex to populate records
- * @return Iterable of records
- */
- protected abstract Iterable<HCatRecord> createRecords(
- Vertex<I, V, E> vertex);
-
- @Override
- public final void writeVertex(Vertex<I, V, E> vertex) throws IOException,
- InterruptedException {
- Iterable<HCatRecord> records = createRecords(vertex);
- for (HCatRecord record : records) {
- getRecordWriter().write(null, record);
- }
- }
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexValueInputFormat.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexValueInputFormat.java
deleted file mode 100644
index 928e40c..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HCatalogVertexValueInputFormat.java
+++ /dev/null
@@ -1,166 +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.giraph.io.hcatalog;
-
-import org.apache.giraph.io.VertexValueInputFormat;
-import org.apache.giraph.io.VertexValueReader;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hcatalog.data.HCatRecord;
-
-import java.io.IOException;
-import java.util.List;
-
-/**
- * HCatalog {@link VertexValueInputFormat} for reading vertex values from
- * Hive/Pig.
- *
- * @param <I> Vertex id
- * @param <V> Vertex value
- */
-public abstract class HCatalogVertexValueInputFormat<I extends
- WritableComparable, V extends Writable>
- extends VertexValueInputFormat<I, V> {
- /**
- * HCatalog input format.
- */
- private GiraphHCatInputFormat hCatInputFormat = new GiraphHCatInputFormat();
-
- @Override
- public List<InputSplit> getSplits(JobContext context, int minSplitCountHint)
- throws IOException, InterruptedException {
- return hCatInputFormat.getVertexSplits(context);
- }
-
- /**
- * {@link VertexValueReader} for {@link HCatalogVertexValueInputFormat}.
- */
- protected abstract class HCatalogVertexValueReader
- extends VertexValueReader<I, V> {
- /** Internal {@link RecordReader}. */
- private RecordReader<WritableComparable, HCatRecord> hCatRecordReader;
- /** Context passed to initialize. */
- private TaskAttemptContext context;
-
- @Override
- public final void initialize(InputSplit inputSplit,
- TaskAttemptContext context)
- throws IOException, InterruptedException {
- super.initialize(inputSplit, context);
- hCatRecordReader =
- hCatInputFormat.createVertexRecordReader(inputSplit, context);
- hCatRecordReader.initialize(inputSplit, context);
- this.context = context;
- }
-
- @Override
- public boolean nextVertex() throws IOException, InterruptedException {
- return hCatRecordReader.nextKeyValue();
- }
-
- @Override
- public final void close() throws IOException {
- hCatRecordReader.close();
- }
-
- @Override
- public final float getProgress() throws IOException, InterruptedException {
- return hCatRecordReader.getProgress();
- }
-
- /**
- * Get the record reader.
- *
- * @return Record reader to be used for reading.
- */
- protected final RecordReader<WritableComparable, HCatRecord>
- getRecordReader() {
- return hCatRecordReader;
- }
-
- /**
- * Get the context.
- *
- * @return Context passed to initialize.
- */
- protected final TaskAttemptContext getContext() {
- return context;
- }
- }
-
- /**
- * Create {@link VertexValueReader}.
-
- * @return {@link HCatalogVertexValueReader} instance.
- */
- protected abstract HCatalogVertexValueReader createVertexValueReader();
-
- @Override
- public final VertexValueReader<I, V>
- createVertexValueReader(InputSplit split, TaskAttemptContext context)
- throws IOException {
- try {
- HCatalogVertexValueReader reader = createVertexValueReader();
- reader.initialize(split, context);
- return reader;
- } catch (InterruptedException e) {
- throw new IllegalStateException(
- "createVertexValueReader: Interrupted creating reader.", e);
- }
- }
-
- /**
- * {@link HCatalogVertexValueReader} for tables holding a complete vertex
- * value in each row.
- */
- protected abstract class SingleRowHCatalogVertexValueReader
- extends HCatalogVertexValueReader {
- /**
- * Get vertex id from a record.
- *
- * @param record Input record
- * @return I Vertex id
- */
- protected abstract I getVertexId(HCatRecord record);
-
- /**
- * Get vertex value from a record.
- *
- * @param record Input record
- * @return V Vertex value
- */
- protected abstract V getVertexValue(HCatRecord record);
-
- @Override
- public final I getCurrentVertexId() throws IOException,
- InterruptedException {
- return getVertexId(getRecordReader().getCurrentValue());
- }
-
- @Override
- public final V getCurrentVertexValue() throws IOException,
- InterruptedException {
- return getVertexValue(getRecordReader().getCurrentValue());
- }
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HiveUtils.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HiveUtils.java
deleted file mode 100644
index fedd92c..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/HiveUtils.java
+++ /dev/null
@@ -1,64 +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.giraph.io.hcatalog;
-
-import com.google.common.base.Splitter;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * Utilities and helpers for working with Hive tables.
- */
-public class HiveUtils {
- // TODO use Hive util class if this is already provided by it
-
- /**
- * Private constructor for helper class.
- */
- private HiveUtils() {
- // Do nothing.
- }
-
- /**
- * @param outputTablePartitionString table partition string
- * @return Map
- */
- public static Map<String, String> parsePartitionValues(
- String outputTablePartitionString) {
- if (outputTablePartitionString == null) {
- return null;
- }
- Splitter commaSplitter = Splitter.on(',').omitEmptyStrings().trimResults();
- Splitter equalSplitter = Splitter.on('=').omitEmptyStrings().trimResults();
- Map<String, String> partitionValues = Maps.newHashMap();
- for (String keyValStr : commaSplitter.split(outputTablePartitionString)) {
- List<String> keyVal = Lists.newArrayList(equalSplitter.split(keyValStr));
- if (keyVal.size() != 2) {
- throw new IllegalArgumentException(
- "Unrecognized partition value format: " +
- outputTablePartitionString);
- }
- partitionValues.put(keyVal.get(0), keyVal.get(1));
- }
- return partitionValues;
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/package-info.java b/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/package-info.java
deleted file mode 100644
index b01e254..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/giraph/io/hcatalog/package-info.java
+++ /dev/null
@@ -1,23 +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 of input and output format classes
- * for loading and storing Hive/Pig data using HCatalog.
- */
-package org.apache.giraph.io.hcatalog;
-
diff --git a/giraph-hcatalog/src/main/java/org/apache/hcatalog/mapreduce/HCatUtils.java b/giraph-hcatalog/src/main/java/org/apache/hcatalog/mapreduce/HCatUtils.java
deleted file mode 100644
index 1f25709..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/hcatalog/mapreduce/HCatUtils.java
+++ /dev/null
@@ -1,189 +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.hcatalog.mapreduce;
-
-import org.apache.giraph.io.hcatalog.GiraphHCatInputFormat;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
-import org.apache.hadoop.hive.metastore.api.MetaException;
-import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
-import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Table;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hcatalog.common.ErrorType;
-import org.apache.hcatalog.common.HCatException;
-import org.apache.hcatalog.common.HCatUtil;
-import org.apache.hcatalog.data.schema.HCatSchema;
-import org.apache.thrift.TException;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-
-/**
- * Utility methods copied from HCatalog because of visibility restrictions.
- */
-public class HCatUtils {
- /**
- * Don't instantiate.
- */
- private HCatUtils() { }
-
- /**
- * Returns the given InputJobInfo after populating with data queried from the
- * metadata service.
- *
- * @param conf Configuration
- * @param inputJobInfo Input job info
- * @return Populated input job info
- * @throws IOException
- */
- public static InputJobInfo getInputJobInfo(
- Configuration conf, InputJobInfo inputJobInfo)
- throws IOException {
- HiveMetaStoreClient client = null;
- HiveConf hiveConf;
- try {
- if (conf != null) {
- hiveConf = HCatUtil.getHiveConf(conf);
- } else {
- hiveConf = new HiveConf(GiraphHCatInputFormat.class);
- }
- client = HCatUtil.getHiveClient(hiveConf);
- Table table = HCatUtil.getTable(client, inputJobInfo.getDatabaseName(),
- inputJobInfo.getTableName());
-
- List<PartInfo> partInfoList = new ArrayList<PartInfo>();
-
- inputJobInfo.setTableInfo(HCatTableInfo.valueOf(table.getTTable()));
- if (table.getPartitionKeys().size() != 0) {
- // Partitioned table
- List<Partition> parts = client.listPartitionsByFilter(
- inputJobInfo.getDatabaseName(),
- inputJobInfo.getTableName(),
- inputJobInfo.getFilter(),
- (short) -1);
-
- if (parts != null) {
- // Default to 100,000 partitions if hive.metastore.maxpartition is not
- // defined
- int maxPart = hiveConf.getInt("hcat.metastore.maxpartitions", 100000);
- if (parts.size() > maxPart) {
- throw new HCatException(ErrorType.ERROR_EXCEED_MAXPART,
- "total number of partitions is " + parts.size());
- }
-
- // Populate partition info
- for (Partition ptn : parts) {
- HCatSchema schema = HCatUtil.extractSchema(
- new org.apache.hadoop.hive.ql.metadata.Partition(table, ptn));
- PartInfo partInfo = extractPartInfo(schema, ptn.getSd(),
- ptn.getParameters(), conf, inputJobInfo);
- partInfo.setPartitionValues(InternalUtil.createPtnKeyValueMap(table,
- ptn));
- partInfoList.add(partInfo);
- }
- }
- } else {
- // Non partitioned table
- HCatSchema schema = HCatUtil.extractSchema(table);
- PartInfo partInfo = extractPartInfo(schema, table.getTTable().getSd(),
- table.getParameters(), conf, inputJobInfo);
- partInfo.setPartitionValues(new HashMap<String, String>());
- partInfoList.add(partInfo);
- }
- inputJobInfo.setPartitions(partInfoList);
- } catch (MetaException e) {
- throw new IOException("Got MetaException", e);
- } catch (NoSuchObjectException e) {
- throw new IOException("Got NoSuchObjectException", e);
- } catch (TException e) {
- throw new IOException("Got TException", e);
- } catch (HiveException e) {
- throw new IOException("Got HiveException", e);
- } finally {
- HCatUtil.closeHiveClientQuietly(client);
- }
- return inputJobInfo;
- }
-
- /**
- * Extract partition info.
- *
- * @param schema Table schema
- * @param sd Storage descriptor
- * @param parameters Parameters
- * @param conf Configuration
- * @param inputJobInfo Input job info
- * @return Partition info
- * @throws IOException
- */
- private static PartInfo extractPartInfo(
- HCatSchema schema, StorageDescriptor sd, Map<String, String> parameters,
- Configuration conf, InputJobInfo inputJobInfo) throws IOException {
- StorerInfo storerInfo = InternalUtil.extractStorerInfo(sd, parameters);
-
- Properties hcatProperties = new Properties();
- HCatStorageHandler storageHandler = HCatUtil.getStorageHandler(conf,
- storerInfo);
-
- // Copy the properties from storageHandler to jobProperties
- Map<String, String> jobProperties =
- HCatUtil.getInputJobProperties(storageHandler, inputJobInfo);
-
- for (Map.Entry<String, String> param : parameters.entrySet()) {
- hcatProperties.put(param.getKey(), param.getValue());
- }
-
- return new PartInfo(schema, storageHandler, sd.getLocation(),
- hcatProperties, jobProperties, inputJobInfo.getTableInfo());
- }
-
- /**
- * Create a new {@link HCatRecordReader}.
- *
- * @param storageHandler Storage handler
- * @param valuesNotInDataCols Values not in data columns
- * @return Record reader
- */
- public static RecordReader newHCatReader(
- HCatStorageHandler storageHandler,
- Map<String, String> valuesNotInDataCols) {
- return new HCatRecordReader(storageHandler, valuesNotInDataCols);
- }
-
- /**
- * Cast an {@link InputSplit} to {@link HCatSplit}.
- *
- * @param split Input split
- * @return {@link HCatSplit}
- * @throws IOException
- */
- public static HCatSplit castToHCatSplit(InputSplit split)
- throws IOException {
- return InternalUtil.castToHCatSplit(split);
- }
-}
diff --git a/giraph-hcatalog/src/main/java/org/apache/hcatalog/mapreduce/package-info.java b/giraph-hcatalog/src/main/java/org/apache/hcatalog/mapreduce/package-info.java
deleted file mode 100644
index e236aaf..0000000
--- a/giraph-hcatalog/src/main/java/org/apache/hcatalog/mapreduce/package-info.java
+++ /dev/null
@@ -1,21 +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 for HCatalog helper utilities.
- */
-package org.apache.hcatalog.mapreduce;
diff --git a/giraph-hcatalog/src/test/java/org/apache/giraph/io/hcatalog/TestHiveUtils.java b/giraph-hcatalog/src/test/java/org/apache/giraph/io/hcatalog/TestHiveUtils.java
deleted file mode 100644
index 421cc28..0000000
--- a/giraph-hcatalog/src/test/java/org/apache/giraph/io/hcatalog/TestHiveUtils.java
+++ /dev/null
@@ -1,36 +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.giraph.io.hcatalog;
-
-import junit.framework.TestCase;
-
-import java.util.Map;
-import org.junit.Test;
-
-public class TestHiveUtils extends TestCase {
- @Test
- public void testParsePartition() {
- String partitionStr = "feature1=2012-10-09, feature2=a1+b2, feature3=ff-gg";
- Map<String, String> partition = HiveUtils.parsePartitionValues(partitionStr);
- assertEquals(3, partition.size());
- assertEquals("2012-10-09", partition.get("feature1"));
- assertEquals("a1+b2", partition.get("feature2"));
- assertEquals("ff-gg", partition.get("feature3"));
- }
-}
diff --git a/pom.xml b/pom.xml
index c70dd75..a14da54 100644
--- a/pom.xml
+++ b/pom.xml
@@ -334,7 +334,6 @@ under the License.
<dep.google.findbugs.version>2.0.2</dep.google.findbugs.version>
<dep.guava.version>21.0</dep.guava.version>
<dep.hbase.version>0.94.16</dep.hbase.version>
- <dep.hcatalog.version>0.5.0-incubating</dep.hcatalog.version>
<dep.hive.version>0.11.0</dep.hive.version>
<dep.hiveio.version>0.26</dep.hiveio.version>
<dep.jaxb-impl.version>2.2.4-1</dep.jaxb-impl.version>
@@ -983,7 +982,6 @@ under the License.
<id>hadoop_1</id>
<modules>
<module>giraph-accumulo</module>
- <module>giraph-hcatalog</module>
<module>giraph-gora</module>
<module>giraph-dist</module>
</modules>
@@ -1135,7 +1133,6 @@ under the License.
<modules>
<module>giraph-accumulo</module>
<module>giraph-hbase</module>
- <module>giraph-hcatalog</module>
<module>giraph-gora</module>
<module>giraph-dist</module>
</modules>
@@ -1575,21 +1572,6 @@ under the License.
</dependency>
<dependency>
<groupId>org.apache.giraph</groupId>
- <artifactId>giraph-hcatalog</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>commons-lang</groupId>
- <artifactId>commons-lang</artifactId>
- </exclusion>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging-api</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.giraph</groupId>
<artifactId>giraph-hbase</artifactId>
<version>${project.version}</version>
</dependency>
@@ -1629,45 +1611,6 @@ under the License.
</exclusions>
</dependency>
<dependency>
- <groupId>org.apache.hcatalog</groupId>
- <artifactId>hcatalog-core</artifactId>
- <version>${dep.hcatalog.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.pig</groupId>
- <artifactId>pig</artifactId>
- </exclusion>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging</artifactId>
- </exclusion>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging-api</artifactId>
- </exclusion>
- <exclusion>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- </exclusion>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-common</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-shims</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>${dep.hive.version}</version>