diff --git a/CHANGES.txt b/CHANGES.txt
index 13f2495c9d..8081dc472d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -47,6 +47,8 @@ dev
  * change multiget key collection from list to set (CASSANDRA-1329)
  * ability to modify keyspaces and column family definitions on a live cluster
    (CASSANDRA-1285)
+ * support for Hadoop Streaming [non-jvm map/reduce via stdin/out]
+   (CASSANDRA-1368)
 
 
 0.7-beta1
diff --git a/NEWS.txt b/NEWS.txt
index ac5bf56a2c..2858470898 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -15,7 +15,8 @@ Features
     - Optional per-Column time-to-live field allows expiring data without
       have to issue explicit remove commands
     - `truncate` thrift method allows clearing an entire ColumnFamily at once
-    - Hadoop OutputFormat support
+    - Hadoop OutputFormat and Streaming [non-jvm map/reduce via stdin/out]
+      support
     - Up to 8x faster reads from row cache
     - A new ByteOrderedPartitioner supports bytes keys with arbitrary content,
       and orders keys by their byte value.  This should be used in new
diff --git a/interface/cassandra.genavro b/interface/cassandra.genavro
index a5b174d464..96eda25763 100644
--- a/interface/cassandra.genavro
+++ b/interface/cassandra.genavro
@@ -169,6 +169,11 @@ protocol Cassandra {
         array<CfDef> cf_defs;
     }
     
+    record StreamingMutation {
+        bytes key;
+        Mutation mutation;
+    }
+
     record MutationsMapEntry {
         bytes key;
         map<array<Mutation>> mutations;
diff --git a/ivy.xml b/ivy.xml
index dfb83e3781..1de56d3066 100644
--- a/ivy.xml
+++ b/ivy.xml
@@ -28,6 +28,7 @@
     <dependency org="commons-logging" name="commons-logging" rev="1.1.1"/>
     <dependency org="org.apache.rat" name="apache-rat" rev="0.6" />
     <dependency org="com.cloudera.hadoop" name="hadoop-core" rev="0.20.2-320"/>
+    <dependency org="com.cloudera.hadoop" name="hadoop-streaming" rev="0.20.2-320"/>
     <dependency org="net.sf.jopt-simple" name="jopt-simple" rev="3.2"/>
 
     <dependency org="net.java.dev.jna" name="jna" rev="3.2.7"/>
diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
index 98fbf70688..4403ed5886 100644
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
+++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
@@ -69,6 +69,7 @@ import org.slf4j.LoggerFactory;
  * </p>
  */
 public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutation>>
+    implements org.apache.hadoop.mapred.OutputFormat<ByteBuffer,List<Mutation>>
 {
     private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyOutputFormat.class);
     
@@ -85,13 +86,17 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
     @Override
     public void checkOutputSpecs(JobContext context)
     {
-        Configuration conf = context.getConfiguration();
+        checkOutputSpecs(context.getConfiguration());
+    }
+
+    private void checkOutputSpecs(Configuration conf)
+    {
         if (ConfigHelper.getOutputKeyspace(conf) == null || ConfigHelper.getOutputColumnFamily(conf) == null)
         {
             throw new UnsupportedOperationException("you must set the keyspace and columnfamily with setColumnFamily()");
         }
     }
-    
+
     /**
      * The OutputCommitter for this format does not write any data to the DFS.
      * 
@@ -107,6 +112,20 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
         return new NullOutputCommitter();
     }
     
+    /** Fills the deprecated OutputFormat interface for streaming. */
+    @Deprecated @Override
+    public void checkOutputSpecs(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job) throws IOException
+    {
+        checkOutputSpecs(job);
+    }
+
+    /** Fills the deprecated OutputFormat interface for streaming. */
+    @Deprecated @Override
+    public ColumnFamilyRecordWriter getRecordWriter(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job, String name, org.apache.hadoop.util.Progressable progress) throws IOException
+    {
+        return new ColumnFamilyRecordWriter(job);
+    }
+
     /**
      * Get the {@link RecordWriter} for the given task.
      * 
@@ -116,7 +135,7 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
      * @throws IOException
      */
     @Override
-    public RecordWriter<ByteBuffer,List<Mutation>> getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
+    public ColumnFamilyRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
     {
         return new ColumnFamilyRecordWriter(context);
     }
@@ -126,30 +145,29 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
      * keyspace, and is logged in with the configured credentials.
      *
      * @param socket  a socket pointing to a particular node, seed or otherwise
-     * @param context a job context
+     * @param conf a job configuration
      * @return a cassandra client
      * @throws InvalidRequestException
      * @throws TException
      * @throws AuthenticationException
      * @throws AuthorizationException
      */
-    public static Cassandra.Client createAuthenticatedClient(TSocket socket, JobContext context)
+    public static Cassandra.Client createAuthenticatedClient(TSocket socket, Configuration conf)
     throws InvalidRequestException, TException, AuthenticationException, AuthorizationException
     {
         TBinaryProtocol binaryProtocol = new TBinaryProtocol(new TFramedTransport(socket));
         Cassandra.Client client = new Cassandra.Client(binaryProtocol);
         socket.open();
-        client.set_keyspace(ConfigHelper.getOutputKeyspace(context.getConfiguration()));
-        if (ConfigHelper.getOutputKeyspaceUserName(context.getConfiguration()) != null)
+        client.set_keyspace(ConfigHelper.getOutputKeyspace(conf));
+        if (ConfigHelper.getOutputKeyspaceUserName(conf) != null)
         {
             Map<String, String> creds = new HashMap<String, String>();
-            creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(context.getConfiguration()));
-            creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(context.getConfiguration()));
+            creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(conf));
+            creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(conf));
             AuthenticationRequest authRequest = new AuthenticationRequest(creds);
             client.login(authRequest);
         }
         return client;
-
     }
 
     /**
diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
index e3bfce6f37..0a9bcbc3fb 100644
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
@@ -47,6 +47,8 @@ import org.apache.cassandra.thrift.SlicePredicate;
 import org.apache.cassandra.thrift.SliceRange;
 import org.apache.cassandra.thrift.SuperColumn;
 import org.apache.cassandra.utils.FBUtilities;
+
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.OutputFormat;
 import org.apache.hadoop.mapreduce.RecordWriter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
@@ -75,10 +77,10 @@ import org.apache.thrift.transport.TSocket;
  * @see OutputFormat
  * 
  */
-final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.apache.cassandra.avro.Mutation>>
+final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.apache.cassandra.avro.Mutation>> implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<org.apache.cassandra.avro.Mutation>>
 {
-    // The task attempt context this writer is associated with.
-    private final TaskAttemptContext context;
+    // The configuration this writer is associated with.
+    private final Configuration conf;
     
     // The batched set of mutations grouped by endpoints.
     private Map<InetAddress,Map<byte[],Map<String,List<Mutation>>>> mutationsByEndpoint;
@@ -104,15 +106,20 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
      */
     ColumnFamilyRecordWriter(TaskAttemptContext context) throws IOException
     {
-        this.context = context;
-        this.mutationsByEndpoint = new HashMap<InetAddress,Map<byte[],Map<String,List<Mutation>>>>();
-        this.ringCache = new RingCache(ConfigHelper.getOutputKeyspace(context.getConfiguration()),
-                                       ConfigHelper.getPartitioner(context.getConfiguration()),
-                                       ConfigHelper.getInitialAddress(context.getConfiguration()),
-                                       ConfigHelper.getRpcPort(context.getConfiguration()));
-        this.batchThreshold = context.getConfiguration().getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, Long.MAX_VALUE);
+        this(context.getConfiguration());
     }
     
+    ColumnFamilyRecordWriter(Configuration conf) throws IOException
+    {
+        this.conf = conf;
+        this.mutationsByEndpoint = new HashMap<InetAddress,Map<byte[],Map<String,List<Mutation>>>>();
+        this.ringCache = new RingCache(ConfigHelper.getOutputKeyspace(conf),
+                                       ConfigHelper.getPartitioner(conf),
+                                       ConfigHelper.getInitialAddress(conf),
+                                       ConfigHelper.getRpcPort(conf));
+        this.batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, Long.MAX_VALUE);
+    }
+
     /**
      * Return the endpoint responsible for the given key. The selected endpoint
      * one whose token range contains the given key.
@@ -145,7 +152,7 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
      * @throws IOException
      */
     @Override
-    public synchronized void write(ByteBuffer keybuff, List<org.apache.cassandra.avro.Mutation> value) throws IOException, InterruptedException
+    public synchronized void write(ByteBuffer keybuff, List<org.apache.cassandra.avro.Mutation> value) throws IOException
     {
         maybeFlush();
         byte[] key = copy(keybuff);
@@ -164,11 +171,11 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
             mutationsByKey.put(key, cfMutation);
         }
 
-        List<Mutation> mutationList = cfMutation.get(ConfigHelper.getOutputColumnFamily(context.getConfiguration()));
+        List<Mutation> mutationList = cfMutation.get(ConfigHelper.getOutputColumnFamily(conf));
         if (mutationList == null)
         {
             mutationList = new ArrayList<Mutation>();
-            cfMutation.put(ConfigHelper.getOutputColumnFamily(context.getConfiguration()), mutationList);
+            cfMutation.put(ConfigHelper.getOutputColumnFamily(conf), mutationList);
         }
 
         for (org.apache.cassandra.avro.Mutation amut : value)
@@ -254,6 +261,13 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
         flush();
     }
 
+    /** Fills the deprecated RecordWriter interface for streaming. */
+    @Deprecated @Override
+    public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException
+    {
+        flush();
+    }
+
     /**
      * Flush the mutations cache, iff more mutations have been cached than
      * {@link #batchThreshold}.
@@ -284,7 +298,7 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
             List<Future<?>> mutationFutures = new ArrayList<Future<?>>();
             for (Map.Entry<InetAddress, Map<byte[], Map<String, List<Mutation>>>> entry : mutationsByEndpoint.entrySet())
             {
-                mutationFutures.add(executor.submit(new EndpointCallable(context, entry.getKey(), entry.getValue())));
+                mutationFutures.add(executor.submit(new EndpointCallable(conf, entry.getKey(), entry.getValue())));
             }
             // wait until we have all the results back
             for (Future<?> mutationFuture : mutationFutures)
@@ -321,7 +335,7 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
     public class EndpointCallable implements Callable<Void>
     {
         // The task attempt context associated with this callable.
-        private TaskAttemptContext taskContext;
+        private Configuration conf;
         // The endpoint of the primary replica for the rows being mutated
         private InetAddress endpoint;
         // The mutations to be performed in the node referenced by {@link
@@ -332,13 +346,14 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
          * Constructs an {@link EndpointCallable} for the given endpoint and set
          * of mutations.
          *
+         * @param conf      job configuration
          * @param endpoint  the endpoint wherein to execute the mutations
          * @param mutations the mutation map expected by
          *                  {@link Cassandra.Client#batch_mutate(Map, ConsistencyLevel)}
          */
-        public EndpointCallable(TaskAttemptContext taskContext, InetAddress endpoint, Map<byte[], Map<String, List<Mutation>>> mutations)
+        public EndpointCallable(Configuration conf, InetAddress endpoint, Map<byte[], Map<String, List<Mutation>>> mutations)
         {
-            this.taskContext = taskContext;
+            this.conf = conf;
             this.endpoint = endpoint;
             this.mutations = mutations;
         }
@@ -352,8 +367,8 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
             TSocket socket = null;
             try
             {
-                socket = new TSocket(endpoint.getHostName(), ConfigHelper.getRpcPort(taskContext.getConfiguration()));
-                Cassandra.Client client = ColumnFamilyOutputFormat.createAuthenticatedClient(socket, taskContext);
+                socket = new TSocket(endpoint.getHostName(), ConfigHelper.getRpcPort(conf));
+                Cassandra.Client client = ColumnFamilyOutputFormat.createAuthenticatedClient(socket, conf);
                 client.batch_mutate(mutations, ConsistencyLevel.ONE);
                 return null;
             }