diff --git a/CHANGES.txt b/CHANGES.txt
index 34e5b52dd1..61bd4b73af 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -12,6 +12,7 @@
 Merged from 1.1:
  * Remove buggy thrift max message length option (CASSANDRA-5529)
  * Fix NPE in Pig's widerow mode (CASSANDRA-5488)
+ * Add split size parameter to Pig and disable split combination (CASSANDRA-5544)
 
 
 1.2.5
diff --git a/examples/pig/README.txt b/examples/pig/README.txt
index 3ef7858678..e3d9af6a82 100644
--- a/examples/pig/README.txt
+++ b/examples/pig/README.txt
@@ -88,3 +88,7 @@ PIG_USE_SECONDARY:  this allows easy use of secondary indexes within your
                     can also be set in the LOAD url by adding the
                     'use_secondary=true' parameter.
 
+PIG_INPUT_SPLIT_SIZE: this sets the split size passed to Hadoop, controlling
+                      the amount of mapper tasks created.  This can also be set in the LOAD url by
+                      adding the 'split_size=X' parameter, where X is an integer amount for the size.
+
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
index 0854758a55..6490d0502f 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
@@ -77,6 +77,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
     public final static String PIG_ALLOW_DELETES = "PIG_ALLOW_DELETES";
     public final static String PIG_WIDEROW_INPUT = "PIG_WIDEROW_INPUT";
     public final static String PIG_USE_SECONDARY = "PIG_USE_SECONDARY";
+    public final static String PIG_INPUT_SPLIT_SIZE = "PIG_INPUT_SPLIT_SIZE";
 
     private final static String DEFAULT_INPUT_FORMAT = "org.apache.cassandra.hadoop.ColumnFamilyInputFormat";
     private final static String DEFAULT_OUTPUT_FORMAT = "org.apache.cassandra.hadoop.ColumnFamilyOutputFormat";
@@ -105,6 +106,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
     private int limit;
     private boolean widerows = false;
     private boolean usePartitionFilter = false;
+    private int splitSize = 64 * 1024;
     // wide row hacks
     private ByteBuffer lastKey;
     private Map<ByteBuffer,IColumn> lastRow;
@@ -516,6 +518,8 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
                     widerows = Boolean.parseBoolean(urlQuery.get("widerows"));
                 if (urlQuery.containsKey("use_secondary"))
                     usePartitionFilter = Boolean.parseBoolean(urlQuery.get("use_secondary"));
+                if (urlQuery.containsKey("split_size"))
+                    splitSize = Integer.parseInt(urlQuery.get("split_size"));
             }
             String[] parts = urlParts[0].split("/+");
             String[] credentialsAndKeyspace = parts[1].split("@");
@@ -591,6 +595,9 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
     public void setLocation(String location, Job job) throws IOException
     {
         conf = job.getConfiguration();
+        
+        // don't combine mappers to a single mapper per node
+        conf.setBoolean("pig.noSplitCombination", true);
         setLocationFromUri(location);
 
         if (ConfigHelper.getInputSlicePredicate(conf) == null)
@@ -603,12 +610,26 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
             widerows = Boolean.valueOf(System.getenv(PIG_WIDEROW_INPUT));
         if (System.getenv(PIG_USE_SECONDARY) != null)
             usePartitionFilter = Boolean.valueOf(System.getenv(PIG_USE_SECONDARY));
+        if (System.getenv(PIG_INPUT_SPLIT_SIZE) != null)
+        {
+            try
+            {
+                ConfigHelper.setInputSplitSize(conf, Integer.valueOf(System.getenv(PIG_INPUT_SPLIT_SIZE)));
+            }
+            catch(NumberFormatException e)
+            {
+                throw new RuntimeException("PIG_INPUT_SPLIT_SIZE is not a number", e);
+            }           
+        }
 
         if (usePartitionFilter && getIndexExpressions() != null)
             ConfigHelper.setInputRange(conf, getIndexExpressions());
 
         if (username != null && password != null)
             ConfigHelper.setInputKeyspaceUserNameAndPassword(conf, username, password);
+        
+        if (splitSize > 0)
+            ConfigHelper.setInputSplitSize(conf, splitSize);
 
         ConfigHelper.setInputColumnFamily(conf, keyspace, column_family, widerows);
         setConnectionInformation();