diff --git a/java/src/com/twitter/pycascading/Util.java b/java/src/com/twitter/pycascading/Util.java
index d4660b6..1e036dc 100755
--- a/java/src/com/twitter/pycascading/Util.java
+++ b/java/src/com/twitter/pycascading/Util.java
@@ -147,6 +147,13 @@ public static void run(String name, int numReducers, Long minSplitSize,  Map<Str
     properties.setProperty("pycascading.running_mode", runningMode);
     properties.setProperty("pycascading.main_file", (String) config.get("pycascading.main_file"));
 
+    key_for_config_options = "pycascading.hadoop.mapred.options"
+    if key_for_config_options in config:
+        # Apply user defined configurations, for example, use gzip to compress output
+        config_options = config.get(key_for_config_options)
+        for (option_name, option_value) in config_options.iteritems():
+            properties.setProperty(option_name, option_value);
+
     Configuration conf = new Configuration();
     TemporaryHdfs tempDir = null;
     if ("hadoop".equals(runningMode)) {
diff --git a/python/pycascading/tap.py b/python/pycascading/tap.py
index 268d098..7708a4c 100755
--- a/python/pycascading/tap.py
+++ b/python/pycascading/tap.py
@@ -234,6 +234,17 @@ def run(self, name="pycascading flow", num_reducers=50, min_split_size=0, config
         source_map = self.__get_active_sources()
         tails = [t.get_assembly() for t in self.tails]
         import pycascading.pipe
+        key_for_config_options = "pycascading.hadoop.mapred.options"
+        if config and key_for_config_options in config:
+            # To use gzip compression, following key/value pair should be included in given config dict
+            #"pycascading.hadoop.mapred.options": {
+            #    "mapred.output.compress": "true",
+            #    "mapred.compress.map.output": "true",
+            #    "mapred.output.compression.type": "BLOCK",
+            #    "mapred.map.output.compression.codec": "org.apache.hadoop.io.compress.GzipCodec",
+            #    "mapred.output.compression.codec": "org.apache.hadoop.io.compress.GzipCodec",
+            #}
+            pycascading.pipe.config[key_for_config_options] = config[key_for_config_options]
         Util.run(name, num_reducers, min_split_size, pycascading.pipe.config, source_map, \
                  self.sink_map, tails)