diff --git flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
index 2d1f7e8..2fc7f84 100644
--- flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
+++ flume-core/src/main/java/com/cloudera/flume/handlers/hdfs/CustomDfsSink.java
@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.compress.CompressionCodec;
@@ -103,6 +104,35 @@ public class CustomDfsSink extends EventSink.Base {
     FileSystem hdfs;
 
     String codecName = conf.getCollectorDfsCompressCodec();
+    CompressionCodec codec = getCodec(conf, codecName);
+
+    if (codec == null) {
+      dstPath = new Path(path);
+      hdfs = dstPath.getFileSystem(conf);
+      writer = hdfs.create(dstPath);
+      LOG.info("Creating HDFS file: " + dstPath.toString());
+      return;
+    }
+
+    Compressor cmp = codec.createCompressor();
+    dstPath = new Path(path + codec.getDefaultExtension());
+    hdfs = dstPath.getFileSystem(conf);
+    writer = hdfs.create(dstPath);
+    try {
+      writer = codec.createOutputStream(writer, cmp);
+    } catch (NullPointerException npe) {
+      // tries to find "native" version of codec, if that fails, then tries to
+      // find java version. If there is no java version, the createOutputStream
+      // exits via NPE. We capture this and convert it into a IOE with a more
+      // useful error message.
+      LOG.error("Unable to load compression codec " + codec);
+      throw new IOException("Unable to load compression codec " + codec);
+    }
+    LOG.info("Creating " + codec + " compressed HDFS file: "
+        + dstPath.toString());
+  }
+
+  public static CompressionCodec getCodec(Configuration conf, String codecName) {
     List<Class<? extends CompressionCodec>> codecs = CompressionCodecFactory
         .getCodecClasses(FlumeConfiguration.get());
     //Wish we could base this on DefaultCodec but appears not all codec's extend DefaultCodec(Lzo)
@@ -125,36 +155,15 @@ public class CustomDfsSink extends EventSink.Base {
 
     if (codec == null) {
       if (!codecName.equalsIgnoreCase("None")) {
-        LOG.warn("Unsupported compression codec " + codecName
-            + ".  Please choose from: " + codecStrs);
+        throw new IllegalArgumentException("Unsupported compression codec "
+            + codecName + ".  Please choose from: " + codecStrs);
       }
-      dstPath = new Path(path);
-      hdfs = dstPath.getFileSystem(conf);
-      writer = hdfs.create(dstPath);
-      LOG.info("Creating HDFS file: " + dstPath.toString());
-      return;
-    }
-    //Must check instanceof codec as BZip2Codec doesn't inherit Configurable
-    if(codec instanceof Configurable){
+    } else if(codec instanceof Configurable) {
+      //Must check instanceof codec as BZip2Codec doesn't inherit Configurable
       //Must set the configuration for Configurable objects that may or do use native libs
       ((Configurable)codec).setConf(conf);
     }
-    Compressor cmp = codec.createCompressor();
-    dstPath = new Path(path + codec.getDefaultExtension());
-    hdfs = dstPath.getFileSystem(conf);
-    writer = hdfs.create(dstPath);
-    try {
-      writer = codec.createOutputStream(writer, cmp);
-    } catch (NullPointerException npe) {
-      // tries to find "native" version of codec, if that fails, then tries to
-      // find java version. If there is no java version, the createOutputStream
-      // exits via NPE. We capture this and convert it into a IOE with a more
-      // useful error message.
-      LOG.error("Unable to load compression codec " + codec);
-      throw new IOException("Unable to load compression codec " + codec);
-    }
-    LOG.info("Creating " + codec + " compressed HDFS file: "
-        + dstPath.toString());
+    return codec;
   }
 
   public static SinkBuilder builder() {
diff --git flume-core/src/main/java/com/cloudera/flume/handlers/seqfile/SequenceFileOutputFormat.java flume-core/src/main/java/com/cloudera/flume/handlers/seqfile/SequenceFileOutputFormat.java
new file mode 100644
index 0000000..52f3d7a
--- /dev/null
+++ flume-core/src/main/java/com/cloudera/flume/handlers/seqfile/SequenceFileOutputFormat.java
@@ -0,0 +1,127 @@
+/**
+ * Licensed to Cloudera, Inc. under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Cloudera, Inc. 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 com.cloudera.flume.handlers.seqfile;
+
+import com.cloudera.flume.conf.FlumeConfiguration;
+import com.cloudera.flume.core.Event;
+import com.cloudera.flume.handlers.hdfs.CustomDfsSink;
+import com.cloudera.flume.handlers.hdfs.WriteableEvent;
+import com.cloudera.flume.handlers.hdfs.WriteableEventKey;
+import com.cloudera.flume.handlers.text.FormatFactory.OutputFormatBuilder;
+import com.cloudera.flume.handlers.text.output.AbstractOutputFormat;
+import com.cloudera.flume.handlers.text.output.OutputFormat;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
+import org.apache.hadoop.io.SequenceFile.Writer;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.DefaultCodec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link OutputFormat} for writing Flume events to Hadoop Sequence Files.
+ *
+ * This class is not thread safe.
+ */
+public class SequenceFileOutputFormat extends AbstractOutputFormat {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(SequenceFileOutputFormat.class);
+
+  private static final String NAME = "seqfile";
+
+  private CompressionType compressionType;
+  private CompressionCodec codec;
+
+  private OutputStream cachedOut;
+  private Writer writer;
+
+  public SequenceFileOutputFormat() {
+    this(SequenceFile.getCompressionType(FlumeConfiguration.get()),
+        new DefaultCodec());
+  }
+
+  public SequenceFileOutputFormat(CompressionType compressionType,
+      CompressionCodec codec) {
+    this.compressionType = compressionType;
+    this.codec = codec;
+  }
+
+  @Override
+  public void format(OutputStream o, Event e) throws IOException {
+    if (writer == null) {
+      cachedOut = o;
+      FSDataOutputStream fsOut;
+      if (o instanceof FSDataOutputStream) {
+        fsOut = (FSDataOutputStream) o;
+      } else {
+        fsOut = new FSDataOutputStream(o, null);
+      }
+      writer = SequenceFile.createWriter(FlumeConfiguration.get(), fsOut,
+              WriteableEventKey.class, WriteableEvent.class,
+              compressionType, codec);
+    }
+    if (cachedOut != o) {
+      // different output than last time, fail here
+      throw new IOException(
+          "OutputFormat instance can only write to the same OutputStream");
+    }
+    writer.append(new WriteableEventKey(e), new WriteableEvent(e));
+  }
+
+
+
+  public static OutputFormatBuilder builder() {
+    return new OutputFormatBuilder() {
+
+      @Override
+      public OutputFormat build(String... args) {
+        Preconditions.checkArgument(args.length <= 1, "usage: seqfile([codec])");
+
+        OutputFormat format;
+
+        String codecName = null;
+        if (args.length > 0) {
+          codecName = args[0];
+          FlumeConfiguration conf = FlumeConfiguration.get();
+          CompressionType compressionType = SequenceFile.getCompressionType(conf);
+          CompressionCodec codec = CustomDfsSink.getCodec(conf, codecName);
+          format = new SequenceFileOutputFormat(compressionType, codec);
+        } else {
+          format = new SequenceFileOutputFormat();
+        }
+
+        format.setBuilder(this);
+
+        return format;
+      }
+
+      @Override
+      public String getName() {
+        return NAME;
+      }
+
+    };
+  }
+}
diff --git flume-core/src/main/java/com/cloudera/flume/handlers/text/FormatFactory.java flume-core/src/main/java/com/cloudera/flume/handlers/text/FormatFactory.java
index f07c109..ad9ea4a 100644
--- flume-core/src/main/java/com/cloudera/flume/handlers/text/FormatFactory.java
+++ flume-core/src/main/java/com/cloudera/flume/handlers/text/FormatFactory.java
@@ -34,6 +34,7 @@ import com.cloudera.flume.conf.FlumeSpecException;
 import com.cloudera.flume.handlers.avro.AvroDataFileOutputFormat;
 import com.cloudera.flume.handlers.avro.AvroJsonOutputFormat;
 import com.cloudera.flume.handlers.avro.AvroNativeFileOutputFormat;
+import com.cloudera.flume.handlers.seqfile.SequenceFileOutputFormat;
 import com.cloudera.flume.handlers.text.output.DebugOutputFormat;
 import com.cloudera.flume.handlers.text.output.JsonOutputFormat;
 import com.cloudera.flume.handlers.text.output.Log4jOutputFormat;
@@ -93,6 +94,7 @@ public class FormatFactory {
       add(AvroJsonOutputFormat.builder());
       add(AvroDataFileOutputFormat.builder());
       add(AvroNativeFileOutputFormat.builder());
+      add(SequenceFileOutputFormat.builder());
     }
   };
 
diff --git flume-core/src/test/java/com/cloudera/flume/handlers/seqfile/TestSequenceFileOutputFormat.java flume-core/src/test/java/com/cloudera/flume/handlers/seqfile/TestSequenceFileOutputFormat.java
new file mode 100644
index 0000000..1659a17
--- /dev/null
+++ flume-core/src/test/java/com/cloudera/flume/handlers/seqfile/TestSequenceFileOutputFormat.java
@@ -0,0 +1,110 @@
+/**
+ * Licensed to Cloudera, Inc. under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Cloudera, Inc. 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 com.cloudera.flume.handlers.seqfile;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.cloudera.flume.conf.FlumeConfiguration;
+import com.cloudera.flume.conf.FlumeSpecException;
+import com.cloudera.flume.core.Event;
+import com.cloudera.flume.handlers.debug.MemorySinkSource;
+import com.cloudera.flume.handlers.hdfs.WriteableEvent;
+import com.cloudera.flume.handlers.hdfs.WriteableEventKey;
+import com.cloudera.flume.handlers.text.FormatFactory;
+import com.cloudera.flume.handlers.text.output.OutputFormat;
+import com.cloudera.util.FileUtil;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This test takes a small canned batch of events, writes them to a sequence
+ * file, and then reads them back checking the values.
+ */
+public class TestSequenceFileOutputFormat {
+  private static final Logger LOG = LoggerFactory
+      .getLogger(TestSequenceFileOutputFormat.class);
+
+  /**
+   * This is a helper method that is a lot like the above, except that it
+   * directly creates the output format so that we can configure it, since
+   * this isn't possible via the configuration language currently.
+   */
+  public void sequenceFileWriteReadHelper(String... args) throws IOException,
+      FlumeSpecException, InterruptedException {
+
+    MemorySinkSource mem = MemorySinkSource.cannedData("test ", 5);
+
+    // setup sink.
+    File f = FileUtil.createTempFile("sequencefile", ".seq");
+    f.deleteOnExit();
+    FileOutputStream fos = new FileOutputStream(f);
+    LOG.info("filename before escaping: " + f.getAbsolutePath());
+    OutputFormat out = FormatFactory.get().getOutputFormat("seqfile", args);
+    mem.open();
+    Event e = mem.next();
+    while (e != null) {
+      out.format(fos, e);
+      e = mem.next();
+    }
+
+    mem.open();
+
+    FlumeConfiguration conf = FlumeConfiguration.get();
+    FileSystem fs = FileSystem.getLocal(conf);
+    SequenceFile.Reader r = new SequenceFile.Reader(fs, new Path(f.toURI()), conf);
+    WriteableEventKey k = new WriteableEventKey();
+    WriteableEvent evt = new WriteableEvent();
+    while (r.next(k, evt)) {
+      Event expected = mem.next();
+      assertEquals(evt.getTimestamp(), expected.getTimestamp());
+      assertEquals(evt.getNanos(), expected.getNanos());
+      assertEquals(evt.getPriority(), expected.getPriority());
+      assertTrue(Arrays.equals(evt.getBody(), expected.getBody()));
+    }
+
+  }
+
+  @Test
+  public void testSequenceFileFormatDefault() throws IOException,
+      FlumeSpecException, InterruptedException {
+      sequenceFileWriteReadHelper();
+  }
+
+  @Test
+  public void testSequenceFileFormatDeflateCodec() throws IOException,
+      FlumeSpecException, InterruptedException {
+      sequenceFileWriteReadHelper("BZip2Codec");
+  }
+
+  @Test(expected=IllegalArgumentException.class)
+  public void testSequenceFileFormatInvalidCodec() throws IOException,
+      FlumeSpecException, InterruptedException {
+      sequenceFileWriteReadHelper("invalid");
+  }
+}
