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

Add wide row support to ColumnFamilyInputFormat 

Patch by jbellis; reviewed by tjake for CASSANDRA-3264
This commit is contained in:
T Jake Luciani 2012-01-25 15:43:55 -05:00
parent 087d24a9d8
commit fba541c0c3
10 changed files with 1772 additions and 220 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -56,6 +56,7 @@
* Make CFMetaData conversions to/from thrift/native schema inverses
(CASSANDRA_3559)
* Add initial code for CQL 3.0-beta (CASSANDRA-3781)
* Add wide row support for ColumnFamilyInputFormat (CASSANDRA-3264)
1.0.8

0
examples/hadoop_word_count/bin/word_count_counters Normal file → Executable file
View File

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

@ -25,8 +25,6 @@ import org.apache.cassandra.hadoop.ColumnFamilyOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static com.google.common.base.Charsets.UTF_8;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.ConfigHelper;
@ -81,22 +79,28 @@ public class WordCount extends Configured implements Tool
protected void setup(org.apache.hadoop.mapreduce.Mapper.Context context)
throws IOException, InterruptedException
{
sourceColumn = ByteBufferUtil.bytes(context.getConfiguration().get(CONF_COLUMN_NAME));
}
public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns, Context context) throws IOException, InterruptedException
{
IColumn column = columns.get(sourceColumn);
if (column == null)
return;
String value = ByteBufferUtil.string(column.value());
logger.debug("read " + key + ":" + value + " from " + context.getInputSplit());
StringTokenizer itr = new StringTokenizer(value);
while (itr.hasMoreTokens())
for (IColumn column : columns.values())
{
word.set(itr.nextToken());
context.write(word, one);
String name = ByteBufferUtil.string(column.name());
String value = null;
if (name.contains("int"))
value = String.valueOf(ByteBufferUtil.toInt(column.value()));
else
value = ByteBufferUtil.string(column.value());
System.err.println("read " + ByteBufferUtil.string(key) + ":" +name + ":" + value + " from " + context.getInputSplit());
StringTokenizer itr = new StringTokenizer(value);
while (itr.hasMoreTokens())
{
word.set(itr.nextToken());
context.write(word, one);
}
}
}
}
@ -155,10 +159,12 @@ public class WordCount extends Configured implements Tool
}
logger.info("output reducer type: " + outputReducerType);
// use a smaller page size that doesn't divide the row count evenly to exercise the paging logic better
ConfigHelper.setRangeBatchSize(getConf(), 99);
for (int i = 0; i < WordCountSetup.TEST_COUNT; i++)
{
String columnName = "text" + i;
getConf().set(CONF_COLUMN_NAME, columnName);
Job job = new Job(getConf(), "wordcount");
job.setJarByClass(WordCount.class);
@ -184,6 +190,7 @@ public class WordCount extends Configured implements Tool
job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE, OUTPUT_COLUMN_FAMILY);
job.getConfiguration().set(CONF_COLUMN_NAME, "sum");
}
job.setInputFormatClass(ColumnFamilyInputFormat.class);
@ -194,12 +201,19 @@ public class WordCount extends Configured implements Tool
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(ByteBufferUtil.bytes(columnName)));
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
if (i == 4)
{
IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("int4"), IndexOperator.EQ, ByteBufferUtil.bytes(0));
ConfigHelper.setInputRange(job.getConfiguration(), Arrays.asList(expr));
}
if (i == 5)
{
// this will cause the predicate to be ignored in favor of scanning everything as a wide row
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY, true);
}
ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setOutputPartitioner(job.getConfiguration(), "RandomPartitioner");

2
examples/hadoop_word_count/src/WordCountSetup.java
View File

@ -36,7 +36,7 @@ public class WordCountSetup
{
private static final Logger logger = LoggerFactory.getLogger(WordCountSetup.class);
public static final int TEST_COUNT = 5;
public static final int TEST_COUNT = 6;
public static void main(String[] args) throws Exception
{

11
interface/cassandra.thrift
View File

@ -46,7 +46,7 @@ namespace rb CassandraThrift
# for every edit that doesn't result in a change to major/minor.
#
# See the Semantic Versioning Specification (SemVer) http://semver.org.
const string VERSION = "19.26.0"
const string VERSION = "19.27.0"
#
@ -535,6 +535,15 @@ service Cassandra {
4:required ConsistencyLevel consistency_level=ConsistencyLevel.ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
/**
returns a range of columns, wrapping to the next rows if necessary to collect max_results.
*/
list<KeySlice> get_paged_slice(1:required string column_family,
2:required KeyRange range,
3:required binary start_column,
4:required ConsistencyLevel consistency_level=ConsistencyLevel.ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
/**
Returns the subset of columns specified in SlicePredicate for the rows matching the IndexClause
@Deprecated; use get_range_slices instead with range.row_filter specified

1553
interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
View File

File diff suppressed because it is too large Load Diff

2
interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java
View File

@ -44,6 +44,6 @@ import org.slf4j.LoggerFactory;
public class Constants {
public static final String VERSION = "19.26.0";
public static final String VERSION = "19.27.0";
}

295
src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
View File

@ -30,6 +30,9 @@ import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.ImmutableSortedMap;
import com.google.common.collect.Iterables;
import org.apache.commons.lang.ArrayUtils;
import org.apache.cassandra.auth.IAuthenticator;
import org.apache.cassandra.config.ConfigurationException;
@ -41,7 +44,6 @@ import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
@ -61,7 +63,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
private SlicePredicate predicate;
private boolean isEmptyPredicate;
private int totalRowCount; // total number of rows to fetch
private int batchRowCount; // fetch this many per batch
private int batchSize; // fetch this many per batch
private String cfName;
private String keyspace;
private TSocket socket;
@ -69,6 +71,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
private ConsistencyLevel consistencyLevel;
private int keyBufferSize = 8192;
private List<IndexExpression> filter;
private boolean widerows;
public ColumnFamilyRecordReader()
{
@ -103,6 +106,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
public float getProgress()
{
// TODO this is totally broken for wide rows
// the progress is likely to be reported slightly off the actual but close enough
return ((float)iter.rowsRead()) / totalRowCount;
}
@ -135,9 +139,10 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
KeyRange jobRange = ConfigHelper.getInputKeyRange(conf);
filter = jobRange == null ? null : jobRange.row_filter;
predicate = ConfigHelper.getInputSlicePredicate(conf);
widerows = ConfigHelper.getInputIsWide(conf);
isEmptyPredicate = isEmptyPredicate(predicate);
totalRowCount = ConfigHelper.getInputSplitSize(conf);
batchRowCount = ConfigHelper.getRangeBatchSize(conf);
batchSize = ConfigHelper.getRangeBatchSize(conf);
cfName = ConfigHelper.getInputColumnFamily(conf);
consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getReadConsistencyLevel(conf));
@ -173,7 +178,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
throw new RuntimeException(e);
}
iter = new RowIterator();
iter = widerows ? new WideRowIterator() : new StaticRowIterator();
}
public boolean nextKeyValue() throws IOException
@ -222,15 +227,15 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
return split.getLocations()[0];
}
private class RowIterator extends AbstractIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>>>
private abstract class RowIterator extends AbstractIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>>>
{
private List<KeySlice> rows;
private String startToken;
private int totalRead = 0;
private int i = 0;
private final AbstractType<?> comparator;
private final AbstractType<?> subComparator;
private final IPartitioner partitioner;
protected List<KeySlice> rows;
protected KeySlice lastRow;
protected int totalRead = 0;
protected int i = 0;
protected final AbstractType<?> comparator;
protected final AbstractType<?> subComparator;
protected final IPartitioner partitioner;
private RowIterator()
{
@ -264,81 +269,6 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
}
}
private void maybeInit()
{
// check if we need another batch
if (rows != null && i >= rows.size())
rows = null;
if (rows != null)
return;
if (startToken == null)
{
startToken = split.getStartToken();
}
else if (startToken.equals(split.getEndToken()))
{
// reached end of the split
rows = null;
return;
}
KeyRange keyRange = new KeyRange(batchRowCount)
.setStart_token(startToken)
.setEnd_token(split.getEndToken())
.setRow_filter(filter);
try
{
rows = client.get_range_slices(new ColumnParent(cfName),
predicate,
keyRange,
consistencyLevel);
// nothing new? reached the end
if (rows.isEmpty())
{
rows = null;
return;
}
// prepare for the next slice to be read
KeySlice lastRow = rows.get(rows.size() - 1);
ByteBuffer rowkey = lastRow.key;
startToken = partitioner.getTokenFactory().toString(partitioner.getToken(rowkey));
// remove ghosts when fetching all columns
if (isEmptyPredicate)
{
Iterator<KeySlice> it = rows.iterator();
while(it.hasNext())
{
KeySlice ks = it.next();
if (ks.getColumnsSize() == 0)
{
it.remove();
}
}
// all ghosts, spooky
if (rows.isEmpty())
{
maybeInit();
return;
}
}
// reset to iterate through this new batch
i = 0;
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
/**
* @return total number of rows read by this record reader
*/
@ -347,24 +277,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
return totalRead;
}
protected Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> computeNext()
{
maybeInit();
if (rows == null)
return endOfData();
totalRead++;
KeySlice ks = rows.get(i++);
SortedMap<ByteBuffer, IColumn> map = new TreeMap<ByteBuffer, IColumn>(comparator);
for (ColumnOrSuperColumn cosc : ks.columns)
{
IColumn column = unthriftify(cosc);
map.put(column.name(), column);
}
return new Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>>(ks.key, map);
}
private IColumn unthriftify(ColumnOrSuperColumn cosc)
protected IColumn unthriftify(ColumnOrSuperColumn cosc)
{
if (cosc.counter_column != null)
return unthriftifyCounter(cosc.counter_column);
@ -407,6 +320,178 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
}
}
private class StaticRowIterator extends RowIterator
{
private void maybeInit()
{
// check if we need another batch
if (rows != null && i >= rows.size())
rows = null;
if (rows != null)
return;
String startToken;
if (lastRow == null)
{
startToken = split.getStartToken();
}
else
{
startToken = partitioner.getTokenFactory().toString(partitioner.getToken(lastRow.key));
if (startToken.equals(split.getEndToken()))
{
// reached end of the split
rows = null;
return;
}
}
KeyRange keyRange = new KeyRange(batchSize)
.setStart_token(startToken)
.setEnd_token(split.getEndToken())
.setRow_filter(filter);
try
{
rows = client.get_range_slices(new ColumnParent(cfName), predicate, keyRange, consistencyLevel);
// nothing new? reached the end
if (rows.isEmpty())
{
rows = null;
return;
}
// prepare for the next slice to be read
lastRow = Iterables.getLast(rows);
// remove ghosts when fetching all columns
if (isEmptyPredicate)
{
Iterator<KeySlice> it = rows.iterator();
while (it.hasNext())
{
KeySlice ks = it.next();
if (ks.getColumnsSize() == 0)
{
it.remove();
}
}
// all ghosts, spooky
if (rows.isEmpty())
{
maybeInit();
return;
}
}
// reset to iterate through this new batch
i = 0;
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
protected Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> computeNext()
{
maybeInit();
if (rows == null)
return endOfData();
totalRead++;
KeySlice ks = rows.get(i++);
SortedMap<ByteBuffer, IColumn> map = new TreeMap<ByteBuffer, IColumn>(comparator);
for (ColumnOrSuperColumn cosc : ks.columns)
{
IColumn column = unthriftify(cosc);
map.put(column.name(), column);
}
return new Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>>(ks.key, map);
}
}
private class WideRowIterator extends RowIterator
{
private Iterator<ColumnOrSuperColumn> wideColumns;
private void maybeInit()
{
if (wideColumns != null && wideColumns.hasNext())
return;
// check if we need another batch
if (rows != null && ++i >= rows.size())
rows = null;
if (rows != null)
{
wideColumns = rows.get(i).columns.iterator();
return;
}
String startToken;
ByteBuffer startColumn;
if (lastRow == null)
{
startToken = split.getStartToken();
startColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER;
}
else
{
startToken = partitioner.getTokenFactory().toString(partitioner.getToken(lastRow.key));
startColumn = Iterables.getLast(lastRow.columns).column.name;
}
KeyRange keyRange = new KeyRange(batchSize)
.setStart_token(startToken)
.setEnd_token(split.getEndToken())
.setRow_filter(filter);
try
{
rows = client.get_paged_slice(cfName, keyRange, startColumn, consistencyLevel);
// nothing found?
if (rows == null || rows.isEmpty() || rows.get(0).columns.isEmpty())
{
rows = null;
return;
}
// nothing new? reached the end
if (lastRow != null && (rows.get(0).key.equals(lastRow.key) || rows.get(0).columns.get(0).column.equals(startColumn)))
{
rows = null;
return;
}
// prepare for the next slice to be read
lastRow = Iterables.getLast(rows);
// reset to iterate through this new batch
i = 0;
wideColumns = rows.get(i).columns.iterator();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
protected Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> computeNext()
{
maybeInit();
if (rows == null)
return endOfData();
totalRead++;
ColumnOrSuperColumn cosc = wideColumns.next();
ImmutableSortedMap<ByteBuffer, IColumn> map = ImmutableSortedMap.of(cosc.column.name, unthriftify(cosc));
return Pair.<ByteBuffer, SortedMap<ByteBuffer, IColumn>>create(rows.get(i).key, map);
}
}
// Because the old Hadoop API wants us to write to the key and value
// and the new asks for them, we need to copy the output of the new API

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

@ -19,10 +19,14 @@ package org.apache.cassandra.hadoop;
* under the License.
*
*/
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.thrift.*;
@ -37,8 +41,6 @@ import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConfigHelper
@ -56,6 +58,7 @@ public class ConfigHelper
private static final String INPUT_PREDICATE_CONFIG = "cassandra.input.predicate";
private static final String INPUT_KEYRANGE_CONFIG = "cassandra.input.keyRange";
private static final String INPUT_SPLIT_SIZE_CONFIG = "cassandra.input.split.size";
private static final String INPUT_WIDEROWS_CONFIG = "cassandra.input.widerows";
private static final int DEFAULT_SPLIT_SIZE = 64 * 1024;
private static final String RANGE_BATCH_SIZE_CONFIG = "cassandra.range.batch.size";
private static final int DEFAULT_RANGE_BATCH_SIZE = 4096;
@ -71,13 +74,13 @@ public class ConfigHelper
/**
* Set the keyspace and column family for the input of this job.
* Comparator and Partitioner types will be read from storage-conf.xml.
*
* @param conf Job configuration you are about to run
* @param keyspace
* @param columnFamily
* @param widerows
*/
public static void setInputColumnFamily(Configuration conf, String keyspace, String columnFamily)
public static void setInputColumnFamily(Configuration conf, String keyspace, String columnFamily, boolean widerows)
{
if (keyspace == null)
{
@ -90,6 +93,19 @@ public class ConfigHelper
conf.set(INPUT_KEYSPACE_CONFIG, keyspace);
conf.set(INPUT_COLUMNFAMILY_CONFIG, columnFamily);
conf.set(INPUT_WIDEROWS_CONFIG, String.valueOf(widerows));
}
/**
* Set the keyspace and column family for the input of this job.
*
* @param conf Job configuration you are about to run
* @param keyspace
* @param columnFamily
*/
public static void setInputColumnFamily(Configuration conf, String keyspace, String columnFamily)
{
setInputColumnFamily(conf, keyspace, columnFamily, false);
}
/**
@ -175,7 +191,7 @@ public class ConfigHelper
{
return predicateFromString(conf.get(INPUT_PREDICATE_CONFIG));
}
public static String getRawInputSlicePredicate(Configuration conf)
{
return conf.get(INPUT_PREDICATE_CONFIG);
@ -299,6 +315,11 @@ public class ConfigHelper
{
return conf.get(INPUT_COLUMNFAMILY_CONFIG);
}
public static boolean getInputIsWide(Configuration conf)
{
return Boolean.valueOf(conf.get(INPUT_WIDEROWS_CONFIG));
}
public static String getOutputColumnFamily(Configuration conf)
{

55
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -711,6 +711,61 @@ public class CassandraServer implements Cassandra.Iface
return thriftifyKeySlices(rows, column_parent, predicate);
}
public List<KeySlice> get_paged_slice(String column_family, KeyRange range, ByteBuffer start_column, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
logger.debug("get_paged_slice");
String keyspace = state().getKeyspace();
state().hasColumnFamilyAccess(column_family, Permission.READ);
CFMetaData metadata = ThriftValidation.validateColumnFamily(keyspace, column_family);
ThriftValidation.validateKeyRange(metadata, null, range);
ThriftValidation.validateConsistencyLevel(keyspace, consistency_level, RequestType.READ);
SlicePredicate predicate = new SlicePredicate().setSlice_range(new SliceRange(start_column, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, -1));
IPartitioner p = StorageService.getPartitioner();
AbstractBounds<RowPosition> bounds;
if (range.start_key == null)
{
Token.TokenFactory tokenFactory = p.getTokenFactory();
Token left = tokenFactory.fromString(range.start_token);
Token right = tokenFactory.fromString(range.end_token);
bounds = Range.makeRowRange(left, right, p);
}
else
{
bounds = new Bounds<RowPosition>(RowPosition.forKey(range.start_key, p), RowPosition.forKey(range.end_key, p));
}
List<Row> rows;
try
{
schedule(DatabaseDescriptor.getRpcTimeout());
try
{
rows = StorageProxy.getRangeSlice(new RangeSliceCommand(keyspace, column_family, null, predicate, bounds, range.row_filter, range.count, true), consistency_level);
}
finally
{
release();
}
assert rows != null;
}
catch (TimeoutException e)
{
logger.debug("... timed out");
throw new TimedOutException();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
return thriftifyKeySlices(rows, new ColumnParent(column_family), predicate);
}
private List<KeySlice> thriftifyKeySlices(List<Row> rows, ColumnParent column_parent, SlicePredicate predicate)
{
List<KeySlice> keySlices = new ArrayList<KeySlice>(rows.size());