cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1001930 - in /cassandra/trunk: ./ contrib/word_count/ contrib/word_count/bin/ contrib/word_count/src/ src/java/org/apache/cassandra/hadoop/
Date Mon, 27 Sep 2010 22:11:06 GMT
Author: jbellis
Date: Mon Sep 27 22:11:05 2010
New Revision: 1001930

URL: http://svn.apache.org/viewvc?rev=1001930&view=rev
Log:
Added option for filesystem/cassandra output.
patch by Jeremy Hanna; reviewed by Stu Hood for CASSANDRA-1342


Modified:
    cassandra/trunk/CHANGES.txt
    cassandra/trunk/contrib/word_count/README.txt
    cassandra/trunk/contrib/word_count/bin/word_count   (contents, props changed)
    cassandra/trunk/contrib/word_count/bin/word_count_setup   (contents, props changed)
    cassandra/trunk/contrib/word_count/src/WordCount.java
    cassandra/trunk/contrib/word_count/src/WordCountSetup.java
    cassandra/trunk/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java

Modified: cassandra/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/CHANGES.txt?rev=1001930&r1=1001929&r2=1001930&view=diff
==============================================================================
--- cassandra/trunk/CHANGES.txt (original)
+++ cassandra/trunk/CHANGES.txt Mon Sep 27 22:11:05 2010
@@ -1,5 +1,7 @@
 dev
  * create EndpointSnitchInfo and MBean to expose rack and DC (CASSANDRA-1491)
+ * added option to contrib/word_count to output results back to Cassandra
+   (CASSANDRA-1342)
 
 
 0.7-beta2

Modified: cassandra/trunk/contrib/word_count/README.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/contrib/word_count/README.txt?rev=1001930&r1=1001929&r2=1001930&view=diff
==============================================================================
--- cassandra/trunk/contrib/word_count/README.txt (original)
+++ cassandra/trunk/contrib/word_count/README.txt Mon Sep 27 22:11:05 2010
@@ -10,14 +10,20 @@ contrib/word_count$ ant
 contrib/word_count$ bin/word_count_setup
 contrib/word_count$ bin/word_count
 
-Output will be in /tmp/word_count*.
+The output of the word count can now be configured. In the bin/word_count
+file, you can specify the OUTPUT_REDUCER. The two options are 'filesystem'
+and 'cassandra'. The filesystem option outputs to the /tmp/word_count*
+directories. The cassandra option outputs to the 'Standard2' column family.
+
+In order to view the results in Cassandra, one can use python/pycassa and
+perform the following operations:
+$ python
+>>> import pycassa
+>>> con = pycassa.connect('Keyspace1')
+>>> cf = pycassa.ColumnFamily(con, 'Standard2')
+>>> list(cf.get_range())
 
 Read the code in src/ for more details.
 
 *If you want to point wordcount at a real cluster, modify the seed
-and listenaddress settings in storage-conf.xml accordingly.
-
-*For Mac users, the storage-conf.xml uses 127.0.0.2 for the 
-word_count_setup. Mac OS X doesn't have that address available.
-To add it, run this before running bin/word_count_setup:
-sudo ifconfig lo0 alias 127.0.0.2 up
+and listenaddress settings accordingly.

Modified: cassandra/trunk/contrib/word_count/bin/word_count
URL: http://svn.apache.org/viewvc/cassandra/trunk/contrib/word_count/bin/word_count?rev=1001930&r1=1001929&r2=1001930&view=diff
==============================================================================
--- cassandra/trunk/contrib/word_count/bin/word_count (original)
+++ cassandra/trunk/contrib/word_count/bin/word_count Mon Sep 27 22:11:05 2010
@@ -53,5 +53,7 @@ if [ "x$JAVA" = "x" ]; then
     exit 1
 fi
 
+OUTPUT_REDUCER=filesystem
+
 #echo $CLASSPATH
-$JAVA -Xmx1G -ea -cp $CLASSPATH WordCount
+$JAVA -Xmx1G -ea -cp $CLASSPATH WordCount output_reducer=$OUTPUT_REDUCER

Propchange: cassandra/trunk/contrib/word_count/bin/word_count
------------------------------------------------------------------------------
    svn:executable = *

Modified: cassandra/trunk/contrib/word_count/bin/word_count_setup
URL: http://svn.apache.org/viewvc/cassandra/trunk/contrib/word_count/bin/word_count_setup?rev=1001930&r1=1001929&r2=1001930&view=diff
==============================================================================
    (empty)

Propchange: cassandra/trunk/contrib/word_count/bin/word_count_setup
------------------------------------------------------------------------------
    svn:executable = *

Modified: cassandra/trunk/contrib/word_count/src/WordCount.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/contrib/word_count/src/WordCount.java?rev=1001930&r1=1001929&r2=1001930&view=diff
==============================================================================
--- cassandra/trunk/contrib/word_count/src/WordCount.java (original)
+++ cassandra/trunk/contrib/word_count/src/WordCount.java Mon Sep 27 22:11:05 2010
@@ -17,10 +17,13 @@
  */
 
 import java.io.IOException;
-import java.util.Arrays;
-import java.util.SortedMap;
-import java.util.StringTokenizer;
+import java.nio.ByteBuffer;
+import java.util.*;
 
+import org.apache.cassandra.avro.Column;
+import org.apache.cassandra.avro.ColumnOrSuperColumn;
+import org.apache.cassandra.avro.Mutation;
+import org.apache.cassandra.hadoop.ColumnFamilyOutputFormat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -52,9 +55,13 @@ public class WordCount extends Configure
 
     static final String KEYSPACE = "Keyspace1";
     static final String COLUMN_FAMILY = "Standard1";
-    private static final String CONF_COLUMN_NAME = "columnname";
+
+    static final String OUTPUT_REDUCER_VAR = "output_reducer";
+    static final String OUTPUT_COLUMN_FAMILY = "Standard2";
     private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count";
 
+    private static final String CONF_COLUMN_NAME = "columnname";
+
     public static void main(String[] args) throws Exception
     {
         // Let ToolRunner handle generic command-line options
@@ -92,7 +99,7 @@ public class WordCount extends Configure
         
     }
 
-    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable>
+    public static class ReducerToFilesystem extends Reducer<Text, IntWritable, Text, IntWritable>
     {
         private IntWritable result = new IntWritable();
 
@@ -109,30 +116,109 @@ public class WordCount extends Configure
         }
     }
 
+    public static class ReducerToCassandra extends Reducer<Text, IntWritable, ByteBuffer,
List<Mutation>>
+    {
+        private List<Mutation> results = new ArrayList<Mutation>();
+        private String columnName;
+
+        public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException
+        {
+            int sum = 0;
+            for (IntWritable val : values)
+            {
+                sum += val.get();
+            }
+
+            results.add(getMutation(key, sum));
+            context.write(ByteBuffer.wrap(columnName.getBytes()), results);
+            results.clear();
+        }
+
+        protected void setup(org.apache.hadoop.mapreduce.Reducer.Context context)
+            throws IOException, InterruptedException
+        {
+            this.columnName = context.getConfiguration().get(CONF_COLUMN_NAME);
+        }
+
+        private static Mutation getMutation(Text key, int sum)
+        {
+            Mutation m = new Mutation();
+            m.column_or_supercolumn = getCoSC(key, sum);
+            return m;
+        }
+
+        private static ColumnOrSuperColumn getCoSC(Text key, int sum)
+        {
+            // Have to convert both the key and the sum to ByteBuffers
+            // for the generalized output format
+            ByteBuffer name = ByteBuffer.wrap(key.getBytes());
+            ByteBuffer value = ByteBuffer.wrap(String.valueOf(sum).getBytes());
+
+            Column c = new Column();
+            c.name = name;
+            c.value = value;
+            c.timestamp = System.currentTimeMillis() * 1000;
+            c.ttl = 0;
+            ColumnOrSuperColumn cosc = new ColumnOrSuperColumn();
+            cosc.column = c;
+            return cosc;
+        }
+    }
+
     public int run(String[] args) throws Exception
     {
+        String outputReducerType = "filesystem";
+        if (args != null && args[0].startsWith(OUTPUT_REDUCER_VAR))
+        {
+            String[] s = args[0].split("=");
+            if (s != null && s.length == 2)
+                outputReducerType = s[1];
+        }
+        logger.info("output reducer type: " + outputReducerType);
 
         for (int i = 0; i < WordCountSetup.TEST_COUNT; i++)
         {
             String columnName = "text" + i;
             getConf().set(CONF_COLUMN_NAME, columnName);
+
             Job job = new Job(getConf(), "wordcount");
             job.setJarByClass(WordCount.class);
             job.setMapperClass(TokenizerMapper.class);
-            job.setCombinerClass(IntSumReducer.class);
-            job.setReducerClass(IntSumReducer.class);
-            job.setOutputKeyClass(Text.class);
-            job.setOutputValueClass(IntWritable.class);
+
+            if (outputReducerType.equalsIgnoreCase("filesystem"))
+            {
+                job.setCombinerClass(ReducerToFilesystem.class);
+                job.setReducerClass(ReducerToFilesystem.class);
+                job.setOutputKeyClass(Text.class);
+                job.setOutputValueClass(IntWritable.class);
+                FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX + i));
+            }
+            else
+            {
+                job.setReducerClass(ReducerToCassandra.class);
+
+                job.setMapOutputKeyClass(Text.class);
+                job.setMapOutputValueClass(IntWritable.class);
+                job.setOutputKeyClass(ByteBuffer.class);
+                job.setOutputValueClass(List.class);
+
+                job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
+                
+                ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE, OUTPUT_COLUMN_FAMILY);
+            }
 
             job.setInputFormatClass(ColumnFamilyInputFormat.class);
-            FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX + i));
+
 
             ConfigHelper.setRpcPort(job.getConfiguration(), "9160");
             ConfigHelper.setInitialAddress(job.getConfiguration(), "localhost");
+            ConfigHelper.setPartitioner(job.getConfiguration(), "org.apache.cassandra.dht.RandomPartitioner");
             ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
             SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(columnName.getBytes()));
             ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
 
+
+
             job.waitForCompletion(true);
         }
         return 0;

Modified: cassandra/trunk/contrib/word_count/src/WordCountSetup.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/contrib/word_count/src/WordCountSetup.java?rev=1001930&r1=1001929&r2=1001930&view=diff
==============================================================================
--- cassandra/trunk/contrib/word_count/src/WordCountSetup.java (original)
+++ cassandra/trunk/contrib/word_count/src/WordCountSetup.java Mon Sep 27 22:11:05 2010
@@ -97,8 +97,8 @@ public class WordCountSetup
     private static void setupKeyspace(Cassandra.Iface client) throws TException, InvalidRequestException
     {
         List<CfDef> cfDefList = new ArrayList<CfDef>();
-        CfDef cfDef = new CfDef(WordCount.KEYSPACE, WordCount.COLUMN_FAMILY);
-        cfDefList.add(cfDef);
+        cfDefList.add(new CfDef(WordCount.KEYSPACE, WordCount.COLUMN_FAMILY));
+        cfDefList.add(new CfDef(WordCount.KEYSPACE, WordCount.OUTPUT_COLUMN_FAMILY));
 
         client.system_add_keyspace(new KsDef(WordCount.KEYSPACE, "org.apache.cassandra.locator.SimpleStrategy",
1, cfDefList));
     }

Modified: cassandra/trunk/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java?rev=1001930&r1=1001929&r2=1001930&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java Mon
Sep 27 22:11:05 2010
@@ -51,7 +51,7 @@ import org.slf4j.LoggerFactory;
  * The <code>ColumnFamilyOutputFormat</code> 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}.
+ * ColumnFamily.
  * 
  * <p>
  * As is the case with the {@link ColumnFamilyInputFormat}, you need to set the



Mime
View raw message