You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gora.apache.org by en...@apache.org on 2010/11/08 06:27:48 UTC
svn commit: r1032463 - in /incubator/gora/trunk: ./
gora-core/src/main/java/org/apache/gora/mapreduce/
Author: enis
Date: Mon Nov 8 05:27:47 2010
New Revision: 1032463
URL: http://svn.apache.org/viewvc?rev=1032463&view=rev
Log:
GORA-6. Add methods that take dataStoreClass instead of data store instance to Gora{Input|Output}Format and Gora{Mapper|Reducer}
Modified:
incubator/gora/trunk/CHANGES.txt
incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java
incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraMapper.java
incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraOutputFormat.java
incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraReducer.java
Modified: incubator/gora/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/incubator/gora/trunk/CHANGES.txt?rev=1032463&r1=1032462&r2=1032463&view=diff
==============================================================================
--- incubator/gora/trunk/CHANGES.txt (original)
+++ incubator/gora/trunk/CHANGES.txt Mon Nov 8 05:27:47 2010
@@ -5,3 +5,6 @@ Trunk (unreleased changes)
* INFRA-3038. Initial import of code.
* GORA-1. Organize source code for Apache. (enis)
+
+* GORA-6. Add methods that take dataStoreClass instead of data store instance
+ to Gora{Input|Output}Format and Gora{Mapper|Reducer}. (enis)
Modified: incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java
URL: http://svn.apache.org/viewvc/incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java?rev=1032463&r1=1032462&r2=1032463&view=diff
==============================================================================
--- incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java (original)
+++ incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java Mon Nov 8 05:27:47 2010
@@ -26,6 +26,7 @@ import org.apache.gora.query.PartitionQu
import org.apache.gora.query.Query;
import org.apache.gora.query.impl.FileSplitPartitionQuery;
import org.apache.gora.store.DataStore;
+import org.apache.gora.store.DataStoreFactory;
import org.apache.gora.store.FileBackedDataStore;
import org.apache.gora.util.IOUtils;
import org.apache.hadoop.conf.Configurable;
@@ -45,6 +46,11 @@ import org.apache.hadoop.mapreduce.lib.i
*
* <p> The {@link InputSplit}s are prepared from the {@link PartitionQuery}s
* obtained by calling {@link DataStore#getPartitions(Query)}.
+ * <p>
+ * Hadoop jobs can be either configured through static
+ * <code>setInput()</code> methods, or from {@link GoraMapper}.
+ *
+ * @see GoraMapper
*/
public class GoraInputFormat<K, T extends Persistent>
extends InputFormat<K, T> implements Configurable {
@@ -150,4 +156,27 @@ public class GoraInputFormat<K, T extend
job.setInputFormatClass(GoraInputFormat.class);
GoraInputFormat.setQuery(job, query);
}
-}
+
+ /**
+ * Sets the input parameters for the job
+ * @param job the job to set the properties for
+ * @param query the query to get the inputs from
+ * @param dataStoreClass the datastore class
+ * @param inKeyClass Map input key class
+ * @param inValueClass Map input value class
+ * @param reuseObjects whether to reuse objects in serialization
+ * @throws IOException
+ */
+ public static <K1, V1 extends Persistent> void setInput(
+ Job job,
+ Class<? extends DataStore<K1,V1>> dataStoreClass,
+ Class<K1> inKeyClass,
+ Class<V1> inValueClass,
+ boolean reuseObjects)
+ throws IOException {
+
+ DataStore<K1,V1> store = DataStoreFactory.getDataStore(dataStoreClass
+ , inKeyClass, inValueClass);
+ setInput(job, store.newQuery(), store, reuseObjects);
+ }
+}
\ No newline at end of file
Modified: incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraMapper.java
URL: http://svn.apache.org/viewvc/incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraMapper.java?rev=1032463&r1=1032462&r2=1032463&view=diff
==============================================================================
--- incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraMapper.java (original)
+++ incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraMapper.java Mon Nov 8 05:27:47 2010
@@ -27,18 +27,106 @@ import org.apache.hadoop.mapreduce.Mappe
import org.apache.hadoop.mapreduce.Partitioner;
/**
- * Optional base class for Gora based {@link Mapper}s.
+ * Base class for Gora based {@link Mapper}s.
*/
public class GoraMapper<K1, V1 extends Persistent, K2, V2>
-extends Mapper<K1, V1, K2, V2> {
+ extends Mapper<K1, V1, K2, V2> {
+ /**
+ * Initializes the Mapper, and sets input parameters for the job. All of
+ * the records in the dataStore are used as the input. If you want to
+ * include a specific subset, use one of the overloaded methods which takes
+ * query parameter.
+ * @param job the job to set the properties for
+ * @param query the query to get the inputs from
+ * @param dataStoreClass the datastore class
+ * @param inKeyClass Map input key class
+ * @param inValueClass Map input value class
+ * @param outKeyClass Map output key class
+ * @param outValueClass Map output value class
+ * @param mapperClass the mapper class extending GoraMapper
+ * @param partitionerClass optional partitioner class
+ * @param reuseObjects whether to reuse objects in serialization
+ */
@SuppressWarnings("rawtypes")
public static <K1, V1 extends Persistent, K2, V2>
- void initMapperJob(Job job, Query<K1,V1> query,
+ void initMapperJob(
+ Job job,
+ Class<? extends DataStore<K1,V1>> dataStoreClass,
+ Class<K1> inKeyClass,
+ Class<V1> inValueClass,
+ Class<K2> outKeyClass,
+ Class<V2> outValueClass,
+ Class<? extends GoraMapper> mapperClass,
+ Class<? extends Partitioner> partitionerClass,
+ boolean reuseObjects)
+ throws IOException {
+
+ //set the input via GoraInputFormat
+ GoraInputFormat.setInput(job, dataStoreClass, inKeyClass, inValueClass, reuseObjects);
+
+ job.setMapperClass(mapperClass);
+ job.setMapOutputKeyClass(outKeyClass);
+ job.setMapOutputValueClass(outValueClass);
+
+ if (partitionerClass != null) {
+ job.setPartitionerClass(partitionerClass);
+ }
+ }
+
+ /**
+ * Initializes the Mapper, and sets input parameters for the job. All of
+ * the records in the dataStore are used as the input. If you want to
+ * include a specific subset, use one of the overloaded methods which takes
+ * query parameter.
+ * @param job the job to set the properties for
+ * @param query the query to get the inputs from
+ * @param dataStoreClass the datastore class
+ * @param inKeyClass Map input key class
+ * @param inValueClass Map input value class
+ * @param outKeyClass Map output key class
+ * @param outValueClass Map output value class
+ * @param mapperClass the mapper class extending GoraMapper
+ * @param reuseObjects whether to reuse objects in serialization
+ */
+ @SuppressWarnings("rawtypes")
+ public static <K1, V1 extends Persistent, K2, V2>
+ void initMapperJob(
+ Job job,
+ Class<? extends DataStore<K1,V1>> dataStoreClass,
+ Class<K1> inKeyClass,
+ Class<V1> inValueClass,
+ Class<K2> outKeyClass,
+ Class<V2> outValueClass,
+ Class<? extends GoraMapper> mapperClass,
+ boolean reuseObjects)
+ throws IOException {
+ initMapperJob(job, dataStoreClass, inKeyClass, inValueClass, outKeyClass
+ , outValueClass, mapperClass, null, reuseObjects);
+ }
+
+ /**
+ * Initializes the Mapper, and sets input parameters for the job
+ * @param job the job to set the properties for
+ * @param query the query to get the inputs from
+ * @param dataStore the datastore as the input
+ * @param outKeyClass Map output key class
+ * @param outValueClass Map output value class
+ * @param mapperClass the mapper class extending GoraMapper
+ * @param partitionerClass optional partitioner class
+ * @param reuseObjects whether to reuse objects in serialization
+ */
+ @SuppressWarnings("rawtypes")
+ public static <K1, V1 extends Persistent, K2, V2>
+ void initMapperJob(
+ Job job,
+ Query<K1,V1> query,
DataStore<K1,V1> dataStore,
- Class<K2> outKeyClass, Class<V2> outValueClass,
+ Class<K2> outKeyClass,
+ Class<V2> outValueClass,
Class<? extends GoraMapper> mapperClass,
- Class<? extends Partitioner> partitionerClass, boolean reuseObjects)
+ Class<? extends Partitioner> partitionerClass,
+ boolean reuseObjects)
throws IOException {
//set the input via GoraInputFormat
GoraInputFormat.setInput(job, query, dataStore, reuseObjects);
@@ -52,11 +140,49 @@ extends Mapper<K1, V1, K2, V2> {
}
}
+ /**
+ * Initializes the Mapper, and sets input parameters for the job
+ * @param job the job to set the properties for
+ * @param dataStore the datastore as the input
+ * @param outKeyClass Map output key class
+ * @param outValueClass Map output value class
+ * @param mapperClass the mapper class extending GoraMapper
+ * @param reuseObjects whether to reuse objects in serialization
+ */
@SuppressWarnings({ "rawtypes" })
public static <K1, V1 extends Persistent, K2, V2>
- void initMapperJob(Job job, Query<K1,V1> query, DataStore<K1,V1> dataStore,
- Class<K2> outKeyClass, Class<V2> outValueClass,
- Class<? extends GoraMapper> mapperClass, boolean reuseObjects)
+ void initMapperJob(
+ Job job,
+ DataStore<K1,V1> dataStore,
+ Class<K2> outKeyClass,
+ Class<V2> outValueClass,
+ Class<? extends GoraMapper> mapperClass,
+ boolean reuseObjects)
+ throws IOException {
+ initMapperJob(job, dataStore.newQuery(), dataStore,
+ outKeyClass, outValueClass, mapperClass, reuseObjects);
+ }
+
+ /**
+ * Initializes the Mapper, and sets input parameters for the job
+ * @param job the job to set the properties for
+ * @param query the query to get the inputs from
+ * @param dataStore the datastore as the input
+ * @param outKeyClass Map output key class
+ * @param outValueClass Map output value class
+ * @param mapperClass the mapper class extending GoraMapper
+ * @param reuseObjects whether to reuse objects in serialization
+ */
+ @SuppressWarnings({ "rawtypes" })
+ public static <K1, V1 extends Persistent, K2, V2>
+ void initMapperJob(
+ Job job,
+ Query<K1,V1> query,
+ DataStore<K1,V1> dataStore,
+ Class<K2> outKeyClass,
+ Class<V2> outValueClass,
+ Class<? extends GoraMapper> mapperClass,
+ boolean reuseObjects)
throws IOException {
initMapperJob(job, query, dataStore, outKeyClass, outValueClass,
Modified: incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraOutputFormat.java
URL: http://svn.apache.org/viewvc/incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraOutputFormat.java?rev=1032463&r1=1032462&r2=1032463&view=diff
==============================================================================
--- incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraOutputFormat.java (original)
+++ incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraOutputFormat.java Mon Nov 8 05:27:47 2010
@@ -33,6 +33,14 @@ import org.apache.hadoop.mapreduce.Recor
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+/**
+ * {@link OutputFormat} for Hadoop jobs that want to store the job outputs
+ * to a Gora store.
+ * <p>
+ * Hadoop jobs can be either configured through static
+ * <code>setOutput()</code> methods, or if the job is not map-only from {@link GoraReducer}.
+ * @see GoraReducer
+ */
public class GoraOutputFormat<K, T extends Persistent>
extends OutputFormat<K, T> {
@@ -105,14 +113,22 @@ public class GoraOutputFormat<K, T exten
*/
public static <K, V extends Persistent> void setOutput(Job job,
DataStore<K,V> dataStore, boolean reuseObjects) {
- setOutput(job, dataStore.getKeyClass(), dataStore.getPersistentClass(),
- dataStore.getClass(), reuseObjects);
+ setOutput(job, dataStore.getClass(), dataStore.getKeyClass()
+ , dataStore.getPersistentClass(), reuseObjects);
}
+ /**
+ * Sets the output parameters for the job
+ * @param job the job to set the properties for
+ * @param dataStoreClass the datastore class
+ * @param keyClass output key class
+ * @param persistentClass output value class
+ * @param reuseObjects whether to reuse objects in serialization
+ */
@SuppressWarnings("rawtypes")
public static <K, V extends Persistent> void setOutput(Job job,
- Class<K> keyClass, Class<V> persistentClass,
Class<? extends DataStore> dataStoreClass,
+ Class<K> keyClass, Class<V> persistentClass,
boolean reuseObjects) {
Configuration conf = job.getConfiguration();
@@ -120,6 +136,8 @@ public class GoraOutputFormat<K, T exten
GoraMapReduceUtils.setIOSerializations(conf, reuseObjects);
job.setOutputFormatClass(GoraOutputFormat.class);
+ job.setOutputKeyClass(keyClass);
+ job.setOutputValueClass(persistentClass);
conf.setClass(GoraOutputFormat.DATA_STORE_CLASS, dataStoreClass,
DataStore.class);
conf.setClass(GoraOutputFormat.OUTPUT_KEY_CLASS, keyClass, Object.class);
Modified: incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraReducer.java
URL: http://svn.apache.org/viewvc/incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraReducer.java?rev=1032463&r1=1032462&r2=1032463&view=diff
==============================================================================
--- incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraReducer.java (original)
+++ incubator/gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraReducer.java Mon Nov 8 05:27:47 2010
@@ -23,24 +23,64 @@ import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
/**
- * Optional base class for Gora based {@link Reducer}s.
+ * Base class for Gora based {@link Reducer}s.
*/
public class GoraReducer<K1, V1, K2, V2 extends Persistent>
-extends Reducer<K1, V1, K2, V2> {
-
+ extends Reducer<K1, V1, K2, V2> {
+
+ /**
+ * Initializes the Reducer, and sets output parameters for the job.
+ * @param job the job to set the properties for
+ * @param dataStoreClass the datastore class
+ * @param keyClass output key class
+ * @param persistentClass output value class
+ * @param reducerClass the reducer class extending GoraReducer
+ * @param reuseObjects whether to reuse objects in serialization
+ */
+ public static <K1, V1, K2, V2 extends Persistent>
+ void initReducerJob(
+ Job job,
+ Class<? extends DataStore<K2,V2>> dataStoreClass,
+ Class<K2> keyClass,
+ Class<V2> persistentClass,
+ Class<? extends GoraReducer<K1, V1, K2, V2>> reducerClass,
+ boolean reuseObjects) {
+
+ GoraOutputFormat.setOutput(job, dataStoreClass, keyClass, persistentClass, reuseObjects);
+
+ job.setReducerClass(reducerClass);
+ }
+
+ /**
+ * Initializes the Reducer, and sets output parameters for the job.
+ * @param job the job to set the properties for
+ * @param dataStore the datastore as the output
+ * @param reducerClass the reducer class extending GoraReducer
+ */
public static <K1, V1, K2, V2 extends Persistent>
- void initReducerJob(Job job, DataStore<K2,V2> dataStore,
+ void initReducerJob(
+ Job job,
+ DataStore<K2,V2> dataStore,
Class<? extends GoraReducer<K1, V1, K2, V2>> reducerClass) {
+
initReducerJob(job, dataStore, reducerClass, true);
}
+ /**
+ * Initializes the Reducer, and sets output parameters for the job.
+ * @param job the job to set the properties for
+ * @param dataStore the datastore as the output
+ * @param reducerClass the reducer class extending GoraReducer
+ * @param reuseObjects whether to reuse objects in serialization
+ */
public static <K1, V1, K2, V2 extends Persistent>
- void initReducerJob(Job job, DataStore<K2,V2> dataStore,
- Class<? extends GoraReducer<K1, V1, K2, V2>> reducerClass,
- boolean reuseObjects) {
-
+ void initReducerJob(
+ Job job,
+ DataStore<K2,V2> dataStore,
+ Class<? extends GoraReducer<K1, V1, K2, V2>> reducerClass,
+ boolean reuseObjects) {
+
GoraOutputFormat.setOutput(job, dataStore, reuseObjects);
-
job.setReducerClass(reducerClass);
}
}