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

rename

This commit is contained in:
Jonathan Ellis 2013-06-06 17:00:08 -05:00
parent ff5b830443
commit a004779763
7 changed files with 38 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
examples/hadoop_cql3_word_count/src/WordCount.java
View File

@ -21,13 +21,12 @@ import java.nio.ByteBuffer;
import java.util.*;
import java.util.Map.Entry;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.hadoop.cql3.ColumnFamilyOutputFormat;
import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
import org.apache.cassandra.hadoop.cql3.CqlOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.hadoop.cql3.CQLConfigHelper;
import org.apache.cassandra.hadoop.cql3.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.hadoop.conf.Configuration;
@ -38,7 +37,6 @@ import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
@ -208,28 +206,28 @@ public class WordCount extends Configured implements Tool
job.setOutputKeyClass(Map.class);
job.setOutputValueClass(List.class);
job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
job.setOutputFormatClass(CqlOutputFormat.class);
ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE, OUTPUT_COLUMN_FAMILY);
job.getConfiguration().set(PRIMARY_KEY, "word,sum");
String query = "INSERT INTO " + KEYSPACE + "." + OUTPUT_COLUMN_FAMILY +
" (row_id1, row_id2, word, count_num) " +
" values (?, ?, ?, ?)";
CQLConfigHelper.setOutputCql(job.getConfiguration(), query);
CqlConfigHelper.setOutputCql(job.getConfiguration(), query);
ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
}
job.setInputFormatClass(ColumnFamilyInputFormat.class);
job.setInputFormatClass(CqlPagingInputFormat.class);
ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
CQLConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), "3");
CqlConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), "3");
//this is the user defined filter clauses, you can comment it out if you want count all titles
CQLConfigHelper.setInputWhereClauses(job.getConfiguration(), "title='A'");
CqlConfigHelper.setInputWhereClauses(job.getConfiguration(), "title='A'");
job.waitForCompletion(true);
return 0;
}

11
examples/hadoop_cql3_word_count/src/WordCountCounters.java
View File

@ -24,23 +24,20 @@ import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
import org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.cassandra.hadoop.cql3.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.cql3.CQLConfigHelper;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -107,14 +104,14 @@ public class WordCountCounters extends Configured implements Tool
job.setOutputValueClass(LongWritable.class);
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX));
job.setInputFormatClass(ColumnFamilyInputFormat.class);
job.setInputFormatClass(CqlPagingInputFormat.class);
ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
ConfigHelper.setInputColumnFamily(job.getConfiguration(), WordCount.KEYSPACE, WordCount.OUTPUT_COLUMN_FAMILY);
CQLConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), "3");
CqlConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), "3");
job.waitForCompletion(true);
return 0;

2
src/java/org/apache/cassandra/hadoop/cql3/CQLConfigHelper.java
View File

@ -21,7 +21,7 @@ package org.apache.cassandra.hadoop.cql3;
*/
import org.apache.hadoop.conf.Configuration;
public class CQLConfigHelper
public class CqlConfigHelper
{
private static final String INPUT_CQL_COLUMNS_CONFIG = "cassandra.input.columnfamily.columns"; // separate by colon ,
private static final String INPUT_CQL_PAGE_ROW_SIZE_CONFIG = "cassandra.input.page.row.size";

12
src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyOutputFormat.java → src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java
View File

@ -38,7 +38,7 @@ import org.apache.hadoop.mapreduce.*;
* <p>
* As is the case with the {@link ColumnFamilyInputFormat}, you need to set the
* prepared statement in your
* Hadoop job Configuration. The {@link CQLConfigHelper} class, through its
* Hadoop job Configuration. The {@link CqlConfigHelper} class, through its
* {@link ConfigHelper#setOutputPreparedStatement} method, is provided to make this
* simple.
* you need to set the Keyspace. The {@link ConfigHelper} class, through its
@ -54,13 +54,13 @@ import org.apache.hadoop.mapreduce.*;
* to Cassandra.
* </p>
*/
public class ColumnFamilyOutputFormat extends AbstractColumnFamilyOutputFormat<Map<String, ByteBuffer>, List<ByteBuffer>>
public class CqlOutputFormat extends AbstractColumnFamilyOutputFormat<Map<String, ByteBuffer>, List<ByteBuffer>>
{
/** Fills the deprecated OutputFormat interface for streaming. */
@Deprecated
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
public CqlRecordWriter 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, new Progressable(progress));
return new CqlRecordWriter(job, new Progressable(progress));
}
/**
@ -71,8 +71,8 @@ public class ColumnFamilyOutputFormat extends AbstractColumnFamilyOutputFormat<M
* @return a {@link RecordWriter} to write the output for the job.
* @throws IOException
*/
public ColumnFamilyRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
public CqlRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
{
return new ColumnFamilyRecordWriter(context);
return new CqlRecordWriter(context);
}
}

6
src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyInputFormat.java → src/java/org/apache/cassandra/hadoop/cql3/CqlPagingInputFormat.java
View File

@ -53,7 +53,7 @@ import org.apache.hadoop.mapreduce.TaskAttemptID;
* the user defined the where clause
* CQLConfigHelper.setInputWhereClauses. The default is no user defined where clause
*/
public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat<Map<String, ByteBuffer>, Map<String, ByteBuffer>>
public class CqlPagingInputFormat extends AbstractColumnFamilyInputFormat<Map<String, ByteBuffer>, Map<String, ByteBuffer>>
{
public RecordReader<Map<String, ByteBuffer>, Map<String, ByteBuffer>> getRecordReader(InputSplit split, JobConf jobConf, final Reporter reporter)
throws IOException
@ -67,7 +67,7 @@ public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat<Map
}
};
ColumnFamilyRecordReader recordReader = new ColumnFamilyRecordReader();
CqlPagingRecordReader recordReader = new CqlPagingRecordReader();
recordReader.initialize((org.apache.hadoop.mapreduce.InputSplit)split, tac);
return recordReader;
}
@ -77,7 +77,7 @@ public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat<Map
org.apache.hadoop.mapreduce.InputSplit arg0, TaskAttemptContext arg1) throws IOException,
InterruptedException
{
return new ColumnFamilyRecordReader();
return new CqlPagingRecordReader();
}
}

14
src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordReader.java → src/java/org/apache/cassandra/hadoop/cql3/CqlPagingRecordReader.java
View File

@ -57,10 +57,10 @@ import org.apache.thrift.transport.TTransport;
* <p/>
* Map<ByteBuffer, IColumn> as column name to columns mappings
*/
public class ColumnFamilyRecordReader extends RecordReader<Map<String, ByteBuffer>, Map<String, ByteBuffer>>
public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>, Map<String, ByteBuffer>>
implements org.apache.hadoop.mapred.RecordReader<Map<String, ByteBuffer>, Map<String, ByteBuffer>>
{
private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyRecordReader.class);
private static final Logger logger = LoggerFactory.getLogger(CqlPagingRecordReader.class);
public static final int DEFAULT_CQL_PAGE_LIMIT = 1000; // TODO: find the number large enough but not OOM
@ -96,7 +96,7 @@ public class ColumnFamilyRecordReader extends RecordReader<Map<String, ByteBuffe
private AbstractType<?> keyValidator;
public ColumnFamilyRecordReader()
public CqlPagingRecordReader()
{
super();
}
@ -111,12 +111,12 @@ public class ColumnFamilyRecordReader extends RecordReader<Map<String, ByteBuffe
cfName = ConfigHelper.getInputColumnFamily(conf);
consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getReadConsistencyLevel(conf));
keyspace = ConfigHelper.getInputKeyspace(conf);
columns = CQLConfigHelper.getInputcolumns(conf);
userDefinedWhereClauses = CQLConfigHelper.getInputWhereClauses(conf);
columns = CqlConfigHelper.getInputcolumns(conf);
userDefinedWhereClauses = CqlConfigHelper.getInputWhereClauses(conf);
try
{
pageRowSize = Integer.parseInt(CQLConfigHelper.getInputPageRowSize(conf));
pageRowSize = Integer.parseInt(CqlConfigHelper.getInputPageRowSize(conf));
}
catch (NumberFormatException e)
{
@ -134,7 +134,7 @@ public class ColumnFamilyRecordReader extends RecordReader<Map<String, ByteBuffe
String location = getLocation();
int port = ConfigHelper.getInputRpcPort(conf);
client = ColumnFamilyInputFormat.createAuthenticatedClient(location, port, conf);
client = CqlPagingInputFormat.createAuthenticatedClient(location, port, conf);
// retrieve partition keys and cluster keys from system.schema_columnfamilies table
retrieveKeys();

18
src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordWriter.java → src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
View File

@ -60,11 +60,11 @@ import org.slf4j.LoggerFactory;
* directly to a responsible endpoint.
* </p>
*
* @see ColumnFamilyOutputFormat
* @see CqlOutputFormat
*/
final class ColumnFamilyRecordWriter extends AbstractColumnFamilyRecordWriter<Map<String, ByteBuffer>, List<ByteBuffer>>
final class CqlRecordWriter extends AbstractColumnFamilyRecordWriter<Map<String, ByteBuffer>, List<ByteBuffer>>
{
private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyRecordWriter.class);
private static final Logger logger = LoggerFactory.getLogger(CqlRecordWriter.class);
// handles for clients for each range running in the threadpool
private final Map<Range, RangeClient> clients;
@ -84,29 +84,29 @@ final class ColumnFamilyRecordWriter extends AbstractColumnFamilyRecordWriter<Ma
* @param context the task attempt context
* @throws IOException
*/
ColumnFamilyRecordWriter(TaskAttemptContext context) throws IOException
CqlRecordWriter(TaskAttemptContext context) throws IOException
{
this(context.getConfiguration());
this.progressable = new Progressable(context);
}
ColumnFamilyRecordWriter(Configuration conf, Progressable progressable) throws IOException
CqlRecordWriter(Configuration conf, Progressable progressable) throws IOException
{
this(conf);
this.progressable = progressable;
}
ColumnFamilyRecordWriter(Configuration conf) throws IOException
CqlRecordWriter(Configuration conf) throws IOException
{
super(conf);
this.clients = new HashMap<Range, RangeClient>();
cql = CQLConfigHelper.getOutputCql(conf);
cql = CqlConfigHelper.getOutputCql(conf);
try
{
String host = getAnyHost();
int port = ConfigHelper.getOutputRpcPort(conf);
Cassandra.Client client = ColumnFamilyOutputFormat.createAuthenticatedClient(host, port, conf);
Cassandra.Client client = CqlOutputFormat.createAuthenticatedClient(host, port, conf);
retrievePartitionKeyValidator(client);
if (client != null)
@ -250,7 +250,7 @@ final class ColumnFamilyRecordWriter extends AbstractColumnFamilyRecordWriter<Ma
InetAddress address = iter.next();
String host = address.getHostName();
int port = ConfigHelper.getOutputRpcPort(conf);
client = ColumnFamilyOutputFormat.createAuthenticatedClient(host, port, conf);
client = CqlOutputFormat.createAuthenticatedClient(host, port, conf);
}
catch (Exception e)
{