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

support for Hadoop Streaming [non-jvm map/reduce via stdin/out]. patch by Stu Hood; reviewed by jbellis for CASSANDRA-1368 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@992475 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-09-03 21:20:10 +00:00
parent 53ea7efe97
commit 8c5069fba2
6 changed files with 72 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -47,6 +47,8 @@ dev
* change multiget key collection from list to set (CASSANDRA-1329)
* ability to modify keyspaces and column family definitions on a live cluster
(CASSANDRA-1285)
* support for Hadoop Streaming [non-jvm map/reduce via stdin/out]
(CASSANDRA-1368)
0.7-beta1

3
NEWS.txt
View File

@ -15,7 +15,8 @@ Features
- Optional per-Column time-to-live field allows expiring data without
have to issue explicit remove commands
- `truncate` thrift method allows clearing an entire ColumnFamily at once
- Hadoop OutputFormat support
- Hadoop OutputFormat and Streaming [non-jvm map/reduce via stdin/out]
support
- Up to 8x faster reads from row cache
- A new ByteOrderedPartitioner supports bytes keys with arbitrary content,
and orders keys by their byte value. This should be used in new

5
interface/cassandra.genavro
View File

@ -169,6 +169,11 @@ protocol Cassandra {
array<CfDef> cf_defs;
}
record StreamingMutation {
bytes key;
Mutation mutation;
}
record MutationsMapEntry {
bytes key;
map<array<Mutation>> mutations;

1
ivy.xml
View File

@ -28,6 +28,7 @@
<dependency org="commons-logging" name="commons-logging" rev="1.1.1"/>
<dependency org="org.apache.rat" name="apache-rat" rev="0.6" />
<dependency org="com.cloudera.hadoop" name="hadoop-core" rev="0.20.2-320"/>
<dependency org="com.cloudera.hadoop" name="hadoop-streaming" rev="0.20.2-320"/>
<dependency org="net.sf.jopt-simple" name="jopt-simple" rev="3.2"/>
<dependency org="net.java.dev.jna" name="jna" rev="3.2.7"/>

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

@ -69,6 +69,7 @@ import org.slf4j.LoggerFactory;
* </p>
*/
public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutation>>
implements org.apache.hadoop.mapred.OutputFormat<ByteBuffer,List<Mutation>>
{
private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyOutputFormat.class);
@ -85,13 +86,17 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
@Override
public void checkOutputSpecs(JobContext context)
{
Configuration conf = context.getConfiguration();
checkOutputSpecs(context.getConfiguration());
}
private void checkOutputSpecs(Configuration conf)
{
if (ConfigHelper.getOutputKeyspace(conf) == null || ConfigHelper.getOutputColumnFamily(conf) == null)
{
throw new UnsupportedOperationException("you must set the keyspace and columnfamily with setColumnFamily()");
}
}
/**
* The OutputCommitter for this format does not write any data to the DFS.
*
@ -107,6 +112,20 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
return new NullOutputCommitter();
}
/** Fills the deprecated OutputFormat interface for streaming. */
@Deprecated @Override
public void checkOutputSpecs(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job) throws IOException
{
checkOutputSpecs(job);
}
/** Fills the deprecated OutputFormat interface for streaming. */
@Deprecated @Override
public ColumnFamilyRecordWriter getRecordWriter(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job, String name, org.apache.hadoop.util.Progressable progress) throws IOException
{
return new ColumnFamilyRecordWriter(job);
}
/**
* Get the {@link RecordWriter} for the given task.
*
@ -116,7 +135,7 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
* @throws IOException
*/
@Override
public RecordWriter<ByteBuffer,List<Mutation>> getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
public ColumnFamilyRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
{
return new ColumnFamilyRecordWriter(context);
}
@ -126,30 +145,29 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
* keyspace, and is logged in with the configured credentials.
*
* @param socket a socket pointing to a particular node, seed or otherwise
* @param context a job context
* @param conf a job configuration
* @return a cassandra client
* @throws InvalidRequestException
* @throws TException
* @throws AuthenticationException
* @throws AuthorizationException
*/
public static Cassandra.Client createAuthenticatedClient(TSocket socket, JobContext context)
public static Cassandra.Client createAuthenticatedClient(TSocket socket, Configuration conf)
throws InvalidRequestException, TException, AuthenticationException, AuthorizationException
{
TBinaryProtocol binaryProtocol = new TBinaryProtocol(new TFramedTransport(socket));
Cassandra.Client client = new Cassandra.Client(binaryProtocol);
socket.open();
client.set_keyspace(ConfigHelper.getOutputKeyspace(context.getConfiguration()));
if (ConfigHelper.getOutputKeyspaceUserName(context.getConfiguration()) != null)
client.set_keyspace(ConfigHelper.getOutputKeyspace(conf));
if (ConfigHelper.getOutputKeyspaceUserName(conf) != null)
{
Map<String, String> creds = new HashMap<String, String>();
creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(context.getConfiguration()));
creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(context.getConfiguration()));
creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(conf));
creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(conf));
AuthenticationRequest authRequest = new AuthenticationRequest(creds);
client.login(authRequest);
}
return client;
}
/**

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

@ -47,6 +47,8 @@ import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.SuperColumn;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
@ -75,10 +77,10 @@ import org.apache.thrift.transport.TSocket;
* @see OutputFormat
*
*/
final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.apache.cassandra.avro.Mutation>>
final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.apache.cassandra.avro.Mutation>> implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<org.apache.cassandra.avro.Mutation>>
{
// The task attempt context this writer is associated with.
private final TaskAttemptContext context;
// The configuration this writer is associated with.
private final Configuration conf;
// The batched set of mutations grouped by endpoints.
private Map<InetAddress,Map<byte[],Map<String,List<Mutation>>>> mutationsByEndpoint;
@ -104,15 +106,20 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
*/
ColumnFamilyRecordWriter(TaskAttemptContext context) throws IOException
{
this.context = context;
this.mutationsByEndpoint = new HashMap<InetAddress,Map<byte[],Map<String,List<Mutation>>>>();
this.ringCache = new RingCache(ConfigHelper.getOutputKeyspace(context.getConfiguration()),
ConfigHelper.getPartitioner(context.getConfiguration()),
ConfigHelper.getInitialAddress(context.getConfiguration()),
ConfigHelper.getRpcPort(context.getConfiguration()));
this.batchThreshold = context.getConfiguration().getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, Long.MAX_VALUE);
this(context.getConfiguration());
}
ColumnFamilyRecordWriter(Configuration conf) throws IOException
{
this.conf = conf;
this.mutationsByEndpoint = new HashMap<InetAddress,Map<byte[],Map<String,List<Mutation>>>>();
this.ringCache = new RingCache(ConfigHelper.getOutputKeyspace(conf),
ConfigHelper.getPartitioner(conf),
ConfigHelper.getInitialAddress(conf),
ConfigHelper.getRpcPort(conf));
this.batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, Long.MAX_VALUE);
}
/**
* Return the endpoint responsible for the given key. The selected endpoint
* one whose token range contains the given key.
@ -145,7 +152,7 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
* @throws IOException
*/
@Override
public synchronized void write(ByteBuffer keybuff, List<org.apache.cassandra.avro.Mutation> value) throws IOException, InterruptedException
public synchronized void write(ByteBuffer keybuff, List<org.apache.cassandra.avro.Mutation> value) throws IOException
{
maybeFlush();
byte[] key = copy(keybuff);
@ -164,11 +171,11 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
mutationsByKey.put(key, cfMutation);
}
List<Mutation> mutationList = cfMutation.get(ConfigHelper.getOutputColumnFamily(context.getConfiguration()));
List<Mutation> mutationList = cfMutation.get(ConfigHelper.getOutputColumnFamily(conf));
if (mutationList == null)
{
mutationList = new ArrayList<Mutation>();
cfMutation.put(ConfigHelper.getOutputColumnFamily(context.getConfiguration()), mutationList);
cfMutation.put(ConfigHelper.getOutputColumnFamily(conf), mutationList);
}
for (org.apache.cassandra.avro.Mutation amut : value)
@ -254,6 +261,13 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
flush();
}
/** Fills the deprecated RecordWriter interface for streaming. */
@Deprecated @Override
public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException
{
flush();
}
/**
* Flush the mutations cache, iff more mutations have been cached than
* {@link #batchThreshold}.
@ -284,7 +298,7 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
List<Future<?>> mutationFutures = new ArrayList<Future<?>>();
for (Map.Entry<InetAddress, Map<byte[], Map<String, List<Mutation>>>> entry : mutationsByEndpoint.entrySet())
{
mutationFutures.add(executor.submit(new EndpointCallable(context, entry.getKey(), entry.getValue())));
mutationFutures.add(executor.submit(new EndpointCallable(conf, entry.getKey(), entry.getValue())));
}
// wait until we have all the results back
for (Future<?> mutationFuture : mutationFutures)
@ -321,7 +335,7 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
public class EndpointCallable implements Callable<Void>
{
// The task attempt context associated with this callable.
private TaskAttemptContext taskContext;
private Configuration conf;
// The endpoint of the primary replica for the rows being mutated
private InetAddress endpoint;
// The mutations to be performed in the node referenced by {@link
@ -332,13 +346,14 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
* Constructs an {@link EndpointCallable} for the given endpoint and set
* of mutations.
*
* @param conf job configuration
* @param endpoint the endpoint wherein to execute the mutations
* @param mutations the mutation map expected by
* {@link Cassandra.Client#batch_mutate(Map, ConsistencyLevel)}
*/
public EndpointCallable(TaskAttemptContext taskContext, InetAddress endpoint, Map<byte[], Map<String, List<Mutation>>> mutations)
public EndpointCallable(Configuration conf, InetAddress endpoint, Map<byte[], Map<String, List<Mutation>>> mutations)
{
this.taskContext = taskContext;
this.conf = conf;
this.endpoint = endpoint;
this.mutations = mutations;
}
@ -352,8 +367,8 @@ final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer,List<org.ap
TSocket socket = null;
try
{
socket = new TSocket(endpoint.getHostName(), ConfigHelper.getRpcPort(taskContext.getConfiguration()));
Cassandra.Client client = ColumnFamilyOutputFormat.createAuthenticatedClient(socket, taskContext);
socket = new TSocket(endpoint.getHostName(), ConfigHelper.getRpcPort(conf));
Cassandra.Client client = ColumnFamilyOutputFormat.createAuthenticatedClient(socket, conf);
client.batch_mutate(mutations, ConsistencyLevel.ONE);
return null;
}