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

update configuration checking post-#3197 

patch by jbellis; reviewed by brandonwilliams for CASSANDRA-3765
This commit is contained in:
Jonathan Ellis 2012-01-21 15:41:06 -06:00
parent ee5dbdb8d3
commit 27b818386c
5 changed files with 15 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
1.1-dev
* (Hadoop) separate input/output configurations (CASSANDRA-3197, 3765)
* avoid returning internal Cassandra classes over JMX (CASSANDRA-2805)
* add row-level isolation via SnapTree (CASSANDRA-2893)
* Optimize key count estimation when opening sstable on startup

5
examples/hadoop_word_count/src/WordCount.java
View File

@ -190,7 +190,7 @@ public class WordCount extends Configured implements Tool
ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setInputPartitioner(job.getConfiguration(), "org.apache.cassandra.dht.RandomPartitioner");
ConfigHelper.setInputPartitioner(job.getConfiguration(), "RandomPartitioner");
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(ByteBufferUtil.bytes(columnName)));
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
@ -200,6 +200,9 @@ public class WordCount extends Configured implements Tool
ConfigHelper.setInputRange(job.getConfiguration(), Arrays.asList(expr));
}
ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setOutputPartitioner(job.getConfiguration(), "RandomPartitioner");
job.waitForCompletion(true);
}
return 0;

4
src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java
View File

@ -100,6 +100,10 @@ public class ColumnFamilyInputFormat extends InputFormat<ByteBuffer, SortedMap<B
{
throw new UnsupportedOperationException("you must set the predicate with setPredicate");
}
if (ConfigHelper.getInputInitialAddress(conf) == null)
throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node");
// input partitioner is optional -- used only if requesting an ordered key scan
}
public List<InputSplit> getSplits(JobContext context) throws IOException

4
src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java
View File

@ -87,6 +87,10 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
{
throw new UnsupportedOperationException("you must set the keyspace and columnfamily with setColumnFamily()");
}
if (ConfigHelper.getOutputPartitioner(conf) == null)
throw new UnsupportedOperationException("You must set the output partitioner to the one used by your Cassandra cluster");
if (ConfigHelper.getOutputInitialAddress(conf) == null)
throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node");
}
/**

4
src/java/org/apache/cassandra/hadoop/ConfigHelper.java
View File

@ -317,7 +317,7 @@ public class ConfigHelper
public static int getInputRpcPort(Configuration conf)
{
return Integer.parseInt(conf.get(INPUT_THRIFT_PORT));
return Integer.parseInt(conf.get(INPUT_THRIFT_PORT, "9160"));
}
public static void setInputRpcPort(Configuration conf, String port)
@ -354,7 +354,7 @@ public class ConfigHelper
public static int getOutputRpcPort(Configuration conf)
{
return Integer.parseInt(conf.get(OUTPUT_THRIFT_PORT));
return Integer.parseInt(conf.get(OUTPUT_THRIFT_PORT, "9160"));
}
public static void setOutputRpcPort(Configuration conf, String port)