natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-1.2' into trunk

This commit is contained in:
Brandon Williams 2013-05-29 12:54:38 -05:00
parent b3101a17ae 6d7404bc2e
commit d78b622646
3 changed files with 26 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -70,6 +70,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

4
examples/pig/README.txt
View File

@ -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.

21
src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
View File

@ -78,6 +78,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";
@ -106,6 +107,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,Column> lastRow;
@ -505,6 +507,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("@");
@ -580,6 +584,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)
@ -592,12 +599,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();