ColumnFamilyOutputFormat acts as a Hadoop-specific
+ * OutputFormat that allows reduce tasks to store keys (and corresponding
+ * values) as Cassandra rows (and respective columns) in a given
+ * {@link ColumnFamily}.
+ *
+ * + * As is the case with the {@link ColumnFamilyInputFormat}, you need to set the + * CF and predicate (description of columns to extract from each row) in your + * Hadoop job Configuration. The {@link ConfigHelper} class, through its + * {@link ConfigHelper#setColumnFamily} and + * {@link ConfigHelper#setSlicePredicate} methods, is provided to make this + * simple. + *
+ * + *+ * By default, it prevents overwriting existing rows in the column family, by + * ensuring at initialization time that it contains no rows in the given slice + * predicate. For the sake of performance, it employs a lazy write-back caching + * mechanism, where its record writer batches mutations created based on the + * reduce's inputs (in a task-specific map). When the writer is closed, then it + * makes the changes official by sending a batch mutate request to Cassandra. + *
+ * + * @author Karthick Sankarachary + */ +public class ColumnFamilyOutputFormat extends OutputFormat+ * This is to validate the output specification for the job when it is a job + * is submitted. By default, it will prevent writes to the given column + * family, if it already contains one or more rows in the given slice + * predicate. If you wish to relax that restriction, you may override this + * method is a sub-class of your choosing. + *
+ * + * @param context + * information about the job + * @throws IOException + * when output should not be attempted + */ + @Override + public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException + { + validateConfiguration(context.getConfiguration()); + String keyspace = ConfigHelper.getKeyspace(context.getConfiguration()); + String columnFamily = ConfigHelper.getColumnFamily(context.getConfiguration()); + SlicePredicate slicePredicate = ConfigHelper.getSlicePredicate(context.getConfiguration()); + assert slicePredicate != null; + if (slicePredicate.column_names == null && slicePredicate.slice_range == null) + slicePredicate = slicePredicate.setColumn_names(new ArrayList+ * This output format employs a lazy write-back caching mechanism, where the + * {@link RecordWriter} is responsible for collecting mutations in the + * {@link #MUTATIONS_CACHE}, and the {@link OutputCommitter} makes the + * changes official by making the change request to Cassandra. + *
+ * + * @param context + * the task context + * @return an output committer + * @throws IOException + * @throws InterruptedException + */ + @Override + public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException + { + return new NullOutputCommitter(); + } + + /** + * Get the {@link RecordWriter} for the given task. + * + *+ * As stated above, this {@link RecordWriter} merely batches the mutations + * that it defines in the {@link #MUTATIONS_CACHE}. In other words, it + * doesn't literally cause any changes on the Cassandra server. + *
+ * + * @param context + * the information about the current task. + * @return a {@link RecordWriter} to write the output for the job. + * @throws IOException + */ + @Override + public RecordWriterColumnFamilyOutputReducer reduces a <key, values>
+ * pair, where the value is a generic iterable type, into a list of columns that
+ * need to be mutated for that key, where each column corresponds to an element
+ * in the value.
+ *
+ *
+ * The default implementation treats the VALUEIN type to be a
+ * {@link ColumnWritable}, in which case this reducer acts as an identity
+ * function.
+ *
+ * @author Karthick Sankarachary
+ *
+ * @param
+ * Note that, given that round trips to the server are fairly expensive, it
+ * merely batches the mutations in-memory (specifically in
+ * {@link ColumnFamilyOutputFormat#MUTATIONS_CACHE}), and leaves it to the
+ * {@link ColumnFamilyOutputCommitter} to send the batched mutations to the
+ * server in one shot.
+ *
+ * Furthermore, this writer groups the mutations by the endpoint responsible for
+ * the rows being affected. This allows the {@link ColumnFamilyOutputCommitter}
+ * to execute the mutations in parallel, on a endpoint-by-endpoint basis.
+ *
+ * If the key is to be associated with a valid value, a mutation is created
+ * for it with the given column family and columns. In the event the value
+ * in the column is missing (i.e., null), then it is marked for
+ * {@link Deletion}. Similarly, if the entire value for a key is missing
+ * (i.e., null), then the entire key is marked for {@link Deletion}.
+ * ColumnFamilyRecordWriter maps the output <key, value>
+ * pairs to a Cassandra column family. In particular, it creates mutations for
+ * each column in the value, which it associates with the key, and in turn the
+ * responsible endpoint.
+ *
+ * RecordWriter to future operations, but not before
+ * flushing out the batched mutations.
+ *
+ * @param context the context of the task
+ * @throws IOException
+ */
+ @Override
+ public void close(TaskAttemptContext context) throws IOException, InterruptedException
+ {
+ flush();
+ }
+
+ /**
+ * Flush the mutations cache, iff more mutations have been cached than
+ * {@link #batchThreshold}.
+ *
+ * @throws IOException
+ */
+ private void maybeFlush() throws IOException
+ {
+ if (++batchSize > batchThreshold)
+ {
+ flush();
+ batchSize = 0L;
+ }
+ }
+
+ /**
+ * Send the batched mutations over to Cassandra, and then clear the
+ * mutations cache.
+ *
+ * @throws IOException
+ */
+ protected synchronized void flush() throws IOException
+ {
+ ExecutorService executor = Executors.newCachedThreadPool();
+
+ try
+ {
+ ListEndpointCallable facilitates an asynchronous call to a
+ * specific node in the ring that commands it to perform a batched set of
+ * mutations. Needless to say, the given mutations are targeted at rows that
+ * the selected endpoint is responsible for (i.e., is the primary replica
+ * for).
+ */
+ public class EndpointCallable implements CallableColumnWritable is a {@link WritableComparable} that denotes
+ * a column name and value.
+ */
+public class ColumnWritable implements WritableComparableo is a ColumnWritable with the same value. */
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof ColumnWritable))
+ return false;
+ ColumnWritable that = (ColumnWritable) o;
+ return compareTo(that) == 0;
+ }
+
+ public int hashCode()
+ {
+ return name.hashCode() + value.hashCode();
+ }
+
+ /** Compares two ColumnWritables. */
+ public int compareTo(ColumnWritable o)
+ {
+ ColumnWritable that = (ColumnWritable) o;
+ int nameComparison = BYTE_ARRAY_COMPARATOR.compare(this.name, that.name);
+ if (nameComparison != 0)
+ return nameComparison;
+ return BYTE_ARRAY_COMPARATOR.compare(this.value, that.value);
+ }
+
+ public String toString()
+ {
+ return "{ " + name.toString() + " : " + value.toString() + " }";
+ }
+}
diff --git a/test/unit/org/apache/cassandra/EmbeddedServer.java b/test/unit/org/apache/cassandra/EmbeddedServer.java
new file mode 100644
index 0000000000..f1484966c2
--- /dev/null
+++ b/test/unit/org/apache/cassandra/EmbeddedServer.java
@@ -0,0 +1,88 @@
+/*
+ * 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.cassandra;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.cassandra.service.CassandraDaemon;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+public class EmbeddedServer extends CleanupHelper
+{
+ protected static CassandraDaemon daemon = null;
+
+ enum GatewayService
+ {
+ Thrift, Avro
+ }
+
+ public static GatewayService getDaemonGatewayService()
+ {
+ return GatewayService.Thrift;
+ }
+
+ static ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ @BeforeClass
+ public static void startCassandra() throws IOException
+
+ {
+ executor.submit(new Runnable()
+ {
+ public void run()
+ {
+ switch (getDaemonGatewayService())
+ {
+ case Avro:
+ daemon = new org.apache.cassandra.avro.CassandraDaemon();
+ break;
+ case Thrift:
+ default:
+ daemon = new org.apache.cassandra.thrift.CassandraDaemon();
+ }
+ daemon.activate();
+ }
+ });
+ try
+ {
+ TimeUnit.SECONDS.sleep(3);
+ }
+ catch (InterruptedException e)
+ {
+ throw new AssertionError(e);
+ }
+ }
+
+ @AfterClass
+ public static void stopCassandra() throws Exception
+ {
+ if (daemon != null)
+ {
+ daemon.deactivate();
+ }
+ executor.shutdown();
+ executor.shutdownNow();
+ }
+
+}
diff --git a/test/unit/org/apache/cassandra/hadoop/ColumnFamilyOutputFormatTest.java b/test/unit/org/apache/cassandra/hadoop/ColumnFamilyOutputFormatTest.java
new file mode 100644
index 0000000000..c90e036d71
--- /dev/null
+++ b/test/unit/org/apache/cassandra/hadoop/ColumnFamilyOutputFormatTest.java
@@ -0,0 +1,222 @@
+package org.apache.cassandra.hadoop;
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import java.io.IOError;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.cassandra.EmbeddedServer;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.thrift.Cassandra;
+import org.apache.cassandra.thrift.CfDef;
+import org.apache.cassandra.thrift.ColumnOrSuperColumn;
+import org.apache.cassandra.thrift.ColumnParent;
+import org.apache.cassandra.thrift.ConsistencyLevel;
+import org.apache.cassandra.thrift.InvalidRequestException;
+import org.apache.cassandra.thrift.KeyRange;
+import org.apache.cassandra.thrift.KeySlice;
+import org.apache.cassandra.thrift.KsDef;
+import org.apache.cassandra.thrift.SlicePredicate;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * A test case for the {@link ColumnFamilyOutputFormat}, which reads each
+ * <key, value> pair from a sequence file, maps them to a <key,
+ * column> pair, and then reduces it by aggregating the columns that
+ * correspond to the same key. Finally, the output <key, columns> pairs
+ * are written into the (Cassandra) column family associated with this
+ * {@link OutputFormat}.
+ *
+ * @author Karthick Sankarachary
+ *
+ */
+
+public class ColumnFamilyOutputFormatTest extends EmbeddedServer
+{
+ static final String KEYSPACE = "ColumnFamilyOutputFormatTestKeyspace";
+ static final String COLUMN_FAMILY = "outputColumnFamily";
+
+ private static final String INPUT_FOLDER = "columnfamily.outputtest";
+
+ private static final String INPUT_FILE = "rows.txt";
+
+ private static final int NUMBER_OF_ROWS = 3;
+ private static final int NUMBER_OF_COLUMNS = 4;
+
+ ListSampleColumnFamilyOutputTool provides a tool interface which
+ * runs a {@link SampleColumnMapper} on the <key, value> pairs obtained
+ * from a sequence file, and then reduces it through the default
+ * {@link ColumnFamilyOutputReducer}.
+ *
+ * @author Karthick Sankarachary
+ *
+ */
+public class SampleColumnFamilyOutputTool extends Configured implements Tool
+{
+ private Path inputdir;
+
+ public SampleColumnFamilyOutputTool(Path inputdir, String columnFamily)
+ {
+ this.inputdir = inputdir;
+ }
+
+ public int run(String[] args)
+ throws InvalidRequestException, TException, IOException, InterruptedException, ClassNotFoundException
+ {
+ Job job = new Job(new Configuration());
+
+ // In case your job runs out of memory, use this setting
+ // (provided you're on Hadoop 0.20.1 or later)
+ // job.getConfiguration().setInt(JobContext.IO_SORT_MB, 1);
+ ConfigHelper.setColumnFamily(job.getConfiguration(),
+ ColumnFamilyOutputFormatTest.KEYSPACE,
+ ColumnFamilyOutputFormatTest.COLUMN_FAMILY);
+ ConfigHelper.setSlicePredicate(job.getConfiguration(), new SlicePredicate());
+
+ SequenceFileInputFormat.addInputPath(job, inputdir);
+
+ job.setMapperClass(SampleColumnMapper.class);
+ job.setMapOutputKeyClass(IntWritable.class);
+ job.setMapOutputValueClass(ColumnWritable.class);
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+
+ job.setReducerClass(ColumnFamilyOutputReducer.class);
+ job.setOutputKeyClass(byte[].class);
+ job.setOutputValueClass(SortedMap.class);
+ job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
+
+ job.waitForCompletion(true);
+ return 0;
+ }
+}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/hadoop/SampleColumnMapper.java b/test/unit/org/apache/cassandra/hadoop/SampleColumnMapper.java
new file mode 100644
index 0000000000..eac30b65a8
--- /dev/null
+++ b/test/unit/org/apache/cassandra/hadoop/SampleColumnMapper.java
@@ -0,0 +1,37 @@
+package org.apache.cassandra.hadoop;
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import java.io.IOException;
+
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.mapreduce.Mapper;
+
+/**
+ * A sample mapper that takes a pair of input <key, value> (writable)
+ * integers, and writes them as <key, column> writables.
+ */
+public class SampleColumnMapper extends Mapper