Merge pull request #18 from derekgr/choose-output-format

Optionally choose a specific OutputFormat class
ifesdjeen · Jan 7, 2014 · 1ff5e33 · 1ff5e33
2 parents dec7b91 + 9ba462d
commit 1ff5e33
Showing 1 changed file with 14 additions and 1 deletion.
diff --git a/src/main/java/com/ifesdjeen/cascading/cassandra/CassandraScheme.java b/src/main/java/com/ifesdjeen/cascading/cassandra/CassandraScheme.java
@@ -158,7 +158,20 @@ public void sinkConfInit(FlowProcess<JobConf> process,
                            Tap<JobConf, RecordReader, OutputCollector> tap,
                            JobConf conf) {
     super.sinkConfInit(process, tap, conf);
-    conf.setOutputFormat(ColumnFamilyOutputFormat.class);
+
+    if (this.settings.containsKey("sink.outputFormat")) {
+      String outputFormatName = (String)this.settings.get("sink.outputFormat");
+      try {
+        Class outputFormat = (Class<? extends OutputFormat>)Class.forName(outputFormatName);
+        conf.setOutputFormat(outputFormat);
+      }
+      catch (ClassNotFoundException e) {
+        throw new IllegalArgumentException("Class not found: " + outputFormatName, e);
+      }
+    }
+    else {
+      conf.setOutputFormat(ColumnFamilyOutputFormat.class);
+    }
   }
 
   /**