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

merge from 0.8 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1126664 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
T Jake Luciani 2011-05-23 19:03:33 +00:00
parent 43aa804048 8becda0810
commit be67fda27e
14 changed files with 52 additions and 471 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGES.txt
View File

@ -18,6 +18,8 @@
* Assert ranges are not overlapping in AB.normalize (CASSANDRA-2641) * Assert ranges are not overlapping in AB.normalize (CASSANDRA-2641)
* Don't write CounterUpdateColumn to disk in tests (CASSANDRA-2650) * Don't write CounterUpdateColumn to disk in tests (CASSANDRA-2650)
* Add sstable bulk loading utility (CASSANDRA-1278) * Add sstable bulk loading utility (CASSANDRA-1278)
* avoid replaying hints to dropped columnfamilies (CASSANDRA-2685)
* add placeholders for missing rows in range query pseudo-RR (CASSANDRA-2680)
0.8.0-final 0.8.0-final
@ -26,7 +28,8 @@
* update CQL consistency levels (CASSANDRA-2566) * update CQL consistency levels (CASSANDRA-2566)
* debian packaging fixes (CASSANDRA-2481, 2647) * debian packaging fixes (CASSANDRA-2481, 2647)
* fix UUIDType, IntegerType for direct buffers (CASSANDRA-2682, 2684) * fix UUIDType, IntegerType for direct buffers (CASSANDRA-2682, 2684)
* switch to native Thrift for Hadoop map/reduce (CASSANDRA-2667)
* fix StackOverflowError when building from eclipse (CASSANDRA-2687)
0.8.0-rc1 0.8.0-rc1
* faster flushes and compaction from fixing excessively pessimistic * faster flushes and compaction from fixing excessively pessimistic

9
NEWS.txt
View File

@ -13,8 +13,10 @@ Upgrading
- 0.8 is fully API-compatible with 0.7. You can continue - 0.8 is fully API-compatible with 0.7. You can continue
to use your 0.7 clients. to use your 0.7 clients.
- Avro record classes used in map/reduce and Hadoop streaming code have - Avro record classes used in map/reduce and Hadoop streaming code have
moved from org.apache.cassandra.avro to org.apache.cassandra.hadoop.avro, been removed. Map/reduce can be switched to Thrift by changing
applications using these classes will need to be updated accordingly. org.apache.cassandra.avro in import statements to
org.apache.cassandra.thrift (no class names change). Streaming support
has been removed for the time being.
- The loadbalance command has been removed from nodetool. For similar - The loadbalance command has been removed from nodetool. For similar
behavior, decommission then rebootstrap with empty initial_token. behavior, decommission then rebootstrap with empty initial_token.
- Thrift unframed mode has been removed. - Thrift unframed mode has been removed.
@ -57,6 +59,9 @@ Other
to compact that anyway (which will free up space if there are to compact that anyway (which will free up space if there are
a lot of expired tombstones), use the new forceUserDefinedCompaction a lot of expired tombstones), use the new forceUserDefinedCompaction
JMX method on CompactionManager. JMX method on CompactionManager.
- most of contrib/ (which was not part of the binary releases)
has been moved either to examples/ or tools/. We plan to move the
rest for 0.8.1.
JMX JMX
--- ---

8
build.xml
View File

@ -1252,10 +1252,10 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<mkdir dir=".externalToolBuilders" /> <mkdir dir=".externalToolBuilders" />
<echo file=".externalToolBuilders/Cassandra_Ant_Builder.launch"><![CDATA[<?xml version="1.0" encoding="UTF-8" standalone="no"?> <echo file=".externalToolBuilders/Cassandra_Ant_Builder.launch"><![CDATA[<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<launchConfiguration type="org.eclipse.ant.AntBuilderLaunchConfigurationType"> <launchConfiguration type="org.eclipse.ant.AntBuilderLaunchConfigurationType">
<stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_AFTER_CLEAN_TARGETS" value="build,build-test,"/> <stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_AFTER_CLEAN_TARGETS" value="build-test,"/>
<stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_AUTO_TARGETS" value="build,build-test,"/> <stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_AUTO_TARGETS" value="build-test,"/>
<stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_CLEAN_TARGETS" value="build,build-test,"/> <stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_CLEAN_TARGETS" value="clean"/>
<stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_MANUAL_TARGETS" value="build,build-test,"/> <stringAttribute key="org.eclipse.ant.ui.ATTR_ANT_MANUAL_TARGETS" value="build-test,"/>
<booleanAttribute key="org.eclipse.ant.ui.ATTR_TARGETS_UPDATED" value="true"/> <booleanAttribute key="org.eclipse.ant.ui.ATTR_TARGETS_UPDATED" value="true"/>
<booleanAttribute key="org.eclipse.ant.ui.DEFAULT_VM_INSTALL" value="false"/> <booleanAttribute key="org.eclipse.ant.ui.DEFAULT_VM_INSTALL" value="false"/>
<listAttribute key="org.eclipse.debug.core.MAPPED_RESOURCE_PATHS"> <listAttribute key="org.eclipse.debug.core.MAPPED_RESOURCE_PATHS">

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

@ -22,6 +22,7 @@ import java.util.*;
import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.marshal.BytesType; import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.thrift.*; import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FBUtilities;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
@ -32,9 +33,9 @@ import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.SuperColumn; import org.apache.cassandra.db.SuperColumn;
import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.hadoop.*; import org.apache.cassandra.hadoop.*;
import org.apache.cassandra.hadoop.avro.Mutation; import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.hadoop.avro.Deletion; import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.hadoop.avro.ColumnOrSuperColumn; import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
@ -158,7 +159,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface
// super // super
ArrayList<Tuple> subcols = new ArrayList<Tuple>(); ArrayList<Tuple> subcols = new ArrayList<Tuple>();
for (IColumn subcol : ((SuperColumn)col).getSubColumns()) for (IColumn subcol : col.getSubColumns())
subcols.add(columnToTuple(subcol.name(), subcol, cfDef)); subcols.add(columnToTuple(subcol.name(), subcol, cfDef));
pair.set(1, new DefaultDataBag(subcols)); pair.set(1, new DefaultDataBag(subcols));
@ -179,8 +180,8 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface
AbstractType default_validator = null; AbstractType default_validator = null;
try try
{ {
comparator = FBUtilities.getComparator(cfDef.comparator_type); comparator = TypeParser.parse(cfDef.comparator_type);
default_validator = FBUtilities.getComparator(cfDef.default_validation_class); default_validator = TypeParser.parse(cfDef.default_validation_class);
} }
catch (ConfigurationException e) catch (ConfigurationException e)
{ {
@ -202,7 +203,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface
AbstractType validator = null; AbstractType validator = null;
try try
{ {
validator = FBUtilities.getComparator(cd.getValidation_class()); validator = TypeParser.parse(cd.getValidation_class());
validators.put(cd.name, validator); validators.put(cd.name, validator);
} }
catch (ConfigurationException e) catch (ConfigurationException e)
@ -385,13 +386,13 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface
if (pair.get(1) == null) if (pair.get(1) == null)
{ {
mutation.deletion = new Deletion(); mutation.deletion = new Deletion();
mutation.deletion.predicate = new org.apache.cassandra.hadoop.avro.SlicePredicate(); mutation.deletion.predicate = new org.apache.cassandra.thrift.SlicePredicate();
mutation.deletion.predicate.column_names = Arrays.asList(objToBB(pair.get(0))); mutation.deletion.predicate.column_names = Arrays.asList(objToBB(pair.get(0)));
mutation.deletion.timestamp = System.currentTimeMillis() * 1000; mutation.deletion.timestamp = System.currentTimeMillis() * 1000;
} }
else else
{ {
org.apache.cassandra.hadoop.avro.Column column = new org.apache.cassandra.hadoop.avro.Column(); org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column();
column.name = marshallers.get(0).decompose((pair.get(0))); column.name = marshallers.get(0).decompose((pair.get(0)));
if (validators.get(column.name) == null) if (validators.get(column.name) == null)
// Have to special case BytesType to convert DataByteArray into ByteBuffer // Have to special case BytesType to convert DataByteArray into ByteBuffer

24
examples/hadoop_streaming_output/README.txt
View File

@ -1,24 +0,0 @@
Hadoop Streaming WordCount example: runs a simple Python wordcount mapper script, and a slightly
more complex Cassandra specific reducer that outputs the results of the wordcount as Avro
records to Cassandra.
Dependencies for the example:
* Hadoop 0.20.x installed, such that the 'bin/hadoop' launcher is available on your path
* Avro's Python library installed
Unfortunately, due to the way Hadoop builds its CLASSPATH, it is also necessary to
upgrade a library that conflicts between Hadoop and Avro. Within your Hadoop distribution,
you'll need to ensure that the following jars are of a sufficiently high version, or else
you will see a classloader error at runtime:
* jackson-core-asl >= 1.4.0
* jackson-mapper-asl >= 1.4.0
To run the example, edit bin/streaming to point to a valid Cassandra cluster and
then execute it over a text input, located on the local filesystem or in Hadoop:
$ bin/streaming -input <mytxtfile.txt>
Hadoop streaming will execute a simple wordcount over your input files, and write the generated
counts to Keyspace1,Standard1. bin/reducer.py gives an example of how to format the output of
a script as Avro records which can be consumed by Cassandra's AvroOutputReader, and bin/streaming
shows the necessary incantations to execute a Hadoop Streaming job with Cassandra as output.

43
examples/hadoop_streaming_output/bin/mapper.py
View File

@ -1,43 +0,0 @@
#!/usr/bin/env python
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Simple reducer for writing to Cassandra using Hadoop streaming, Python and Avro,
# based on http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python
# Wordcount mapper
# based on http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python
import sys
# input comes from STDIN (standard input)
for line in sys.stdin:
# remove leading and trailing whitespace
line = line.strip()
# split the line into words
words = line.split()
# increase counters
for word in words:
# write the results to STDOUT (standard output);
# what we output here will be the input for the
# Reduce step, i.e. the input for reducer.py
#
# tab-delimited; the trivial word count is 1
print '%s\t%s' % (word, 1)

72
examples/hadoop_streaming_output/bin/reducer.py
View File

@ -1,72 +0,0 @@
#!/usr/bin/env python
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Wordcount reducer for writing to Cassandra using Hadoop streaming and Avro,
# based on http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python
from avro.io import BinaryEncoder, DatumWriter
import avro.protocol
import sys,time
# input comes from STDIN (standard input)
word2count = {}
for line in sys.stdin:
# remove leading and trailing whitespace
line = line.strip()
# parse the input we got from mapper.py
word, count = line.split('\t', 1)
# convert count (currently a string) to int
try:
count = int(count)
word2count[word] = word2count.get(word, 0) + count
except ValueError:
# count was not a number, so silently
# ignore/discard this line
pass
#
# NB: the AvroOutputReader specific portion begins here
#
def new_column(name, value):
column = dict()
column['name'] = '%s' % name
column['value'] = '%s' % value
column['timestamp'] = long(time.time() * 1e6)
column['ttl'] = 0
return column
# parse the current avro schema
proto = avro.protocol.parse(open('cassandra.avpr').read())
schema = proto.types_dict['StreamingMutation']
# open an avro encoder and writer for stdout
enc = BinaryEncoder(sys.stdout)
writer = DatumWriter(schema)
# output a series of objects matching 'StreamingMutation' in the Avro interface
smutation = dict()
try:
for word, count in word2count.iteritems():
smutation['key'] = word
smutation['mutation'] = {'column_or_supercolumn': {'column': new_column('count', count)}}
writer.write(smutation, enc)
finally:
sys.stdout.flush()

40
examples/hadoop_streaming_output/bin/streaming
View File

@ -1,40 +0,0 @@
#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
cwd=`dirname $0`
ARCHIVES=`ls -1 $cwd/../../../build/apache-cassandra*.jar`
for jar in `ls -1 $cwd/../../../build/lib/jars/*.jar $cwd/../../../lib/*.jar`; do
ARCHIVES=$ARCHIVES,$jar
done
hadoop jar $cwd/../../../build/lib/jars/hadoop-streaming*.jar \
-D stream.reduce.output=cassandra_avro_output \
-D stream.io.identifier.resolver.class=org.apache.cassandra.hadoop.streaming.AvroResolver \
-D cassandra.output.keyspace=Keyspace1 \
-D cassandra.output.columnfamily=Standard1 \
-D cassandra.partitioner.class=org.apache.cassandra.dht.RandomPartitioner \
-D cassandra.thrift.address=127.0.0.1 \
-D cassandra.thrift.port=9160 \
-libjars $ARCHIVES \
-file $cwd/../../../src/gen-java/org/apache/cassandra/hadoop/hadoop.avpr \
-outputformat org.apache.cassandra.hadoop.ColumnFamilyOutputFormat \
-output /tmp/ignored \
-mapper $cwd/mapper.py -reducer $cwd/reducer.py \
$*

8
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -30,6 +30,7 @@ import java.util.concurrent.TimeoutException;
import static com.google.common.base.Charsets.UTF_8; import static com.google.common.base.Charsets.UTF_8;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FBUtilities;
import org.apache.commons.lang.ArrayUtils; import org.apache.commons.lang.ArrayUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -127,8 +128,12 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
return false; return false;
} }
if (CFMetaData.getId(tableName, cfName) == null)
{
logger_.debug("Discarding hints for dropped keyspace or columnfamily {}/{}", tableName, cfName);
return true;
}
Table table = Table.open(tableName); Table table = Table.open(tableName);
DecoratedKey<?> dkey = StorageService.getPartitioner().decorateKey(key);
ColumnFamilyStore cfs = table.getColumnFamilyStore(cfName); ColumnFamilyStore cfs = table.getColumnFamilyStore(cfName);
int pageSize = PAGE_SIZE; int pageSize = PAGE_SIZE;
@ -141,6 +146,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
logger_.debug("average hinted-row column size is {}; using pageSize of {}", averageColumnSize, pageSize); logger_.debug("average hinted-row column size is {}; using pageSize of {}", averageColumnSize, pageSize);
} }
DecoratedKey<?> dkey = StorageService.getPartitioner().decorateKey(key);
ByteBuffer startColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER; ByteBuffer startColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER;
while (true) while (true)
{ {

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

@ -31,7 +31,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.apache.cassandra.auth.SimpleAuthenticator; import org.apache.cassandra.auth.SimpleAuthenticator;
import org.apache.cassandra.hadoop.avro.Mutation;
import org.apache.cassandra.thrift.*; import org.apache.cassandra.thrift.*;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.*; import org.apache.hadoop.mapreduce.*;

79
src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
View File

@ -59,8 +59,8 @@ import org.apache.cassandra.utils.ByteBufferUtil;
* @see OutputFormat * @see OutputFormat
* *
*/ */
final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.apache.cassandra.hadoop.avro.Mutation>> final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<Mutation>>
implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<org.apache.cassandra.hadoop.avro.Mutation>> implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<Mutation>>
{ {
// The configuration this writer is associated with. // The configuration this writer is associated with.
private final Configuration conf; private final Configuration conf;
@ -122,7 +122,7 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<org.apache.cass
* @throws IOException * @throws IOException
*/ */
@Override @Override
public void write(ByteBuffer keybuff, List<org.apache.cassandra.hadoop.avro.Mutation> value) throws IOException public void write(ByteBuffer keybuff, List<Mutation> value) throws IOException
{ {
Range range = ringCache.getRange(keybuff); Range range = ringCache.getRange(keybuff);
@ -136,77 +136,8 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<org.apache.cass
clients.put(range, client); clients.put(range, client);
} }
for (org.apache.cassandra.hadoop.avro.Mutation amut : value) for (Mutation amut : value)
client.put(new Pair<ByteBuffer,Mutation>(keybuff, avroToThrift(amut))); client.put(new Pair<ByteBuffer,Mutation>(keybuff, amut));
}
/**
* Deep copies the given Avro mutation into a new Thrift mutation.
*/
private Mutation avroToThrift(org.apache.cassandra.hadoop.avro.Mutation amut)
{
Mutation mutation = new Mutation();
org.apache.cassandra.hadoop.avro.ColumnOrSuperColumn acosc = amut.column_or_supercolumn;
if (acosc == null)
{
// deletion
assert amut.deletion != null;
Deletion deletion = new Deletion().setTimestamp(amut.deletion.timestamp);
mutation.setDeletion(deletion);
org.apache.cassandra.hadoop.avro.SlicePredicate apred = amut.deletion.predicate;
if (apred == null && amut.deletion.super_column == null)
{
// leave Deletion alone to delete entire row
}
else if (amut.deletion.super_column != null)
{
// super column
deletion.setSuper_column(ByteBufferUtil.getArray(amut.deletion.super_column));
}
else if (apred.column_names != null)
{
// column names
List<ByteBuffer> names = new ArrayList<ByteBuffer>(apred.column_names.size());
for (ByteBuffer name : apred.column_names)
names.add(name);
deletion.setPredicate(new SlicePredicate().setColumn_names(names));
}
else
{
// range
deletion.setPredicate(new SlicePredicate().setSlice_range(avroToThrift(apred.slice_range)));
}
}
else
{
// creation
ColumnOrSuperColumn cosc = new ColumnOrSuperColumn();
mutation.setColumn_or_supercolumn(cosc);
if (acosc.column != null)
// standard column
cosc.setColumn(avroToThrift(acosc.column));
else
{
// super column
ByteBuffer scolname = acosc.super_column.name;
List<Column> scolcols = new ArrayList<Column>(acosc.super_column.columns.size());
for (org.apache.cassandra.hadoop.avro.Column acol : acosc.super_column.columns)
scolcols.add(avroToThrift(acol));
cosc.setSuper_column(new SuperColumn(scolname, scolcols));
}
}
return mutation;
}
private SliceRange avroToThrift(org.apache.cassandra.hadoop.avro.SliceRange asr)
{
return new SliceRange(asr.start, asr.finish, asr.reversed, asr.count);
}
private Column avroToThrift(org.apache.cassandra.hadoop.avro.Column acol)
{
return new Column(acol.name).setValue(acol.value).setTimestamp(acol.timestamp);
} }
/** /**

147
src/java/org/apache/cassandra/hadoop/streaming/AvroOutputReader.java
View File

@ -1,147 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.hadoop.streaming;
import java.io.DataInput;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.cassandra.hadoop.avro.Mutation;
import org.apache.cassandra.hadoop.avro.StreamingMutation;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.hadoop.streaming.PipeMapRed;
import org.apache.hadoop.streaming.io.OutputReader;
/**
* An OutputReader that reads sequential StreamingMutations (from Cassandra's Avro client API), and converts them to
* the objects used by CassandraOutputFormat. This allows Hadoop Streaming to output efficiently to Cassandra via
* a familiar API.
*
* Avro requires the reader's and writer's schema: otherwise, it assumes they are the same.
* If the canonical schema that the Cassandra side uses changes, and somebody packaged the {{avpr}}
* up in their application somehow, or generated code, they'd see a runtime failure.
* We could allow specifying an alternate Avro schema using a Configuration property to work around this.
*/
public class AvroOutputReader extends OutputReader<ByteBuffer, List<Mutation>>
{
private BinaryDecoder decoder;
private SpecificDatumReader<StreamingMutation> reader;
// reusable values
private final StreamingMutation entry = new StreamingMutation();
private final ArrayList<Mutation> mutations = new ArrayList<Mutation>(1);
@Override
public void initialize(PipeMapRed pmr) throws IOException
{
super.initialize(pmr);
// set up decoding around the DataInput (hmm) provided by streaming
InputStream in;
if (pmr.getClientInput() instanceof InputStream)
// let's hope this is the case
in = (InputStream)pmr.getClientInput();
else
// ...because this is relatively slow
in = new FromDataInputStream(pmr.getClientInput());
decoder = DecoderFactory.defaultFactory().createBinaryDecoder(in, null);
reader = new SpecificDatumReader<StreamingMutation>(StreamingMutation.SCHEMA$);
}
@Override
public boolean readKeyValue() throws IOException
{
try
{
reader.read(entry, decoder);
}
catch (EOFException e)
{
return false;
}
mutations.clear();
mutations.add(entry.mutation);
return true;
}
@Override
public ByteBuffer getCurrentKey() throws IOException
{
return entry.key;
}
@Override
public List<Mutation> getCurrentValue() throws IOException
{
return mutations;
}
@Override
public String getLastOutput()
{
return entry.toString();
}
/**
* Wraps a DataInput to extend InputStream. The exception handling in read() is likely to be ridiculous slow.
*/
private static final class FromDataInputStream extends InputStream
{
private final DataInput in;
public FromDataInputStream(DataInput in)
{
this.in = in;
}
@Override
public boolean markSupported()
{
return false;
}
@Override
public int read() throws IOException
{
try
{
return in.readUnsignedByte();
}
catch (EOFException e)
{
return -1;
}
}
@Override
public long skip(long n) throws IOException
{
FileUtils.skipBytesFully(in, n);
return n;
}
}
}

50
src/java/org/apache/cassandra/hadoop/streaming/AvroResolver.java
View File

@ -1,50 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.hadoop.streaming;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.hadoop.streaming.io.IdentifierResolver;
/**
* Resolves AVRO_ID to the appropriate OutputReader and K/V classes for Cassandra output.
*
* TODO: usage explanation
*/
public class AvroResolver extends IdentifierResolver
{
public static final String AVRO_ID = "cassandra_avro_output";
@Override
public void resolve(String identifier)
{
if (!identifier.equalsIgnoreCase(AVRO_ID))
{
super.resolve(identifier);
return;
}
setInputWriterClass(null);
setOutputReaderClass(AvroOutputReader.class);
setOutputKeyClass(ByteBuffer.class);
setOutputValueClass(List.class);
}
}

18
src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
View File

@ -24,8 +24,6 @@ import java.util.*;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
import com.google.common.collect.AbstractIterator; import com.google.common.collect.AbstractIterator;
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import org.apache.commons.collections.iterators.CollatingIterator; import org.apache.commons.collections.iterators.CollatingIterator;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -104,7 +102,21 @@ public class RangeSliceResponseResolver implements IResponseResolver<Iterable<Ro
protected Row getReduced() protected Row getReduced()
{ {
ColumnFamily resolved = RowRepairResolver.resolveSuperset(versions); ColumnFamily resolved = versions.size() > 1
? RowRepairResolver.resolveSuperset(versions)
: versions.get(0);
if (versions.size() < sources.size())
{
// add placeholder rows for sources that didn't have any data, so maybeScheduleRepairs sees them
for (InetAddress source : sources)
{
if (!versionSources.contains(source))
{
versions.add(null);
versionSources.add(source);
}
}
}
RowRepairResolver.maybeScheduleRepairs(resolved, table, key, versions, versionSources); RowRepairResolver.maybeScheduleRepairs(resolved, table, key, versions, versionSources);
versions.clear(); versions.clear();
versionSources.clear(); versionSources.clear();