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

fix bulk loader build. (untested, but compiling is an improvement over not compiling). patch by jbellis 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/branches/cassandra-0.6@916910 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-02-27 04:59:27 +00:00
parent 4f8e363bd4
commit 80abcc2f53
2 changed files with 23 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
contrib/bmt_example/CassandraBulkLoader.java
View File

@ -63,7 +63,6 @@ import java.net.UnknownHostException;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.SelectorManager;
import org.apache.cassandra.service.StorageService;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
@ -102,7 +101,14 @@ public class CassandraBulkLoader {
System.setProperty("storage-config",cassConfig);
StorageService.instance().startClient();
try
{
StorageService.instance.initClient();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
try
{
Thread.sleep(10*1000);
@ -137,21 +143,21 @@ public class CassandraBulkLoader {
{
throw new RuntimeException(e);
}
StorageService.instance().stopClient();
StorageService.instance.stopClient();
}
public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException
{
ColumnFamily columnFamily;
String Keyspace = "Keyspace1";
String CFName = "Super1";
String keyspace = "Keyspace1";
String cfName = "Super1";
Message message;
List<ColumnFamily> columnFamilies;
columnFamilies = new LinkedList<ColumnFamily>();
String line;
/* Create a column family */
columnFamily = ColumnFamily.create(Keyspace, CFName);
columnFamily = ColumnFamily.create(keyspace, cfName);
while (values.hasNext()) {
// Split the value (line based on your own delimiter)
line = values.next().toString();
@ -160,17 +166,17 @@ public class CassandraBulkLoader {
String ColumnName = fields[2];
String ColumnValue = fields[3];
int timestamp = 0;
columnFamily.addColumn(new QueryPath(CFName, SuperColumnName.getBytes("UTF-8"), ColumnName.getBytes("UTF-8")), ColumnValue.getBytes(), timestamp);
columnFamily.addColumn(new QueryPath(cfName, SuperColumnName.getBytes("UTF-8"), ColumnName.getBytes("UTF-8")), ColumnValue.getBytes(), timestamp);
}
columnFamilies.add(columnFamily);
/* Get serialized message to send to cluster */
message = createMessage(Keyspace, key.toString(), CFName, columnFamilies);
for (InetAddress endpoint: StorageService.instance().getNaturalEndPoints(key.toString()))
message = createMessage(keyspace, key.toString(), cfName, columnFamilies);
for (InetAddress endpoint: StorageService.instance.getNaturalEndpoints(keyspace, key.toString()))
{
/* Send message to end point */
MessagingService.instance().sendOneWay(message, endpoint);
MessagingService.instance.sendOneWay(message, endpoint);
}
output.collect(key, new Text(" inserted into Cassandra node(s)"));
@ -248,7 +254,7 @@ public class CassandraBulkLoader {
try
{
/* Make message */
message = rm.makeRowMutationMessage(StorageService.binaryVerbHandler_);
message = rm.makeRowMutationMessage(StorageService.Verb.BINARY);
}
catch (IOException e)
{

7
src/java/org/apache/cassandra/db/RowMutation.java
View File

@ -210,11 +210,16 @@ public class RowMutation
}
public Message makeRowMutationMessage() throws IOException
{
return makeRowMutationMessage(StorageService.Verb.MUTATION);
}
public Message makeRowMutationMessage(StorageService.Verb verb) throws IOException
{
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(bos);
serializer().serialize(this, dos);
return new Message(FBUtilities.getLocalAddress(), StageManager.MUTATION_STAGE, StorageService.Verb.MUTATION, bos.toByteArray());
return new Message(FBUtilities.getLocalAddress(), StageManager.MUTATION_STAGE, verb, bos.toByteArray());
}
public static RowMutation getRowMutationFromMutations(String keyspace, String key, Map<String, List<Mutation>> cfmap)