From 90052d5a460efefd2dca7a0a5c5fbd54c9674d75 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 6 Jun 2013 14:16:28 -0500 Subject: [PATCH] merge from 1.2 --- examples/hadoop_cql3_word_count/README.txt | 47 ++ .../hadoop_cql3_word_count/bin/word_count | 61 ++ .../bin/word_count_counters | 58 ++ .../bin/word_count_setup | 61 ++ examples/hadoop_cql3_word_count/build.xml | 113 +++ .../conf/log4j.properties | 32 + examples/hadoop_cql3_word_count/ivy.xml | 24 + .../hadoop_cql3_word_count/src/WordCount.java | 236 ++++++ .../src/WordCountCounters.java | 122 +++ .../src/WordCountSetup.java | 213 +++++ .../AbstractColumnFamilyInputFormat.java | 346 ++++++++ .../AbstractColumnFamilyOutputFormat.java | 159 ++++ .../AbstractColumnFamilyRecordWriter.java | 193 +++++ .../hadoop/ColumnFamilyInputFormat.java | 307 +------ .../hadoop/ColumnFamilyOutputFormat.java | 122 +-- .../hadoop/ColumnFamilyRecordReader.java | 58 +- .../hadoop/ColumnFamilyRecordWriter.java | 190 +---- .../cassandra/hadoop/ColumnFamilySplit.java | 4 +- .../apache/cassandra/hadoop/ConfigHelper.java | 59 +- .../apache/cassandra/hadoop/Progressable.java | 4 +- .../hadoop/cql3/CQLConfigHelper.java | 109 +++ .../hadoop/cql3/ColumnFamilyInputFormat.java | 83 ++ .../hadoop/cql3/ColumnFamilyOutputFormat.java | 78 ++ .../hadoop/cql3/ColumnFamilyRecordReader.java | 763 ++++++++++++++++++ .../hadoop/cql3/ColumnFamilyRecordWriter.java | 386 +++++++++ .../thrift/TClientTransportFactory.java | 70 ++ .../thrift/TFramedTransportFactory.java | 24 +- 27 files changed, 3271 insertions(+), 651 deletions(-) create mode 100644 examples/hadoop_cql3_word_count/README.txt create mode 100644 examples/hadoop_cql3_word_count/bin/word_count create mode 100644 examples/hadoop_cql3_word_count/bin/word_count_counters create mode 100644 examples/hadoop_cql3_word_count/bin/word_count_setup create mode 100644 examples/hadoop_cql3_word_count/build.xml create mode 100644 examples/hadoop_cql3_word_count/conf/log4j.properties create mode 100644 examples/hadoop_cql3_word_count/ivy.xml create mode 100644 examples/hadoop_cql3_word_count/src/WordCount.java create mode 100644 examples/hadoop_cql3_word_count/src/WordCountCounters.java create mode 100644 examples/hadoop_cql3_word_count/src/WordCountSetup.java create mode 100644 src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java create mode 100644 src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyOutputFormat.java create mode 100644 src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyRecordWriter.java create mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CQLConfigHelper.java create mode 100644 src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyInputFormat.java create mode 100644 src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyOutputFormat.java create mode 100644 src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordReader.java create mode 100644 src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordWriter.java create mode 100644 src/java/org/apache/cassandra/thrift/TClientTransportFactory.java diff --git a/examples/hadoop_cql3_word_count/README.txt b/examples/hadoop_cql3_word_count/README.txt new file mode 100644 index 0000000000..f984b0294f --- /dev/null +++ b/examples/hadoop_cql3_word_count/README.txt @@ -0,0 +1,47 @@ +Introduction +============ + +WordCount hadoop example: Inserts a bunch of words across multiple rows, +and counts them, with RandomPartitioner. The word_count_counters example sums +the value of counter columns for a key. + +The scripts in bin/ assume you are running with cwd of contrib/word_count. + + +Running +======= + +First build and start a Cassandra server with the default configuration*, +then run + +contrib/word_count$ ant +contrib/word_count$ bin/word_count_setup +contrib/word_count$ bin/word_count +contrib/word_count$ bin/word_count_counters + +In order to view the results in Cassandra, one can use bin/cassandra-cli and +perform the following operations: +$ bin/cqlsh localhost +> use cql3_worldcount; +> select * from output_words; + +The output of the word count can now be configured. In the bin/word_count +file, you can specify the OUTPUT_REDUCER. The two options are 'filesystem' +and 'cassandra'. The filesystem option outputs to the /tmp/word_count* +directories. The cassandra option outputs to the 'output_words' column family +in the 'cql3_worldcount' keyspace. 'cassandra' is the default. + +Read the code in src/ for more details. + +The word_count_counters example sums the counter columns for a row. The output +is written to a text file in /tmp/word_count_counters. + +*If you want to point wordcount at a real cluster, modify the seed +and listenaddress settings accordingly. + + +Troubleshooting +=============== + +word_count uses conf/log4j.properties to log to wc.out. + diff --git a/examples/hadoop_cql3_word_count/bin/word_count b/examples/hadoop_cql3_word_count/bin/word_count new file mode 100644 index 0000000000..a0c5aa02f1 --- /dev/null +++ b/examples/hadoop_cql3_word_count/bin/word_count @@ -0,0 +1,61 @@ +#!/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` + +# Cassandra class files. +if [ ! -d $cwd/../../../build/classes/main ]; then + echo "Unable to locate cassandra class files" >&2 + exit 1 +fi + +# word_count Jar. +if [ ! -e $cwd/../build/word_count.jar ]; then + echo "Unable to locate word_count jar" >&2 + exit 1 +fi + +CLASSPATH=$CLASSPATH:$cwd/../conf +CLASSPATH=$CLASSPATH:$cwd/../build/word_count.jar +CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/main +CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/thrift +for jar in $cwd/../build/lib/jars/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done +for jar in $cwd/../../../lib/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done +for jar in $cwd/../../../build/lib/jars/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done + +if [ -x $JAVA_HOME/bin/java ]; then + JAVA=$JAVA_HOME/bin/java +else + JAVA=`which java` +fi + +if [ "x$JAVA" = "x" ]; then + echo "Java executable not found (hint: set JAVA_HOME)" >&2 + exit 1 +fi + +OUTPUT_REDUCER=cassandra + +#echo $CLASSPATH +$JAVA -Xmx1G -ea -cp $CLASSPATH WordCount output_reducer=$OUTPUT_REDUCER diff --git a/examples/hadoop_cql3_word_count/bin/word_count_counters b/examples/hadoop_cql3_word_count/bin/word_count_counters new file mode 100644 index 0000000000..779347743f --- /dev/null +++ b/examples/hadoop_cql3_word_count/bin/word_count_counters @@ -0,0 +1,58 @@ +#!/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` + +# Cassandra class files. +if [ ! -d $cwd/../../../build/classes/main ]; then + echo "Unable to locate cassandra class files" >&2 + exit 1 +fi + +# word_count Jar. +if [ ! -e $cwd/../build/word_count.jar ]; then + echo "Unable to locate word_count jar" >&2 + exit 1 +fi + +CLASSPATH=$CLASSPATH:$cwd/../build/word_count.jar +CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/main +CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/thrift +for jar in $cwd/../build/lib/jars/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done +for jar in $cwd/../../../lib/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done +for jar in $cwd/../../../build/lib/jars/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done + +if [ -x $JAVA_HOME/bin/java ]; then + JAVA=$JAVA_HOME/bin/java +else + JAVA=`which java` +fi + +if [ "x$JAVA" = "x" ]; then + echo "Java executable not found (hint: set JAVA_HOME)" >&2 + exit 1 +fi + +#echo $CLASSPATH +$JAVA -Xmx1G -ea -cp $CLASSPATH WordCountCounters diff --git a/examples/hadoop_cql3_word_count/bin/word_count_setup b/examples/hadoop_cql3_word_count/bin/word_count_setup new file mode 100644 index 0000000000..d194a452b4 --- /dev/null +++ b/examples/hadoop_cql3_word_count/bin/word_count_setup @@ -0,0 +1,61 @@ +#!/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` + +# Cassandra class files. +if [ ! -d $cwd/../../../build/classes/main ]; then + echo "Unable to locate cassandra class files" >&2 + exit 1 +fi + +# word_count Jar. +if [ ! -e $cwd/../build/word_count.jar ]; then + echo "Unable to locate word_count jar" >&2 + exit 1 +fi + +CLASSPATH=$CLASSPATH:$cwd/../build/word_count.jar +CLASSPATH=$CLASSPATH:.:$cwd/../../../build/classes/main +CLASSPATH=$CLASSPATH:.:$cwd/../../../build/classes/thrift +for jar in $cwd/../build/lib/jars/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done +for jar in $cwd/../../../lib/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done +for jar in $cwd/../../../build/lib/jars/*.jar; do + CLASSPATH=$CLASSPATH:$jar +done + +if [ -x $JAVA_HOME/bin/java ]; then + JAVA=$JAVA_HOME/bin/java +else + JAVA=`which java` +fi + +if [ "x$JAVA" = "x" ]; then + echo "Java executable not found (hint: set JAVA_HOME)" >&2 + exit 1 +fi + +HOST=localhost +PORT=9160 +FRAMED=true + +$JAVA -Xmx1G -ea -Dcassandra.host=$HOST -Dcassandra.port=$PORT -Dcassandra.framed=$FRAMED -cp $CLASSPATH WordCountSetup diff --git a/examples/hadoop_cql3_word_count/build.xml b/examples/hadoop_cql3_word_count/build.xml new file mode 100644 index 0000000000..939e1b39d0 --- /dev/null +++ b/examples/hadoop_cql3_word_count/build.xml @@ -0,0 +1,113 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Downloading Ivy... + + + + + + + + + + + + + + diff --git a/examples/hadoop_cql3_word_count/conf/log4j.properties b/examples/hadoop_cql3_word_count/conf/log4j.properties new file mode 100644 index 0000000000..70f7657591 --- /dev/null +++ b/examples/hadoop_cql3_word_count/conf/log4j.properties @@ -0,0 +1,32 @@ +# 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. + +log4j.rootLogger=DEBUG,stdout,F + +#stdout +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%5p %d{HH:mm:ss,SSS} %m%n + +# log file +log4j.appender.F=org.apache.log4j.FileAppender +log4j.appender.F.Append=false +log4j.appender.F.layout=org.apache.log4j.PatternLayout +log4j.appender.F.layout.ConversionPattern=%5p [%t] %d{ISO8601} %F (line %L) %m%n +# Edit the next line to point to your logs directory +log4j.appender.F.File=wc.out + diff --git a/examples/hadoop_cql3_word_count/ivy.xml b/examples/hadoop_cql3_word_count/ivy.xml new file mode 100644 index 0000000000..9d448959be --- /dev/null +++ b/examples/hadoop_cql3_word_count/ivy.xml @@ -0,0 +1,24 @@ + + + + + + + \ No newline at end of file diff --git a/examples/hadoop_cql3_word_count/src/WordCount.java b/examples/hadoop_cql3_word_count/src/WordCount.java new file mode 100644 index 0000000000..09dd9e47ec --- /dev/null +++ b/examples/hadoop_cql3_word_count/src/WordCount.java @@ -0,0 +1,236 @@ +/** + * 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. + */ + +import java.io.IOException; +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.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.ConfigHelper; +import org.apache.cassandra.utils.ByteBufferUtil; +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.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; + +import java.nio.charset.CharacterCodingException; + +/** + * This counts the occurrences of words in ColumnFamily + * cql3_worldcount ( user_id text, + * category_id text, + * sub_category_id text, + * title text, + * body text, + * PRIMARY KEY (user_id, category_id, sub_category_id)) + * + * For each word, we output the total number of occurrences across all body texts. + * + * When outputting to Cassandra, we write the word counts to column family + * output_words ( row_id1 text, + * row_id2 text, + * word text, + * count_num text, + * PRIMARY KEY ((row_id1, row_id2), word)) + * as a {word, count} to columns: word, count_num with a row key of "word sum" + */ +public class WordCount extends Configured implements Tool +{ + private static final Logger logger = LoggerFactory.getLogger(WordCount.class); + + static final String KEYSPACE = "cql3_worldcount"; + static final String COLUMN_FAMILY = "inputs"; + + static final String OUTPUT_REDUCER_VAR = "output_reducer"; + static final String OUTPUT_COLUMN_FAMILY = "output_words"; + + private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count"; + + private static final String PRIMARY_KEY = "row_key"; + + public static void main(String[] args) throws Exception + { + // Let ToolRunner handle generic command-line options + ToolRunner.run(new Configuration(), new WordCount(), args); + System.exit(0); + } + + public static class TokenizerMapper extends Mapper, Map, Text, IntWritable> + { + private final static IntWritable one = new IntWritable(1); + private Text word = new Text(); + private ByteBuffer sourceColumn; + + protected void setup(org.apache.hadoop.mapreduce.Mapper.Context context) + throws IOException, InterruptedException + { + } + + public void map(Map keys, Map columns, Context context) throws IOException, InterruptedException + { + for (Entry column : columns.entrySet()) + { + if (!"body".equalsIgnoreCase(column.getKey())) + continue; + + String value = ByteBufferUtil.string(column.getValue()); + + logger.debug("read {}:{}={} from {}", + new Object[] {toString(keys), column.getKey(), value, context.getInputSplit()}); + + StringTokenizer itr = new StringTokenizer(value); + while (itr.hasMoreTokens()) + { + word.set(itr.nextToken()); + context.write(word, one); + } + } + } + + private String toString(Map keys) + { + String result = ""; + try + { + for (ByteBuffer key : keys.values()) + result = result + ByteBufferUtil.string(key) + ":"; + } + catch (CharacterCodingException e) + { + logger.error("Failed to print keys", e); + } + return result; + } + } + + public static class ReducerToFilesystem extends Reducer + { + public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException + { + int sum = 0; + for (IntWritable val : values) + sum += val.get(); + context.write(key, new IntWritable(sum)); + } + } + + public static class ReducerToCassandra extends Reducer, List> + { + private Map keys; + private ByteBuffer key; + protected void setup(org.apache.hadoop.mapreduce.Reducer.Context context) + throws IOException, InterruptedException + { + keys = new LinkedHashMap(); + String[] partitionKeys = context.getConfiguration().get(PRIMARY_KEY).split(","); + keys.put("row_id1", ByteBufferUtil.bytes(partitionKeys[0])); + keys.put("row_id2", ByteBufferUtil.bytes(partitionKeys[1])); + } + + public void reduce(Text word, Iterable values, Context context) throws IOException, InterruptedException + { + int sum = 0; + for (IntWritable val : values) + sum += val.get(); + context.write(keys, getBindVariables(word, sum)); + } + + private List getBindVariables(Text word, int sum) + { + List variables = new ArrayList(); + variables.add(keys.get("row_id1")); + variables.add(keys.get("row_id2")); + variables.add(ByteBufferUtil.bytes(word.toString())); + variables.add(ByteBufferUtil.bytes(String.valueOf(sum))); + return variables; + } + } + + public int run(String[] args) throws Exception + { + String outputReducerType = "filesystem"; + if (args != null && args[0].startsWith(OUTPUT_REDUCER_VAR)) + { + String[] s = args[0].split("="); + if (s != null && s.length == 2) + outputReducerType = s[1]; + } + logger.info("output reducer type: " + outputReducerType); + + Job job = new Job(getConf(), "wordcount"); + job.setJarByClass(WordCount.class); + job.setMapperClass(TokenizerMapper.class); + + if (outputReducerType.equalsIgnoreCase("filesystem")) + { + job.setCombinerClass(ReducerToFilesystem.class); + job.setReducerClass(ReducerToFilesystem.class); + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX)); + } + else + { + job.setReducerClass(ReducerToCassandra.class); + + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(IntWritable.class); + job.setOutputKeyClass(Map.class); + job.setOutputValueClass(List.class); + + job.setOutputFormatClass(ColumnFamilyOutputFormat.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); + ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost"); + ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner"); + } + + job.setInputFormatClass(ColumnFamilyInputFormat.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"); + //this is the user defined filter clauses, you can comment it out if you want count all titles + CQLConfigHelper.setInputWhereClauses(job.getConfiguration(), "title='A'"); + job.waitForCompletion(true); + return 0; + } +} \ No newline at end of file diff --git a/examples/hadoop_cql3_word_count/src/WordCountCounters.java b/examples/hadoop_cql3_word_count/src/WordCountCounters.java new file mode 100644 index 0000000000..1cf5539460 --- /dev/null +++ b/examples/hadoop_cql3_word_count/src/WordCountCounters.java @@ -0,0 +1,122 @@ +/** + * 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. + */ + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.util.*; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +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; + + +/** + * This sums the word count stored in the input_words_count ColumnFamily for the key "sum". + * + * Output is written to a text file. + */ +public class WordCountCounters extends Configured implements Tool +{ + private static final Logger logger = LoggerFactory.getLogger(WordCountCounters.class); + + static final String COUNTER_COLUMN_FAMILY = "input_words_count"; + private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count_counters"; + + public static void main(String[] args) throws Exception + { + // Let ToolRunner handle generic command-line options + ToolRunner.run(new Configuration(), new WordCountCounters(), args); + System.exit(0); + } + + public static class SumMapper extends Mapper, Map, Text, LongWritable> + { + long sum = -1; + public void map(Map key, Map columns, Context context) throws IOException, InterruptedException + { + if (sum < 0) + sum = 0; + + logger.debug("read " + toString(key) + ":count_num from " + context.getInputSplit()); + sum += Long.valueOf(ByteBufferUtil.string(columns.get("count_num"))); + } + + protected void cleanup(Context context) throws IOException, InterruptedException { + if (sum > 0) + context.write(new Text("total_count"), new LongWritable(sum)); + } + + private String toString(Map keys) + { + String result = ""; + try + { + for (ByteBuffer key : keys.values()) + result = result + ByteBufferUtil.string(key) + ":"; + } + catch (CharacterCodingException e) + { + logger.error("Failed to print keys", e); + } + return result; + } + } + + + public int run(String[] args) throws Exception + { + Job job = new Job(getConf(), "wordcountcounters"); + job.setJarByClass(WordCountCounters.class); + job.setMapperClass(SumMapper.class); + + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(LongWritable.class); + FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX)); + + job.setInputFormatClass(ColumnFamilyInputFormat.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"); + + job.waitForCompletion(true); + return 0; + } +} diff --git a/examples/hadoop_cql3_word_count/src/WordCountSetup.java b/examples/hadoop_cql3_word_count/src/WordCountSetup.java new file mode 100644 index 0000000000..0acb8f79c3 --- /dev/null +++ b/examples/hadoop_cql3_word_count/src/WordCountSetup.java @@ -0,0 +1,213 @@ +/** + * 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. + */ + +import java.nio.ByteBuffer; +import java.util.*; + +import org.apache.cassandra.thrift.*; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.thrift.TException; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; +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 WordCountSetup +{ + private static final Logger logger = LoggerFactory.getLogger(WordCountSetup.class); + + public static final int TEST_COUNT = 6; + + public static void main(String[] args) throws Exception + { + Cassandra.Iface client = createConnection(); + + setupKeyspace(client); + client.set_keyspace(WordCount.KEYSPACE); + setupTable(client); + insertData(client); + + System.exit(0); + } + + private static void setupKeyspace(Cassandra.Iface client) + throws InvalidRequestException, + UnavailableException, + TimedOutException, + SchemaDisagreementException, + TException + { + KsDef ks; + try + { + ks = client.describe_keyspace(WordCount.KEYSPACE); + } + catch(NotFoundException e) + { + logger.info("set up keyspace " + WordCount.KEYSPACE); + String query = "CREATE KEYSPACE " + WordCount.KEYSPACE + + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 1}"; + + client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE); + + int magnitude = client.describe_ring(WordCount.KEYSPACE).size(); + try + { + Thread.sleep(1000 * magnitude); + } + catch (InterruptedException ie) + { + throw new RuntimeException(ie); + } + } + } + + private static void setupTable(Cassandra.Iface client) + throws InvalidRequestException, + UnavailableException, + TimedOutException, + SchemaDisagreementException, + TException + { + String query = "CREATE TABLE " + WordCount.KEYSPACE + "." + WordCount.COLUMN_FAMILY + + " ( user_id text," + + " category_id text, " + + " sub_category_id text," + + " title text," + + " body text," + + " PRIMARY KEY (user_id, category_id, sub_category_id) ) "; + + try + { + logger.info("set up table " + WordCount.COLUMN_FAMILY); + client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE); + } + catch (InvalidRequestException e) + { + logger.error("failed to create table " + WordCount.KEYSPACE + "." + WordCount.COLUMN_FAMILY, e); + } + + query = "CREATE INDEX title on " + WordCount.COLUMN_FAMILY + "(title)"; + try + { + logger.info("set up index on title column "); + client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE); + } + catch (InvalidRequestException e) + { + logger.error("Failed to create index on title", e); + } + + query = "CREATE TABLE " + WordCount.KEYSPACE + "." + WordCount.OUTPUT_COLUMN_FAMILY + + " ( row_id text," + + " word text, " + + " count_num text," + + " PRIMARY KEY (row_id, word) ) "; + + try + { + logger.info("set up table " + WordCount.OUTPUT_COLUMN_FAMILY); + client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE); + } + catch (InvalidRequestException e) + { + logger.error("failed to create table " + WordCount.KEYSPACE + "." + WordCount.OUTPUT_COLUMN_FAMILY, e); + } + } + + private static Cassandra.Iface createConnection() throws TTransportException + { + if (System.getProperty("cassandra.host") == null || System.getProperty("cassandra.port") == null) + { + logger.warn("cassandra.host or cassandra.port is not defined, using default"); + } + return createConnection(System.getProperty("cassandra.host", "localhost"), + Integer.valueOf(System.getProperty("cassandra.port", "9160"))); + } + + private static Cassandra.Client createConnection(String host, Integer port) throws TTransportException + { + TSocket socket = new TSocket(host, port); + TTransport trans = new TFramedTransport(socket); + trans.open(); + TProtocol protocol = new TBinaryProtocol(trans); + + return new Cassandra.Client(protocol); + } + + private static void insertData(Cassandra.Iface client) + throws InvalidRequestException, + UnavailableException, + TimedOutException, + SchemaDisagreementException, + TException + { + String query = "INSERT INTO " + WordCount.COLUMN_FAMILY + + "(user_id, category_id, sub_category_id, title, body ) " + + " values (?, ?, ?, ?, ?) "; + CqlPreparedResult result = client.prepare_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE); + + String [] title = titleData(); + String [] body = bodyData(); + for (int i=1; i<5; i++) + { + for (int j=1; j<444; j++) + { + for (int k=1; k<4; k++) + { + List values = new ArrayList(); + values.add(ByteBufferUtil.bytes(String.valueOf(j))); + values.add(ByteBufferUtil.bytes(String.valueOf(i))); + values.add(ByteBufferUtil.bytes(String.valueOf(k))); + values.add(ByteBufferUtil.bytes(title[i])); + values.add(ByteBufferUtil.bytes(body[i])); + client.execute_prepared_cql3_query(result.itemId, values, ConsistencyLevel.ONE); + } + } + } + } + + private static String[] bodyData() + { // Public domain context, source http://en.wikisource.org/wiki/If%E2%80%94 + return new String[]{ + "", + "If you can keep your head when all about you", + "Are losing theirs and blaming it on you", + "If you can trust yourself when all men doubt you,", + "But make allowance for their doubting too:", + "If you can wait and not be tired by waiting," + }; + } + + private static String[] titleData() + { // Public domain context, source http://en.wikisource.org/wiki/If%E2%80%94 + return new String[]{ + "", + "A", + "B", + "C", + "D", + "E" + }; + } +} diff --git a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java new file mode 100644 index 0000000000..1c8fd0b1ff --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java @@ -0,0 +1,346 @@ +/* + * 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; + +import java.io.IOException; +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import org.apache.cassandra.auth.IAuthenticator; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.thrift.AuthenticationRequest; +import org.apache.cassandra.thrift.Cassandra; +import org.apache.cassandra.thrift.CfSplit; +import org.apache.cassandra.thrift.InvalidRequestException; +import org.apache.cassandra.thrift.KeyRange; +import org.apache.cassandra.thrift.TokenRange; +import org.apache.commons.lang.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapreduce.InputFormat; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.JobContext; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.TaskAttemptID; +import org.apache.thrift.TApplicationException; +import org.apache.thrift.TException; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.transport.TTransport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; + +public abstract class AbstractColumnFamilyInputFormat extends InputFormat implements org.apache.hadoop.mapred.InputFormat +{ + private static final Logger logger = LoggerFactory.getLogger(AbstractColumnFamilyInputFormat.class); + + public static final String MAPRED_TASK_ID = "mapred.task.id"; + // The simple fact that we need this is because the old Hadoop API wants us to "write" + // to the key and value whereas the new asks for it. + // I choose 8kb as the default max key size (instanciated only once), but you can + // override it in your jobConf with this setting. + public static final String CASSANDRA_HADOOP_MAX_KEY_SIZE = "cassandra.hadoop.max_key_size"; + public static final int CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT = 8192; + + private String keyspace; + private String cfName; + private IPartitioner partitioner; + + protected void validateConfiguration(Configuration conf) + { + if (ConfigHelper.getInputKeyspace(conf) == null || ConfigHelper.getInputColumnFamily(conf) == null) + { + throw new UnsupportedOperationException("you must set the keyspace and columnfamily with setInputColumnFamily()"); + } + if (ConfigHelper.getInputInitialAddress(conf) == null) + throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node with setInputInitialAddress"); + if (ConfigHelper.getInputPartitioner(conf) == null) + throw new UnsupportedOperationException("You must set the Cassandra partitioner class with setInputPartitioner"); + } + + public static Cassandra.Client createAuthenticatedClient(String location, int port, Configuration conf) throws Exception + { + logger.debug("Creating authenticated client for CF input format"); + TTransport transport = ConfigHelper.getClientTransportFactory(conf).openTransport(location, port, conf); + TProtocol binaryProtocol = new TBinaryProtocol(transport, true, true); + Cassandra.Client client = new Cassandra.Client(binaryProtocol); + + // log in + client.set_keyspace(ConfigHelper.getInputKeyspace(conf)); + if (ConfigHelper.getInputKeyspaceUserName(conf) != null) + { + Map creds = new HashMap(); + creds.put(IAuthenticator.USERNAME_KEY, ConfigHelper.getInputKeyspaceUserName(conf)); + creds.put(IAuthenticator.PASSWORD_KEY, ConfigHelper.getInputKeyspacePassword(conf)); + AuthenticationRequest authRequest = new AuthenticationRequest(creds); + client.login(authRequest); + } + logger.debug("Authenticated client for CF input format created successfully"); + return client; + } + + public List getSplits(JobContext context) throws IOException + { + Configuration conf = context.getConfiguration(); + + validateConfiguration(conf); + + // cannonical ranges and nodes holding replicas + List masterRangeNodes = getRangeMap(conf); + + keyspace = ConfigHelper.getInputKeyspace(context.getConfiguration()); + cfName = ConfigHelper.getInputColumnFamily(context.getConfiguration()); + partitioner = ConfigHelper.getInputPartitioner(context.getConfiguration()); + logger.debug("partitioner is " + partitioner); + + // cannonical ranges, split into pieces, fetching the splits in parallel + ExecutorService executor = Executors.newCachedThreadPool(); + List splits = new ArrayList(); + + try + { + List>> splitfutures = new ArrayList>>(); + KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf); + Range jobRange = null; + if (jobKeyRange != null) + { + if (jobKeyRange.start_key == null) + { + logger.warn("ignoring jobKeyRange specified without start_key"); + } + else + { + if (!partitioner.preservesOrder()) + throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving paritioner"); + if (jobKeyRange.start_token != null) + throw new IllegalArgumentException("only start_key supported"); + if (jobKeyRange.end_token != null) + throw new IllegalArgumentException("only start_key supported"); + jobRange = new Range(partitioner.getToken(jobKeyRange.start_key), + partitioner.getToken(jobKeyRange.end_key), + partitioner); + } + } + + for (TokenRange range : masterRangeNodes) + { + if (jobRange == null) + { + // for each range, pick a live owner and ask it to compute bite-sized splits + splitfutures.add(executor.submit(new SplitCallable(range, conf))); + } + else + { + Range dhtRange = new Range(partitioner.getTokenFactory().fromString(range.start_token), + partitioner.getTokenFactory().fromString(range.end_token), + partitioner); + + if (dhtRange.intersects(jobRange)) + { + for (Range intersection: dhtRange.intersectionWith(jobRange)) + { + range.start_token = partitioner.getTokenFactory().toString(intersection.left); + range.end_token = partitioner.getTokenFactory().toString(intersection.right); + // for each range, pick a live owner and ask it to compute bite-sized splits + splitfutures.add(executor.submit(new SplitCallable(range, conf))); + } + } + } + } + + // wait until we have all the results back + for (Future> futureInputSplits : splitfutures) + { + try + { + splits.addAll(futureInputSplits.get()); + } + catch (Exception e) + { + throw new IOException("Could not get input splits", e); + } + } + } + finally + { + executor.shutdownNow(); + } + + assert splits.size() > 0; + Collections.shuffle(splits, new Random(System.nanoTime())); + return splits; + } + + /** + * Gets a token range and splits it up according to the suggested + * size into input splits that Hadoop can use. + */ + class SplitCallable implements Callable> + { + + private final TokenRange range; + private final Configuration conf; + + public SplitCallable(TokenRange tr, Configuration conf) + { + this.range = tr; + this.conf = conf; + } + + public List call() throws Exception + { + ArrayList splits = new ArrayList(); + List subSplits = getSubSplits(keyspace, cfName, range, conf); + assert range.rpc_endpoints.size() == range.endpoints.size() : "rpc_endpoints size must match endpoints size"; + // turn the sub-ranges into InputSplits + String[] endpoints = range.endpoints.toArray(new String[range.endpoints.size()]); + // hadoop needs hostname, not ip + int endpointIndex = 0; + for (String endpoint: range.rpc_endpoints) + { + String endpoint_address = endpoint; + if (endpoint_address == null || endpoint_address.equals("0.0.0.0")) + endpoint_address = range.endpoints.get(endpointIndex); + endpoints[endpointIndex++] = InetAddress.getByName(endpoint_address).getHostName(); + } + + Token.TokenFactory factory = partitioner.getTokenFactory(); + for (CfSplit subSplit : subSplits) + { + Token left = factory.fromString(subSplit.getStart_token()); + Token right = factory.fromString(subSplit.getEnd_token()); + Range range = new Range(left, right, partitioner); + List> ranges = range.isWrapAround() ? range.unwrap() : ImmutableList.of(range); + for (Range subrange : ranges) + { + ColumnFamilySplit split = + new ColumnFamilySplit( + factory.toString(subrange.left), + factory.toString(subrange.right), + subSplit.getRow_count(), + endpoints); + + logger.debug("adding " + split); + splits.add(split); + } + } + return splits; + } + } + + private List getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException + { + int splitsize = ConfigHelper.getInputSplitSize(conf); + for (int i = 0; i < range.rpc_endpoints.size(); i++) + { + String host = range.rpc_endpoints.get(i); + + if (host == null || host.equals("0.0.0.0")) + host = range.endpoints.get(i); + + try + { + Cassandra.Client client = ConfigHelper.createConnection(conf, host, ConfigHelper.getInputRpcPort(conf)); + client.set_keyspace(keyspace); + + try + { + return client.describe_splits_ex(cfName, range.start_token, range.end_token, splitsize); + } + catch (TApplicationException e) + { + // fallback to guessing split size if talking to a server without describe_splits_ex method + if (e.getType() == TApplicationException.UNKNOWN_METHOD) + { + List splitPoints = client.describe_splits(cfName, range.start_token, range.end_token, splitsize); + return tokenListToSplits(splitPoints, splitsize); + } + throw e; + } + } + catch (IOException e) + { + logger.debug("failed connect to endpoint " + host, e); + } + catch (InvalidRequestException e) + { + throw new RuntimeException(e); + } + catch (TException e) + { + throw new RuntimeException(e); + } + } + throw new IOException("failed connecting to all endpoints " + StringUtils.join(range.endpoints, ",")); + } + + private List tokenListToSplits(List splitTokens, int splitsize) + { + List splits = Lists.newArrayListWithExpectedSize(splitTokens.size() - 1); + for (int j = 0; j < splitTokens.size() - 1; j++) + splits.add(new CfSplit(splitTokens.get(j), splitTokens.get(j + 1), splitsize)); + return splits; + } + + private List getRangeMap(Configuration conf) throws IOException + { + Cassandra.Client client = ConfigHelper.getClientFromInputAddressList(conf); + + List map; + try + { + map = client.describe_ring(ConfigHelper.getInputKeyspace(conf)); + } + catch (InvalidRequestException e) + { + throw new RuntimeException(e); + } + catch (TException e) + { + throw new RuntimeException(e); + } + return map; + } + + // + // Old Hadoop API + // + public org.apache.hadoop.mapred.InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException + { + TaskAttemptContext tac = new TaskAttemptContext(jobConf, new TaskAttemptID()); + List newInputSplits = this.getSplits(tac); + org.apache.hadoop.mapred.InputSplit[] oldInputSplits = new org.apache.hadoop.mapred.InputSplit[newInputSplits.size()]; + for (int i = 0; i < newInputSplits.size(); i++) + oldInputSplits[i] = (ColumnFamilySplit)newInputSplits.get(i); + return oldInputSplits; + } +} diff --git a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyOutputFormat.java b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyOutputFormat.java new file mode 100644 index 0000000000..5a03777d07 --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyOutputFormat.java @@ -0,0 +1,159 @@ +/* + * 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; + + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.auth.IAuthenticator; +import org.apache.cassandra.thrift.*; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.*; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.transport.TTransport; + +/** + * The ColumnFamilyOutputFormat acts as a Hadoop-specific + * OutputFormat that allows reduce tasks to store keys (and corresponding + * values) as Cassandra rows (and respective columns) in a given + * ColumnFamily. + * + *

+ * As is the case with the {@link ColumnFamilyInputFormat}, you need to set the + * Keyspace and ColumnFamily in your + * Hadoop job Configuration. The {@link ConfigHelper} class, through its + * {@link ConfigHelper#setOutputColumnFamily} method, is provided to make this + * simple. + *

+ * + *

+ * For the sake of performance, this class employs a lazy write-back caching + * mechanism, where its record writer batches mutations created based on the + * reduce's inputs (in a task-specific map), and periodically makes the changes + * official by sending a batch mutate request to Cassandra. + *

+ * @param + */ +public abstract class AbstractColumnFamilyOutputFormat extends OutputFormat implements org.apache.hadoop.mapred.OutputFormat +{ + public static final String BATCH_THRESHOLD = "mapreduce.output.columnfamilyoutputformat.batch.threshold"; + public static final String QUEUE_SIZE = "mapreduce.output.columnfamilyoutputformat.queue.size"; + private static final Logger logger = LoggerFactory.getLogger(AbstractColumnFamilyOutputFormat.class); + + + /** + * Check for validity of the output-specification for the job. + * + * @param context + * information about the job + * @throws IOException + * when output should not be attempted + */ + public void checkOutputSpecs(JobContext context) + { + checkOutputSpecs(context.getConfiguration()); + } + + protected void checkOutputSpecs(Configuration conf) + { + if (ConfigHelper.getOutputKeyspace(conf) == null) + throw new UnsupportedOperationException("You must set the keyspace with setOutputKeyspace()"); + if (ConfigHelper.getOutputPartitioner(conf) == null) + throw new UnsupportedOperationException("You must set the output partitioner to the one used by your Cassandra cluster"); + if (ConfigHelper.getOutputInitialAddress(conf) == null) + throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node"); + } + + /** Fills the deprecated OutputFormat interface for streaming. */ + @Deprecated + public void checkOutputSpecs(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job) throws IOException + { + checkOutputSpecs(job); + } + + /** + * The OutputCommitter for this format does not write any data to the DFS. + * + * @param context + * the task context + * @return an output committer + * @throws IOException + * @throws InterruptedException + */ + public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException + { + return new NullOutputCommitter(); + } + + /** + * Connects to the given server:port and returns a client based on the given socket that points to the configured + * keyspace, and is logged in with the configured credentials. + * + * @param host fully qualified host name to connect to + * @param port RPC port of the server + * @param conf a job configuration + * @return a cassandra client + * @throws Exception set of thrown exceptions may be implementation defined, + * depending on the used transport factory + */ + public static Cassandra.Client createAuthenticatedClient(String host, int port, Configuration conf) throws Exception + { + logger.debug("Creating authenticated client for CF output format"); + TTransport transport = ConfigHelper.getClientTransportFactory(conf).openTransport(host, port, conf); + TProtocol binaryProtocol = new TBinaryProtocol(transport, true, true); + Cassandra.Client client = new Cassandra.Client(binaryProtocol); + client.set_keyspace(ConfigHelper.getOutputKeyspace(conf)); + if (ConfigHelper.getOutputKeyspaceUserName(conf) != null) + { + Map creds = new HashMap(); + creds.put(IAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(conf)); + creds.put(IAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(conf)); + AuthenticationRequest authRequest = new AuthenticationRequest(creds); + client.login(authRequest); + } + logger.debug("Authenticated client for CF output format created successfully"); + return client; + } + + /** + * An {@link OutputCommitter} that does nothing. + */ + private static class NullOutputCommitter extends OutputCommitter + { + public void abortTask(TaskAttemptContext taskContext) { } + + public void cleanupJob(JobContext jobContext) { } + + public void commitTask(TaskAttemptContext taskContext) { } + + public boolean needsTaskCommit(TaskAttemptContext taskContext) + { + return false; + } + + public void setupJob(JobContext jobContext) { } + + public void setupTask(TaskAttemptContext taskContext) { } + } +} diff --git a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyRecordWriter.java b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyRecordWriter.java new file mode 100644 index 0000000000..6428db353e --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyRecordWriter.java @@ -0,0 +1,193 @@ +/* + * 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; + + +import java.io.IOException; +import java.net.InetAddress; +import java.nio.ByteBuffer; +import java.util.*; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.apache.cassandra.client.RingCache; +import org.apache.cassandra.thrift.*; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.RecordWriter; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.thrift.transport.TTransport; + + +/** + * The ColumnFamilyRecordWriter maps the output <key, value> + * pairs to a Cassandra column family. In particular, it applies all mutations + * in the value, which it associates with the key, and in turn the responsible + * endpoint. + * + *

+ * Furthermore, this writer groups the mutations by the endpoint responsible for + * the rows being affected. This allows the mutations to be executed in parallel, + * directly to a responsible endpoint. + *

+ * + * @see ColumnFamilyOutputFormat + */ +public abstract class AbstractColumnFamilyRecordWriter extends RecordWriter implements org.apache.hadoop.mapred.RecordWriter +{ + // The configuration this writer is associated with. + protected final Configuration conf; + + // The ring cache that describes the token ranges each node in the ring is + // responsible for. This is what allows us to group the mutations by + // the endpoints they should be targeted at. The targeted endpoint + // essentially + // acts as the primary replica for the rows being affected by the mutations. + protected final RingCache ringCache; + + // The number of mutations to buffer per endpoint + protected final int queueSize; + + protected final long batchThreshold; + + protected final ConsistencyLevel consistencyLevel; + protected Progressable progressable; + + protected AbstractColumnFamilyRecordWriter(Configuration conf) + { + this.conf = conf; + this.ringCache = new RingCache(conf); + this.queueSize = conf.getInt(AbstractColumnFamilyOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors()); + batchThreshold = conf.getLong(AbstractColumnFamilyOutputFormat.BATCH_THRESHOLD, 32); + consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getWriteConsistencyLevel(conf)); + } + + /** + * Close this RecordWriter to future operations, but not before + * flushing out the batched mutations. + * + * @param context the context of the task + * @throws IOException + */ + public void close(TaskAttemptContext context) throws IOException, InterruptedException + { + close(); + } + + /** Fills the deprecated RecordWriter interface for streaming. */ + @Deprecated + public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException + { + close(); + } + + protected abstract void close() throws IOException; + + /** + * A client that runs in a threadpool and connects to the list of endpoints for a particular + * range. Mutations for keys in that range are sent to this client via a queue. + */ + public abstract class AbstractRangeClient extends Thread + { + // The list of endpoints for this range + protected final List endpoints; + // A bounded queue of incoming mutations for this range + protected final BlockingQueue> queue = new ArrayBlockingQueue>(queueSize); + + protected volatile boolean run = true; + // we want the caller to know if something went wrong, so we record any unrecoverable exception while writing + // so we can throw it on the caller's stack when he calls put() again, or if there are no more put calls, + // when the client is closed. + protected volatile IOException lastException; + + protected Cassandra.Client client; + + /** + * Constructs an {@link AbstractRangeClient} for the given endpoints. + * @param endpoints the possible endpoints to execute the mutations on + */ + public AbstractRangeClient(List endpoints) + { + super("client-" + endpoints); + this.endpoints = endpoints; + } + + /** + * enqueues the given value to Cassandra + */ + public void put(Pair value) throws IOException + { + while (true) + { + if (lastException != null) + throw lastException; + try + { + if (queue.offer(value, 100, TimeUnit.MILLISECONDS)) + break; + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + } + } + + public void close() throws IOException + { + // stop the run loop. this will result in closeInternal being called by the time join() finishes. + run = false; + interrupt(); + try + { + this.join(); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + + if (lastException != null) + throw lastException; + } + + protected void closeInternal() + { + if (client != null) + { + TTransport transport = client.getOutputProtocol().getTransport(); + if (transport.isOpen()) + transport.close(); + } + } + + /** + * Loops collecting mutations from the queue and sending to Cassandra + */ + public abstract void run(); + + @Override + public String toString() + { + return "#"; + } + } +} + diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java index 404ef60185..1b5a4e29e1 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java @@ -18,35 +18,14 @@ package org.apache.cassandra.hadoop; import java.io.IOException; -import java.net.InetAddress; import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Random; -import java.util.SortedMap; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; - -import com.google.common.collect.ImmutableList; -import com.google.common.collect.Lists; -import org.apache.commons.lang.StringUtils; -import org.apache.thrift.TApplicationException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import java.util.*; import org.apache.cassandra.db.Column; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.thrift.*; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapreduce.*; -import org.apache.thrift.TException; /** * Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily. @@ -65,283 +44,14 @@ import org.apache.thrift.TException; * * The default split size is 64k rows. */ -public class ColumnFamilyInputFormat extends InputFormat> - implements org.apache.hadoop.mapred.InputFormat> +public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat> { - private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyInputFormat.class); - - public static final String MAPRED_TASK_ID = "mapred.task.id"; - // The simple fact that we need this is because the old Hadoop API wants us to "write" - // to the key and value whereas the new asks for it. - // I choose 8kb as the default max key size (instanciated only once), but you can - // override it in your jobConf with this setting. - public static final String CASSANDRA_HADOOP_MAX_KEY_SIZE = "cassandra.hadoop.max_key_size"; - public static final int CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT = 8192; - - private String keyspace; - private String cfName; - private IPartitioner partitioner; - - private static void validateConfiguration(Configuration conf) - { - if (ConfigHelper.getInputKeyspace(conf) == null || ConfigHelper.getInputColumnFamily(conf) == null) - { - throw new UnsupportedOperationException("you must set the keyspace and columnfamily with setInputColumnFamily()"); - } - if (ConfigHelper.getInputSlicePredicate(conf) == null) - { - throw new UnsupportedOperationException("you must set the predicate with setInputSlicePredicate"); - } - if (ConfigHelper.getInputInitialAddress(conf) == null) - throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node with setInputInitialAddress"); - if (ConfigHelper.getInputPartitioner(conf) == null) - throw new UnsupportedOperationException("You must set the Cassandra partitioner class with setInputPartitioner"); - } - - public List getSplits(JobContext context) throws IOException - { - Configuration conf = context.getConfiguration(); - - validateConfiguration(conf); - - // cannonical ranges and nodes holding replicas - List masterRangeNodes = getRangeMap(conf); - - keyspace = ConfigHelper.getInputKeyspace(context.getConfiguration()); - cfName = ConfigHelper.getInputColumnFamily(context.getConfiguration()); - partitioner = ConfigHelper.getInputPartitioner(context.getConfiguration()); - logger.debug("partitioner is " + partitioner); - - // cannonical ranges, split into pieces, fetching the splits in parallel - ExecutorService executor = Executors.newCachedThreadPool(); - List splits = new ArrayList(); - - try - { - List>> splitfutures = new ArrayList>>(); - KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf); - Range jobRange = null; - if (jobKeyRange != null) - { - if (jobKeyRange.start_key == null) - { - logger.warn("ignoring jobKeyRange specified without start_key"); - } - else - { - if (!partitioner.preservesOrder()) - throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving paritioner"); - if (jobKeyRange.start_token != null) - throw new IllegalArgumentException("only start_key supported"); - if (jobKeyRange.end_token != null) - throw new IllegalArgumentException("only start_key supported"); - jobRange = new Range(partitioner.getToken(jobKeyRange.start_key), - partitioner.getToken(jobKeyRange.end_key), - partitioner); - } - } - - for (TokenRange range : masterRangeNodes) - { - if (jobRange == null) - { - // for each range, pick a live owner and ask it to compute bite-sized splits - splitfutures.add(executor.submit(new SplitCallable(range, conf))); - } - else - { - Range dhtRange = new Range(partitioner.getTokenFactory().fromString(range.start_token), - partitioner.getTokenFactory().fromString(range.end_token), - partitioner); - - if (dhtRange.intersects(jobRange)) - { - for (Range intersection: dhtRange.intersectionWith(jobRange)) - { - range.start_token = partitioner.getTokenFactory().toString(intersection.left); - range.end_token = partitioner.getTokenFactory().toString(intersection.right); - // for each range, pick a live owner and ask it to compute bite-sized splits - splitfutures.add(executor.submit(new SplitCallable(range, conf))); - } - } - } - } - - // wait until we have all the results back - for (Future> futureInputSplits : splitfutures) - { - try - { - splits.addAll(futureInputSplits.get()); - } - catch (Exception e) - { - throw new IOException("Could not get input splits", e); - } - } - } - finally - { - executor.shutdownNow(); - } - - assert splits.size() > 0; - Collections.shuffle(splits, new Random(System.nanoTime())); - return splits; - } - - /** - * Gets a token range and splits it up according to the suggested - * size into input splits that Hadoop can use. - */ - class SplitCallable implements Callable> - { - - private final TokenRange range; - private final Configuration conf; - - public SplitCallable(TokenRange tr, Configuration conf) - { - this.range = tr; - this.conf = conf; - } - - public List call() throws Exception - { - ArrayList splits = new ArrayList(); - List subSplits = getSubSplits(keyspace, cfName, range, conf); - assert range.rpc_endpoints.size() == range.endpoints.size() : "rpc_endpoints size must match endpoints size"; - // turn the sub-ranges into InputSplits - String[] endpoints = range.endpoints.toArray(new String[range.endpoints.size()]); - // hadoop needs hostname, not ip - int endpointIndex = 0; - for (String endpoint: range.rpc_endpoints) - { - String endpoint_address = endpoint; - if (endpoint_address == null || endpoint_address.equals("0.0.0.0")) - endpoint_address = range.endpoints.get(endpointIndex); - endpoints[endpointIndex++] = InetAddress.getByName(endpoint_address).getHostName(); - } - - Token.TokenFactory factory = partitioner.getTokenFactory(); - for (CfSplit subSplit : subSplits) - { - Token left = factory.fromString(subSplit.getStart_token()); - Token right = factory.fromString(subSplit.getEnd_token()); - Range range = new Range(left, right, partitioner); - List> ranges = range.isWrapAround() ? range.unwrap() : ImmutableList.of(range); - for (Range subrange : ranges) - { - ColumnFamilySplit split = - new ColumnFamilySplit( - factory.toString(subrange.left), - factory.toString(subrange.right), - subSplit.getRow_count(), - endpoints); - - logger.debug("adding " + split); - splits.add(split); - } - } - return splits; - } - } - - private List getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException - { - int splitsize = ConfigHelper.getInputSplitSize(conf); - for (int i = 0; i < range.rpc_endpoints.size(); i++) - { - String host = range.rpc_endpoints.get(i); - - if (host == null || host.equals("0.0.0.0")) - host = range.endpoints.get(i); - - try - { - Cassandra.Client client = ConfigHelper.createConnection(conf, host, ConfigHelper.getInputRpcPort(conf)); - client.set_keyspace(keyspace); - - try - { - return client.describe_splits_ex(cfName, range.start_token, range.end_token, splitsize); - } - catch (TApplicationException e) - { - // fallback to guessing split size if talking to a server without describe_splits_ex method - if (e.getType() == TApplicationException.UNKNOWN_METHOD) - { - List splitPoints = client.describe_splits(cfName, range.start_token, range.end_token, splitsize); - return tokenListToSplits(splitPoints, splitsize); - } - throw e; - } - } - catch (IOException e) - { - logger.debug("failed connect to endpoint " + host, e); - } - catch (InvalidRequestException e) - { - throw new RuntimeException(e); - } - catch (TException e) - { - throw new RuntimeException(e); - } - } - throw new IOException("failed connecting to all endpoints " + StringUtils.join(range.endpoints, ",")); - } - - - private List tokenListToSplits(List splitTokens, int splitsize) - { - List splits = Lists.newArrayListWithExpectedSize(splitTokens.size() - 1); - for (int j = 0; j < splitTokens.size() - 1; j++) - splits.add(new CfSplit(splitTokens.get(j), splitTokens.get(j + 1), splitsize)); - return splits; - } - - - private List getRangeMap(Configuration conf) throws IOException - { - Cassandra.Client client = ConfigHelper.getClientFromInputAddressList(conf); - - List map; - try - { - map = client.describe_ring(ConfigHelper.getInputKeyspace(conf)); - } - catch (InvalidRequestException e) - { - throw new RuntimeException(e); - } - catch (TException e) - { - throw new RuntimeException(e); - } - return map; - } - + public RecordReader> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { return new ColumnFamilyRecordReader(); } - - // - // Old Hadoop API - // - public org.apache.hadoop.mapred.InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException - { - TaskAttemptContext tac = new TaskAttemptContext(jobConf, new TaskAttemptID()); - List newInputSplits = this.getSplits(tac); - org.apache.hadoop.mapred.InputSplit[] oldInputSplits = new org.apache.hadoop.mapred.InputSplit[newInputSplits.size()]; - for (int i = 0; i < newInputSplits.size(); i++) - oldInputSplits[i] = (ColumnFamilySplit)newInputSplits.get(i); - return oldInputSplits; - } - public org.apache.hadoop.mapred.RecordReader> getRecordReader(org.apache.hadoop.mapred.InputSplit split, JobConf jobConf, final Reporter reporter) throws IOException { TaskAttemptContext tac = new TaskAttemptContext(jobConf, TaskAttemptID.forName(jobConf.get(MAPRED_TASK_ID))) @@ -357,5 +67,16 @@ public class ColumnFamilyInputFormat extends InputFormatColumnFamilyOutputFormat acts as a Hadoop-specific @@ -59,62 +45,10 @@ import javax.security.auth.login.LoginException; * official by sending a batch mutate request to Cassandra. *

*/ -public class ColumnFamilyOutputFormat extends OutputFormat> - implements org.apache.hadoop.mapred.OutputFormat> +public class ColumnFamilyOutputFormat extends AbstractColumnFamilyOutputFormat> { - public static final String BATCH_THRESHOLD = "mapreduce.output.columnfamilyoutputformat.batch.threshold"; - public static final String QUEUE_SIZE = "mapreduce.output.columnfamilyoutputformat.queue.size"; - private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyOutputFormat.class); - - - /** - * Check for validity of the output-specification for the job. - * - * @param context - * information about the job - * @throws IOException - * when output should not be attempted - */ - @Override - public void checkOutputSpecs(JobContext context) - { - checkOutputSpecs(context.getConfiguration()); - } - - private void checkOutputSpecs(Configuration conf) - { - if (ConfigHelper.getOutputKeyspace(conf) == null) - throw new UnsupportedOperationException("You must set the keyspace with setOutputKeyspace()"); - if (ConfigHelper.getOutputPartitioner(conf) == null) - throw new UnsupportedOperationException("You must set the output partitioner to the one used by your Cassandra cluster"); - if (ConfigHelper.getOutputInitialAddress(conf) == null) - throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node"); - } - - /** - * The OutputCommitter for this format does not write any data to the DFS. - * - * @param context - * the task context - * @return an output committer - * @throws IOException - * @throws InterruptedException - */ - @Override - public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException - { - return new NullOutputCommitter(); - } - /** Fills the deprecated OutputFormat interface for streaming. */ @Deprecated - 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) { return new ColumnFamilyRecordWriter(job, new Progressable(progress)); @@ -128,62 +62,8 @@ public class ColumnFamilyOutputFormat extends OutputFormat creds = new HashMap(); - creds.put(IAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(conf)); - creds.put(IAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(conf)); - AuthenticationRequest authRequest = new AuthenticationRequest(creds); - client.login(authRequest); - } - logger.debug("Authenticated client for CF output format created successfully"); - return client; - } - - /** - * An {@link OutputCommitter} that does nothing. - */ - public static class NullOutputCommitter extends OutputCommitter - { - public void abortTask(TaskAttemptContext taskContext) { } - - public void cleanupJob(JobContext jobContext) { } - - public void commitTask(TaskAttemptContext taskContext) { } - - public boolean needsTaskCommit(TaskAttemptContext taskContext) - { - return false; - } - - public void setupJob(JobContext jobContext) { } - - public void setupTask(TaskAttemptContext taskContext) { } - } } diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java index 2b258b2d3e..9022202f81 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java @@ -24,33 +24,16 @@ import java.nio.ByteBuffer; import java.util.*; import com.google.common.collect.*; - -import org.apache.thrift.protocol.TBinaryProtocol; -import org.apache.thrift.transport.TTransport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.cassandra.auth.IAuthenticator; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.db.Column; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.CompositeType; import org.apache.cassandra.db.marshal.TypeParser; import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.thrift.AuthenticationRequest; -import org.apache.cassandra.thrift.Cassandra; -import org.apache.cassandra.thrift.CfDef; -import org.apache.cassandra.thrift.ColumnOrSuperColumn; -import org.apache.cassandra.thrift.ColumnParent; -import org.apache.cassandra.thrift.ConsistencyLevel; -import org.apache.cassandra.thrift.CounterColumn; -import org.apache.cassandra.thrift.CounterSuperColumn; -import org.apache.cassandra.thrift.IndexExpression; -import org.apache.cassandra.thrift.KeyRange; -import org.apache.cassandra.thrift.KeySlice; -import org.apache.cassandra.thrift.KsDef; -import org.apache.cassandra.thrift.SlicePredicate; -import org.apache.cassandra.thrift.SuperColumn; +import org.apache.cassandra.thrift.*; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; @@ -59,7 +42,7 @@ import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.thrift.TException; -import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; public class ColumnFamilyRecordReader extends RecordReader> implements org.apache.hadoop.mapred.RecordReader> @@ -75,14 +58,14 @@ public class ColumnFamilyRecordReader extends RecordReader filter; + public ColumnFamilyRecordReader() { this(ColumnFamilyRecordReader.CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT); @@ -96,11 +79,11 @@ public class ColumnFamilyRecordReader extends RecordReader creds = new HashMap(); - creds.put(IAuthenticator.USERNAME_KEY, ConfigHelper.getInputKeyspaceUserName(conf)); - creds.put(IAuthenticator.PASSWORD_KEY, ConfigHelper.getInputKeyspacePassword(conf)); - AuthenticationRequest authRequest = new AuthenticationRequest(creds); - client.login(authRequest); - } + int port = ConfigHelper.getInputRpcPort(conf); + client = ColumnFamilyInputFormat.createAuthenticatedClient(location, port, conf); + } catch (Exception e) { diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java index 10de8ce3fc..50ec0599a8 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java @@ -22,21 +22,14 @@ import java.io.IOException; import java.net.InetAddress; import java.nio.ByteBuffer; import java.util.*; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.TimeUnit; -import org.apache.cassandra.client.RingCache; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.thrift.*; -import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapreduce.RecordWriter; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.thrift.TException; -import org.apache.thrift.transport.TSocket; /** @@ -52,38 +45,18 @@ import org.apache.thrift.transport.TSocket; *

* * @see ColumnFamilyOutputFormat - * @see OutputFormat - * */ -final class ColumnFamilyRecordWriter extends RecordWriter> -implements org.apache.hadoop.mapred.RecordWriter> +final class ColumnFamilyRecordWriter extends AbstractColumnFamilyRecordWriter> { - // The configuration this writer is associated with. - private final Configuration conf; - - // The ring cache that describes the token ranges each node in the ring is - // responsible for. This is what allows us to group the mutations by - // the endpoints they should be targeted at. The targeted endpoint - // essentially - // acts as the primary replica for the rows being affected by the mutations. - private final RingCache ringCache; - - // The number of mutations to buffer per endpoint - private final int queueSize; - // handles for clients for each range running in the threadpool - private final Map clients; - private final long batchThreshold; - - private final ConsistencyLevel consistencyLevel; - private Progressable progressable; - - + private final Map clients; + /** * Upon construction, obtain the map that this writer will use to collect * mutations, and the ring cache for the given keyspace. * * @param context the task attempt context + * @throws IOException */ ColumnFamilyRecordWriter(TaskAttemptContext context) { @@ -99,14 +72,30 @@ implements org.apache.hadoop.mapred.RecordWriter> ColumnFamilyRecordWriter(Configuration conf) { - this.conf = conf; - this.ringCache = new RingCache(conf); - this.queueSize = conf.getInt(ColumnFamilyOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors()); - this.clients = new HashMap(); - batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, 32); - consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getWriteConsistencyLevel(conf)); + super(conf); + this.clients = new HashMap(); } - + + @Override + public void close() throws IOException + { + // close all the clients before throwing anything + IOException clientException = null; + for (RangeClient client : clients.values()) + { + try + { + client.close(); + } + catch (IOException e) + { + clientException = e; + } + } + if (clientException != null) + throw clientException; + } + /** * If the key is to be associated with a valid value, a mutation is created * for it with the given column family and columns. In the event the value @@ -141,125 +130,23 @@ implements org.apache.hadoop.mapred.RecordWriter> progressable.progress(); } - /** - * Close this RecordWriter to future operations, but not before - * flushing out the batched mutations. - * - * @param context the context of the task - * @throws IOException - */ - @Override - public void close(TaskAttemptContext context) throws IOException, InterruptedException - { - close(); - } - - /** Fills the deprecated RecordWriter interface for streaming. */ - @Deprecated - public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException - { - close(); - } - - private void close() throws IOException - { - // close all the clients before throwing anything - IOException clientException = null; - for (RangeClient client : clients.values()) - { - try - { - client.close(); - } - catch (IOException e) - { - clientException = e; - } - } - if (clientException != null) - throw clientException; - } - /** * A client that runs in a threadpool and connects to the list of endpoints for a particular * range. Mutations for keys in that range are sent to this client via a queue. */ - public class RangeClient extends Thread + public class RangeClient extends AbstractRangeClient { - // The list of endpoints for this range - private final List endpoints; - private final String columnFamily = ConfigHelper.getOutputColumnFamily(conf); - // A bounded queue of incoming mutations for this range - private final BlockingQueue> queue = new ArrayBlockingQueue>(queueSize); - - private volatile boolean run = true; - // we want the caller to know if something went wrong, so we record any unrecoverable exception while writing - // so we can throw it on the caller's stack when he calls put() again, or if there are no more put calls, - // when the client is closed. - private volatile IOException lastException; - - private Cassandra.Client thriftClient; - private TSocket thriftSocket; - + public final String columnFamily = ConfigHelper.getOutputColumnFamily(conf); + /** * Constructs an {@link RangeClient} for the given endpoints. * @param endpoints the possible endpoints to execute the mutations on */ public RangeClient(List endpoints) { - super("client-" + endpoints); - this.endpoints = endpoints; + super(endpoints); } - - /** - * enqueues the given value to Cassandra - */ - public void put(Pair value) throws IOException - { - while (true) - { - if (lastException != null) - throw lastException; - try - { - if (queue.offer(value, 100, TimeUnit.MILLISECONDS)) - break; - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } - } - } - - public void close() throws IOException - { - // stop the run loop. this will result in closeInternal being called by the time join() finishes. - run = false; - interrupt(); - try - { - this.join(); - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } - - if (lastException != null) - throw lastException; - } - - private void closeInternal() - { - if (thriftSocket != null) - { - thriftSocket.close(); - thriftSocket = null; - thriftClient = null; - } - } - + /** * Loops collecting mutations from the queue and sending to Cassandra */ @@ -302,7 +189,7 @@ implements org.apache.hadoop.mapred.RecordWriter> // send the mutation to the last-used endpoint. first time through, this will NPE harmlessly. try { - thriftClient.batch_mutate(batch, consistencyLevel); + client.batch_mutate(batch, consistencyLevel); break; } catch (Exception e) @@ -319,8 +206,9 @@ implements org.apache.hadoop.mapred.RecordWriter> try { InetAddress address = iter.next(); - thriftSocket = new TSocket(address.getHostName(), ConfigHelper.getOutputRpcPort(conf)); - thriftClient = ColumnFamilyOutputFormat.createAuthenticatedClient(thriftSocket, conf); + String host = address.getHostName(); + int port = ConfigHelper.getOutputRpcPort(conf); + client = ColumnFamilyOutputFormat.createAuthenticatedClient(host, port, conf); } catch (Exception e) { @@ -336,11 +224,5 @@ implements org.apache.hadoop.mapred.RecordWriter> } } } - - @Override - public String toString() - { - return "#"; - } } } diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java index ed1f1607c1..69c7ddba76 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java @@ -22,6 +22,7 @@ import org.apache.hadoop.mapreduce.InputSplit; import java.io.DataInput; import java.io.DataOutput; +import java.io.EOFException; import java.io.IOException; import java.util.Arrays; @@ -79,7 +80,6 @@ public class ColumnFamilySplit extends InputSplit implements Writable, org.apach { out.writeUTF(startToken); out.writeUTF(endToken); - out.writeLong(length); out.writeInt(dataNodes.length); for (String endpoint : dataNodes) { @@ -91,8 +91,6 @@ public class ColumnFamilySplit extends InputSplit implements Writable, org.apach { startToken = in.readUTF(); endToken = in.readUTF(); - length = in.readLong(); - int numOfEndpoints = in.readInt(); dataNodes = new String[numOfEndpoints]; for(int i = 0; i < numOfEndpoints; i++) diff --git a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java b/src/java/org/apache/cassandra/hadoop/ConfigHelper.java index 90f5045baf..3dcfdd7faf 100644 --- a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java +++ b/src/java/org/apache/cassandra/hadoop/ConfigHelper.java @@ -21,11 +21,9 @@ package org.apache.cassandra.hadoop; */ import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; +import com.google.common.collect.Maps; import org.apache.cassandra.io.compress.CompressionParameters; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,11 +39,7 @@ import org.apache.thrift.TDeserializer; import org.apache.thrift.TException; import org.apache.thrift.TSerializer; import org.apache.thrift.protocol.TBinaryProtocol; -import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; -import org.apache.thrift.transport.TTransportException; - -import javax.security.auth.login.LoginException; public class ConfigHelper @@ -75,6 +69,7 @@ public class ConfigHelper private static final String WRITE_CONSISTENCY_LEVEL = "cassandra.consistencylevel.write"; private static final String OUTPUT_COMPRESSION_CLASS = "cassandra.output.compression.class"; private static final String OUTPUT_COMPRESSION_CHUNK_LENGTH = "cassandra.output.compression.length"; + private static final String INPUT_TRANSPORT_FACTORY_CLASS = "cassandra.input.transport.factory.class"; private static final String OUTPUT_TRANSPORT_FACTORY_CLASS = "cassandra.output.transport.factory.class"; private static final String THRIFT_FRAMED_TRANSPORT_SIZE_IN_MB = "cassandra.thrift.framed.size_mb"; @@ -562,48 +557,44 @@ public class ConfigHelper { try { - TSocket socket = new TSocket(host, port); - TTransport transport = getInputTransportFactory(conf).openTransport(socket, conf); + TTransport transport = getClientTransportFactory(conf).openTransport(host, port, conf); return new Cassandra.Client(new TBinaryProtocol(transport, true, true)); } - catch (LoginException e) - { - throw new IOException("Unable to login to server " + host + ":" + port, e); - } - catch (TTransportException e) + catch (Exception e) { throw new IOException("Unable to connect to server " + host + ":" + port, e); } } - public static ITransportFactory getInputTransportFactory(Configuration conf) + public static TClientTransportFactory getClientTransportFactory(Configuration conf) { - return getTransportFactory(conf.get(INPUT_TRANSPORT_FACTORY_CLASS, TFramedTransportFactory.class.getName())); + String factoryClassName = conf.get( + TClientTransportFactory.PROPERTY_KEY, + TFramedTransportFactory.class.getName()); + TClientTransportFactory factory = getClientTransportFactory(factoryClassName); + Map options = getOptions(conf, factory.supportedOptions()); + factory.setOptions(options); + return factory; } - public static void setInputTransportFactoryClass(Configuration conf, String classname) - { - conf.set(INPUT_TRANSPORT_FACTORY_CLASS, classname); - } - - public static ITransportFactory getOutputTransportFactory(Configuration conf) - { - return getTransportFactory(conf.get(OUTPUT_TRANSPORT_FACTORY_CLASS, TFramedTransportFactory.class.getName())); - } - - public static void setOutputTransportFactoryClass(Configuration conf, String classname) - { - conf.set(OUTPUT_TRANSPORT_FACTORY_CLASS, classname); - } - - private static ITransportFactory getTransportFactory(String factoryClassName) { + private static TClientTransportFactory getClientTransportFactory(String factoryClassName) { try { - return (ITransportFactory) Class.forName(factoryClassName).newInstance(); + return (TClientTransportFactory) Class.forName(factoryClassName).newInstance(); } catch (Exception e) { throw new RuntimeException("Failed to instantiate transport factory:" + factoryClassName, e); } } + private static Map getOptions(Configuration conf, Set supportedOptions) { + Map options = Maps.newHashMap(); + for (String optionKey : supportedOptions) + { + String optionValue = conf.get(optionKey); + if (optionValue != null) + options.put(optionKey, optionValue); + } + return options; + } } diff --git a/src/java/org/apache/cassandra/hadoop/Progressable.java b/src/java/org/apache/cassandra/hadoop/Progressable.java index 091a828776..ac253ef744 100644 --- a/src/java/org/apache/cassandra/hadoop/Progressable.java +++ b/src/java/org/apache/cassandra/hadoop/Progressable.java @@ -29,12 +29,12 @@ public class Progressable private TaskAttemptContext context; private org.apache.hadoop.util.Progressable progressable; - Progressable(TaskAttemptContext context) + public Progressable(TaskAttemptContext context) { this.context = context; } - Progressable(org.apache.hadoop.util.Progressable progressable) + public Progressable(org.apache.hadoop.util.Progressable progressable) { this.progressable = progressable; } diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CQLConfigHelper.java b/src/java/org/apache/cassandra/hadoop/cql3/CQLConfigHelper.java new file mode 100644 index 0000000000..66bcfdb6eb --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/cql3/CQLConfigHelper.java @@ -0,0 +1,109 @@ +package org.apache.cassandra.hadoop.cql3; +/* +* +* 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. +* +*/ +import org.apache.hadoop.conf.Configuration; + +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"; + private static final String INPUT_CQL_WHERE_CLAUSE_CONFIG = "cassandra.input.where.clause"; + private static final String OUTPUT_CQL = "cassandra.output.cql"; + + /** + * Set the CQL columns for the input of this job. + * + * @param conf Job configuration you are about to run + * @param columns + */ + public static void setInputColumns(Configuration conf, String columns) + { + if (columns == null || columns.isEmpty()) + return; + + conf.set(INPUT_CQL_COLUMNS_CONFIG, columns); + } + + /** + * Set the CQL query Limit for the input of this job. + * + * @param conf Job configuration you are about to run + * @param cqlPageRowSize + */ + public static void setInputCQLPageRowSize(Configuration conf, String cqlPageRowSize) + { + if (cqlPageRowSize == null) + { + throw new UnsupportedOperationException("cql page row size may not be null"); + } + + conf.set(INPUT_CQL_PAGE_ROW_SIZE_CONFIG, cqlPageRowSize); + } + + /** + * Set the CQL user defined where clauses for the input of this job. + * + * @param conf Job configuration you are about to run + * @param clauses + */ + public static void setInputWhereClauses(Configuration conf, String clauses) + { + if (clauses == null || clauses.isEmpty()) + return; + + conf.set(INPUT_CQL_WHERE_CLAUSE_CONFIG, clauses); + } + + /** + * Set the CQL prepared statement for the output of this job. + * + * @param conf Job configuration you are about to run + * @param cql + */ + public static void setOutputCql(Configuration conf, String cql) + { + if (cql == null || cql.isEmpty()) + return; + + conf.set(OUTPUT_CQL, cql); + } + + + public static String getInputcolumns(Configuration conf) + { + return conf.get(INPUT_CQL_COLUMNS_CONFIG); + } + + public static String getInputPageRowSize(Configuration conf) + { + return conf.get(INPUT_CQL_PAGE_ROW_SIZE_CONFIG); + } + + public static String getInputWhereClauses(Configuration conf) + { + return conf.get(INPUT_CQL_WHERE_CLAUSE_CONFIG); + } + + public static String getOutputCql(Configuration conf) + { + return conf.get(OUTPUT_CQL); + } +} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyInputFormat.java new file mode 100644 index 0000000000..525ed89578 --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyInputFormat.java @@ -0,0 +1,83 @@ +/* + * 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.cql3; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Map; + +import org.apache.cassandra.hadoop.AbstractColumnFamilyInputFormat; +import org.apache.hadoop.mapred.InputSplit; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.RecordReader; +import org.apache.hadoop.mapred.Reporter; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.TaskAttemptID; + +/** + * Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily. + * + * At minimum, you need to set the KS and CF in your Hadoop job Configuration. + * The ConfigHelper class is provided to make this + * simple: + * ConfigHelper.setInputColumnFamily + * + * You can also configure the number of rows per InputSplit with + * ConfigHelper.setInputSplitSize. The default split size is 64k rows. + * the number of CQL rows per page + * + * the number of CQL rows per page + * CQLConfigHelper.setInputCQLPageRowSize. The default page row size is 1000. You + * should set it to "as big as possible, but no bigger." It set the LIMIT for the CQL + * query, so you need set it big enough to minimize the network overhead, and also + * not too big to avoid out of memory issue. + * + * the column names of the select CQL query. The default is all columns + * CQLConfigHelper.setInputColumns + * + * the user defined the where clause + * CQLConfigHelper.setInputWhereClauses. The default is no user defined where clause + */ +public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat, Map> +{ + public RecordReader, Map> getRecordReader(InputSplit split, JobConf jobConf, final Reporter reporter) + throws IOException + { + TaskAttemptContext tac = new TaskAttemptContext(jobConf, TaskAttemptID.forName(jobConf.get(MAPRED_TASK_ID))) + { + @Override + public void progress() + { + reporter.progress(); + } + }; + + ColumnFamilyRecordReader recordReader = new ColumnFamilyRecordReader(); + recordReader.initialize((org.apache.hadoop.mapreduce.InputSplit)split, tac); + return recordReader; + } + + @Override + public org.apache.hadoop.mapreduce.RecordReader, Map> createRecordReader( + org.apache.hadoop.mapreduce.InputSplit arg0, TaskAttemptContext arg1) throws IOException, + InterruptedException + { + return new ColumnFamilyRecordReader(); + } + +} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyOutputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyOutputFormat.java new file mode 100644 index 0000000000..3f6e2afafd --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyOutputFormat.java @@ -0,0 +1,78 @@ +/* + * 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.cql3; + + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; + +import org.apache.cassandra.hadoop.AbstractColumnFamilyOutputFormat; +import org.apache.cassandra.hadoop.ColumnFamilyInputFormat; +import org.apache.cassandra.hadoop.ConfigHelper; +import org.apache.cassandra.hadoop.Progressable; +import org.apache.hadoop.mapreduce.*; + +/** + * The ColumnFamilyOutputFormat acts as a Hadoop-specific + * OutputFormat that allows reduce tasks to store keys (and corresponding + * binded variable values) as CQL rows (and respective columns) in a given + * ColumnFamily. + * + *

+ * 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 + * {@link ConfigHelper#setOutputPreparedStatement} method, is provided to make this + * simple. + * you need to set the Keyspace. The {@link ConfigHelper} class, through its + * {@link ConfigHelper#setOutputColumnFamily} method, is provided to make this + * simple. + *

+ * + *

+ * For the sake of performance, this class employs a lazy write-back caching + * mechanism, where its record writer prepared statement binded variable values + * created based on the reduce's inputs (in a task-specific map), and periodically + * makes the changes official by sending a execution of prepared statement request + * to Cassandra. + *

+ */ +public class ColumnFamilyOutputFormat extends AbstractColumnFamilyOutputFormat, List> +{ + /** 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 + { + return new ColumnFamilyRecordWriter(job, new Progressable(progress)); + } + + /** + * Get the {@link RecordWriter} for the given task. + * + * @param context + * the information about the current task. + * @return a {@link RecordWriter} to write the output for the job. + * @throws IOException + */ + public ColumnFamilyRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException + { + return new ColumnFamilyRecordWriter(context); + } +} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordReader.java b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordReader.java new file mode 100644 index 0000000000..03d7af582d --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordReader.java @@ -0,0 +1,763 @@ +/* + * 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.cql3; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.util.*; + +import com.google.common.collect.AbstractIterator; +import com.google.common.collect.Iterables; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.db.marshal.CompositeType; +import org.apache.cassandra.db.marshal.LongType; +import org.apache.cassandra.db.marshal.TypeParser; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.exceptions.SyntaxException; +import org.apache.cassandra.hadoop.ColumnFamilySplit; +import org.apache.cassandra.hadoop.ConfigHelper; +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.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.thrift.TException; +import org.apache.thrift.transport.TTransport; + +/** + * Hadoop RecordReader read the values return from the CQL query + * It use CQL key range query to page through the wide rows. + *

+ * Return List as keys columns + *

+ * Map as column name to columns mappings + */ +public class ColumnFamilyRecordReader extends RecordReader, Map> + implements org.apache.hadoop.mapred.RecordReader, Map> +{ + private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyRecordReader.class); + + public static final int DEFAULT_CQL_PAGE_LIMIT = 1000; // TODO: find the number large enough but not OOM + + private ColumnFamilySplit split; + private RowIterator rowIterator; + + private Pair, Map> currentRow; + private int totalRowCount; // total number of rows to fetch + private String keyspace; + private String cfName; + private Cassandra.Client client; + private ConsistencyLevel consistencyLevel; + + // partition keys -- key aliases + private List partitionBoundColumns = new ArrayList(); + + // cluster keys -- column aliases + private List clusterColumns = new ArrayList(); + + // map prepared query type to item id + private Map preparedQueryIds = new HashMap(); + + // cql query select columns + private String columns; + + // the number of cql rows per page + private int pageRowSize; + + // user defined where clauses + private String userDefinedWhereClauses; + + private IPartitioner partitioner; + + private AbstractType keyValidator; + + public ColumnFamilyRecordReader() + { + super(); + } + + public void initialize(InputSplit split, TaskAttemptContext context) throws IOException + { + this.split = (ColumnFamilySplit) split; + Configuration conf = context.getConfiguration(); + totalRowCount = (this.split.getLength() < Long.MAX_VALUE) + ? (int) this.split.getLength() + : ConfigHelper.getInputSplitSize(conf); + cfName = ConfigHelper.getInputColumnFamily(conf); + consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getReadConsistencyLevel(conf)); + keyspace = ConfigHelper.getInputKeyspace(conf); + columns = CQLConfigHelper.getInputcolumns(conf); + userDefinedWhereClauses = CQLConfigHelper.getInputWhereClauses(conf); + + try + { + pageRowSize = Integer.parseInt(CQLConfigHelper.getInputPageRowSize(conf)); + } + catch (NumberFormatException e) + { + pageRowSize = DEFAULT_CQL_PAGE_LIMIT; + } + + partitioner = ConfigHelper.getInputPartitioner(context.getConfiguration()); + + try + { + if (client != null) + return; + + // create connection using thrift + String location = getLocation(); + + int port = ConfigHelper.getInputRpcPort(conf); + client = ColumnFamilyInputFormat.createAuthenticatedClient(location, port, conf); + + // retrieve partition keys and cluster keys from system.schema_columnfamilies table + retrieveKeys(); + + client.set_keyspace(keyspace); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + + rowIterator = new RowIterator(); + + logger.debug("created {}", rowIterator); + } + + public void close() + { + if (client != null) + { + TTransport transport = client.getOutputProtocol().getTransport(); + if (transport.isOpen()) + transport.close(); + client = null; + } + } + + public Map getCurrentKey() + { + return currentRow.left; + } + + public Map getCurrentValue() + { + return currentRow.right; + } + + public float getProgress() + { + if (!rowIterator.hasNext()) + return 1.0F; + + // the progress is likely to be reported slightly off the actual but close enough + float progress = ((float) rowIterator.totalRead / totalRowCount); + return progress > 1.0F ? 1.0F : progress; + } + + public boolean nextKeyValue() throws IOException + { + if (!rowIterator.hasNext()) + { + logger.debug("Finished scanning " + rowIterator.totalRead + " rows (estimate was: " + totalRowCount + ")"); + return false; + } + + try + { + currentRow = rowIterator.next(); + } + catch (Exception e) + { + // throw it as IOException, so client can catch it and handle it at client side + IOException ioe = new IOException(e.getMessage()); + ioe.initCause(ioe.getCause()); + throw ioe; + } + return true; + } + + // we don't use endpointsnitch since we are trying to support hadoop nodes that are + // not necessarily on Cassandra machines, too. This should be adequate for single-DC clusters, at least. + private String getLocation() + { + Collection localAddresses = FBUtilities.getAllLocalAddresses(); + + for (InetAddress address : localAddresses) + { + for (String location : split.getLocations()) + { + InetAddress locationAddress; + try + { + locationAddress = InetAddress.getByName(location); + } + catch (UnknownHostException e) + { + throw new AssertionError(e); + } + if (address.equals(locationAddress)) + { + return location; + } + } + } + return split.getLocations()[0]; + } + + // 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 + // to the old. Thus, expect a small performance hit. + // And obviously this wouldn't work for wide rows. But since ColumnFamilyInputFormat + // and ColumnFamilyRecordReader don't support them, it should be fine for now. + public boolean next(Map keys, Map value) throws IOException + { + if (nextKeyValue()) + { + value.clear(); + value.putAll(getCurrentValue()); + + keys.clear(); + keys.putAll(getCurrentKey()); + + return true; + } + return false; + } + + public long getPos() throws IOException + { + return (long) rowIterator.totalRead; + } + + public Map createKey() + { + return new LinkedHashMap(); + } + + public Map createValue() + { + return new LinkedHashMap(); + } + + /** CQL row iterator */ + private class RowIterator extends AbstractIterator, Map>> + { + protected int totalRead = 0; // total number of cf rows read + protected Iterator rows; + private int pageRows = 0; // the number of cql rows read of this page + private String previousRowKey = null; // previous CF row key + private String partitionKeyString; // keys in , , string format + private String partitionKeyMarkers; // question marks in ? , ? , ? format which matches the number of keys + + public RowIterator() + { + // initial page + executeQuery(); + } + + protected Pair, Map> computeNext() + { + if (rows == null) + return endOfData(); + + int index = -2; + //check there are more page to read + while (!rows.hasNext()) + { + // no more data + if (index == -1 || emptyPartitionKeyValues()) + { + logger.debug("no more data."); + return endOfData(); + } + + index = setTailNull(clusterColumns); + logger.debug("set tail to null, index: " + index); + executeQuery(); + pageRows = 0; + + if (rows == null || !rows.hasNext() && index < 0) + { + logger.debug("no more data."); + return endOfData(); + } + } + + Map valueColumns = createValue(); + Map keyColumns = createKey(); + int i = 0; + CqlRow row = rows.next(); + for (Column column : row.columns) + { + String columnName = stringValue(ByteBuffer.wrap(column.getName())); + logger.debug("column: " + columnName); + + if (i < partitionBoundColumns.size() + clusterColumns.size()) + keyColumns.put(stringValue(column.name), column.value); + else + valueColumns.put(stringValue(column.name), column.value); + + i++; + } + + // increase total CQL row read for this page + pageRows++; + + // increase total CF row read + if (newRow(keyColumns, previousRowKey)) + totalRead++; + + // read full page + if (pageRows >= pageRowSize || !rows.hasNext()) + { + Iterator newKeys = keyColumns.keySet().iterator(); + for (BoundColumn column : partitionBoundColumns) + column.value = keyColumns.get(newKeys.next()); + + for (BoundColumn column : clusterColumns) + column.value = keyColumns.get(newKeys.next()); + + executeQuery(); + pageRows = 0; + } + + return Pair.create(keyColumns, valueColumns); + } + + /** check whether start to read a new CF row by comparing the partition keys */ + private boolean newRow(Map keyColumns, String previousRowKey) + { + if (keyColumns.isEmpty()) + return false; + + String rowKey = ""; + if (keyColumns.size() == 1) + { + rowKey = partitionBoundColumns.get(0).validator.getString(keyColumns.get(partitionBoundColumns.get(0).name)); + } + else + { + Iterator iter = keyColumns.values().iterator(); + for (BoundColumn column : partitionBoundColumns) + rowKey = rowKey + column.validator.getString(ByteBufferUtil.clone(iter.next())) + ":"; + } + + logger.debug("previous RowKey: " + previousRowKey + ", new row key: " + rowKey); + if (previousRowKey == null) + { + this.previousRowKey = rowKey; + return true; + } + + if (rowKey.equals(previousRowKey)) + return false; + + this.previousRowKey = rowKey; + return true; + } + + /** set the last non-null key value to null, and return the previous index */ + private int setTailNull(List values) + { + if (values.isEmpty()) + return -1; + + Iterator iterator = values.iterator(); + int previousIndex = -1; + BoundColumn current; + while (iterator.hasNext()) + { + current = iterator.next(); + if (current.value == null) + { + int index = previousIndex > 0 ? previousIndex : 0; + BoundColumn column = values.get(index); + logger.debug("set key " + column.name + " value to null"); + column.value = null; + return previousIndex - 1; + } + + previousIndex++; + } + + BoundColumn column = values.get(previousIndex); + logger.debug("set key " + column.name + " value to null"); + column.value = null; + return previousIndex - 1; + } + + /** compose the prepared query, pair.left is query id, pair.right is query */ + private Pair composeQuery(String columns) + { + Pair clause = whereClause(); + if (columns == null) + { + columns = "*"; + } + else + { + // add keys in the front in order + String partitionKey = keyString(partitionBoundColumns); + String clusterKey = keyString(clusterColumns); + + columns = withoutKeyColumns(columns); + columns = (clusterKey == null || "".equals(clusterKey)) + ? partitionKey + "," + columns + : partitionKey + "," + clusterKey + "," + columns; + } + + return Pair.create(clause.left, + "SELECT " + columns + + " FROM " + cfName + + clause.right + + (userDefinedWhereClauses == null ? "" : " AND " + userDefinedWhereClauses) + + " LIMIT " + pageRowSize + + " ALLOW FILTERING"); + } + + + /** remove key columns from the column string */ + private String withoutKeyColumns(String columnString) + { + Set keyNames = new HashSet(); + for (BoundColumn column : Iterables.concat(partitionBoundColumns, clusterColumns)) + keyNames.add(column.name); + + String[] columns = columnString.split(","); + String result = null; + for (String column : columns) + { + String trimmed = column.trim(); + if (keyNames.contains(trimmed)) + continue; + + result = result == null ? trimmed : result + "," + trimmed; + } + return result; + } + + /** compose the where clause */ + private Pair whereClause() + { + if (partitionKeyString == null) + partitionKeyString = keyString(partitionBoundColumns); + + if (partitionKeyMarkers == null) + partitionKeyMarkers = partitionKeyMarkers(); + // initial query token(k) >= start_token and token(k) <= end_token + if (emptyPartitionKeyValues()) + return Pair.create(0, " WHERE token(" + partitionKeyString + ") > ? AND token(" + partitionKeyString + ") <= ?"); + + // query token(k) > token(pre_partition_key) and token(k) <= end_token + if (clusterColumns.size() == 0 || clusterColumns.get(0).value == null) + return Pair.create(1, + " WHERE token(" + partitionKeyString + ") > token(" + partitionKeyMarkers + ") " + + " AND token(" + partitionKeyString + ") <= ?"); + + // query token(k) = token(pre_partition_key) and m = pre_cluster_key_m and n > pre_cluster_key_n + Pair clause = whereClause(clusterColumns, 0); + return Pair.create(clause.left, + " WHERE token(" + partitionKeyString + ") = token(" + partitionKeyMarkers + ") " + clause.right); + } + + /** recursively compose the where clause */ + private Pair whereClause(List column, int position) + { + if (position == column.size() - 1 || column.get(position + 1).value == null) + return Pair.create(position + 2, " AND " + column.get(position).name + " > ? "); + + Pair clause = whereClause(column, position + 1); + return Pair.create(clause.left, " AND " + column.get(position).name + " = ? " + clause.right); + } + + /** check whether all key values are null */ + private boolean emptyPartitionKeyValues() + { + for (BoundColumn column : partitionBoundColumns) + { + if (column.value != null) + return false; + } + return true; + } + + /** compose the partition key string in format of , , */ + private String keyString(List columns) + { + String result = null; + for (BoundColumn column : columns) + result = result == null ? column.name : result + "," + column.name; + + return result == null ? "" : result; + } + + /** compose the question marks for partition key string in format of ?, ? , ? */ + private String partitionKeyMarkers() + { + String result = null; + for (BoundColumn column : partitionBoundColumns) + result = result == null ? "?" : result + ",?"; + + return result; + } + + /** compose the query binding variables, pair.left is query id, pair.right is the binding variables */ + private Pair> preparedQueryBindValues() + { + List values = new LinkedList(); + + // initial query token(k) >= start_token and token(k) <= end_token + if (emptyPartitionKeyValues()) + { + values.add(partitioner.getTokenValidator().fromString(split.getStartToken())); + values.add(partitioner.getTokenValidator().fromString(split.getEndToken())); + return Pair.create(0, values); + } + else + { + for (BoundColumn partitionBoundColumn1 : partitionBoundColumns) + values.add(partitionBoundColumn1.value); + + if (clusterColumns.size() == 0 || clusterColumns.get(0).value == null) + { + // query token(k) > token(pre_partition_key) and token(k) <= end_token + values.add(partitioner.getTokenValidator().fromString(split.getEndToken())); + return Pair.create(1, values); + } + else + { + // query token(k) = token(pre_partition_key) and m = pre_cluster_key_m and n > pre_cluster_key_n + int type = preparedQueryBindValues(clusterColumns, 0, values); + return Pair.create(type, values); + } + } + } + + /** recursively compose the query binding variables */ + private int preparedQueryBindValues(List column, int position, List bindValues) + { + if (position == column.size() - 1 || column.get(position + 1).value == null) + { + bindValues.add(column.get(position).value); + return position + 2; + } + else + { + bindValues.add(column.get(position).value); + return preparedQueryBindValues(column, position + 1, bindValues); + } + } + + /** get the prepared query item Id */ + private int prepareQuery(int type) throws InvalidRequestException, TException + { + Integer itemId = preparedQueryIds.get(type); + if (itemId != null) + return itemId; + + Pair query = null; + query = composeQuery(columns); + logger.debug("type:" + query.left + ", query: " + query.right); + CqlPreparedResult cqlPreparedResult = client.prepare_cql3_query(ByteBufferUtil.bytes(query.right), Compression.NONE); + preparedQueryIds.put(query.left, cqlPreparedResult.itemId); + return cqlPreparedResult.itemId; + } + + /** execute the prepared query */ + private void executeQuery() + { + Pair> bindValues = preparedQueryBindValues(); + logger.debug("query type: " + bindValues.left); + + // check whether it reach end of range for type 1 query CASSANDRA-5573 + if (bindValues.left == 1 && reachEndRange()) + { + rows = null; + return; + } + + int retries = 0; + // only try three times for TimedOutException and UnavailableException + while (retries < 3) + { + try + { + CqlResult cqlResult = client.execute_prepared_cql3_query(prepareQuery(bindValues.left), bindValues.right, consistencyLevel); + if (cqlResult != null && cqlResult.rows != null) + rows = cqlResult.rows.iterator(); + return; + } + catch (TimedOutException e) + { + retries++; + if (retries >= 3) + { + rows = null; + RuntimeException rte = new RuntimeException(e.getMessage()); + rte.initCause(e); + throw rte; + } + } + catch (UnavailableException e) + { + retries++; + if (retries >= 3) + { + rows = null; + RuntimeException rte = new RuntimeException(e.getMessage()); + rte.initCause(e); + throw rte; + } + } + catch (Exception e) + { + rows = null; + RuntimeException rte = new RuntimeException(e.getMessage()); + rte.initCause(e); + throw rte; + } + } + } + } + + /** retrieve the partition keys and cluster keys from system.schema_columnfamilies table */ + private void retrieveKeys() throws Exception + { + String query = "select key_aliases," + + "column_aliases, " + + "key_validator, " + + "comparator " + + "from system.schema_columnfamilies " + + "where keyspace_name='%s' and columnfamily_name='%s'"; + String formatted = String.format(query, keyspace, cfName); + CqlResult result = client.execute_cql3_query(ByteBufferUtil.bytes(formatted), Compression.NONE, ConsistencyLevel.ONE); + + CqlRow cqlRow = result.rows.get(0); + String keyString = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(0).getValue())); + logger.debug("partition keys: " + keyString); + List keys = FBUtilities.fromJsonList(keyString); + + for (String key : keys) + partitionBoundColumns.add(new BoundColumn(key)); + + keyString = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(1).getValue())); + logger.debug("cluster columns: " + keyString); + keys = FBUtilities.fromJsonList(keyString); + + for (String key : keys) + clusterColumns.add(new BoundColumn(key)); + + Column rawKeyValidator = cqlRow.columns.get(2); + String validator = ByteBufferUtil.string(ByteBuffer.wrap(rawKeyValidator.getValue())); + logger.debug("row key validator: " + validator); + keyValidator = parseType(validator); + + if (keyValidator instanceof CompositeType) + { + List> types = ((CompositeType) keyValidator).types; + for (int i = 0; i < partitionBoundColumns.size(); i++) + partitionBoundColumns.get(i).validator = types.get(i); + } + else + { + partitionBoundColumns.get(0).validator = keyValidator; + } + } + + /** check whether current row is at the end of range */ + private boolean reachEndRange() + { + // current row key + ByteBuffer rowKey; + if (keyValidator instanceof CompositeType) + { + ByteBuffer[] keys = new ByteBuffer[partitionBoundColumns.size()]; + for (int i = 0; i < partitionBoundColumns.size(); i++) + keys[i] = partitionBoundColumns.get(i).value.duplicate(); + + rowKey = ((CompositeType) keyValidator).build(keys); + } + else + { + rowKey = partitionBoundColumns.get(0).value; + } + + String endToken = split.getEndToken(); + String currentToken = partitioner.getToken(rowKey).toString(); + logger.debug("End token: " + endToken + ", current token: " + currentToken); + + return endToken.equals(currentToken); + } + + private static AbstractType parseType(String type) throws IOException + { + try + { + // always treat counters like longs, specifically CCT.compose is not what we need + if (type != null && type.equals("org.apache.cassandra.db.marshal.CounterColumnType")) + return LongType.instance; + return TypeParser.parse(type); + } + catch (ConfigurationException e) + { + throw new IOException(e); + } + catch (SyntaxException e) + { + throw new IOException(e); + } + } + + private class BoundColumn + { + final String name; + ByteBuffer value; + AbstractType validator; + + public BoundColumn(String name) + { + this.name = name; + } + } + + /** get string from a ByteBuffer, catch the exception and throw it as runtime exception*/ + private static String stringValue(ByteBuffer value) + { + try + { + return ByteBufferUtil.string(value); + } + catch (CharacterCodingException e) + { + throw new RuntimeException(e); + } + } +} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordWriter.java new file mode 100644 index 0000000000..3939e0b12a --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/cql3/ColumnFamilyRecordWriter.java @@ -0,0 +1,386 @@ +/* + * 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.cql3; + +import java.io.IOException; +import java.net.InetAddress; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.cassandra.thrift.*; +import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.db.marshal.CompositeType; +import org.apache.cassandra.db.marshal.LongType; +import org.apache.cassandra.db.marshal.TypeParser; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.exceptions.SyntaxException; +import org.apache.cassandra.hadoop.AbstractColumnFamilyRecordWriter; +import org.apache.cassandra.hadoop.ConfigHelper; +import org.apache.cassandra.hadoop.Progressable; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.thrift.TException; +import org.apache.thrift.transport.TTransport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The ColumnFamilyRecordWriter maps the output <key, value> + * pairs to a Cassandra column family. In particular, it applies the binded variables + * in the value to the prepared statement, which it associates with the key, and in + * turn the responsible endpoint. + * + *

+ * Furthermore, this writer groups the cql queries by the endpoint responsible for + * the rows being affected. This allows the cql queries to be executed in parallel, + * directly to a responsible endpoint. + *

+ * + * @see ColumnFamilyOutputFormat + */ +final class ColumnFamilyRecordWriter extends AbstractColumnFamilyRecordWriter, List> +{ + private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyRecordWriter.class); + + // handles for clients for each range running in the threadpool + private final Map clients; + + // host to prepared statement id mappings + private ConcurrentHashMap preparedStatements = new ConcurrentHashMap(); + + private final String cql; + + private AbstractType keyValidator; + private String [] partitionkeys; + + /** + * Upon construction, obtain the map that this writer will use to collect + * mutations, and the ring cache for the given keyspace. + * + * @param context the task attempt context + * @throws IOException + */ + ColumnFamilyRecordWriter(TaskAttemptContext context) throws IOException + { + this(context.getConfiguration()); + this.progressable = new Progressable(context); + } + + ColumnFamilyRecordWriter(Configuration conf, Progressable progressable) throws IOException + { + this(conf); + this.progressable = progressable; + } + + ColumnFamilyRecordWriter(Configuration conf) throws IOException + { + super(conf); + this.clients = new HashMap(); + cql = CQLConfigHelper.getOutputCql(conf); + + try + { + String host = getAnyHost(); + int port = ConfigHelper.getOutputRpcPort(conf); + Cassandra.Client client = ColumnFamilyOutputFormat.createAuthenticatedClient(host, port, conf); + retrievePartitionKeyValidator(client); + + if (client != null) + { + TTransport transport = client.getOutputProtocol().getTransport(); + if (transport.isOpen()) + transport.close(); + client = null; + } + } + catch (Exception e) + { + throw new IOException(e); + } + } + + @Override + public void close() throws IOException + { + // close all the clients before throwing anything + IOException clientException = null; + for (RangeClient client : clients.values()) + { + try + { + client.close(); + } + catch (IOException e) + { + clientException = e; + } + } + + if (clientException != null) + throw clientException; + } + + /** + * If the key is to be associated with a valid value, a mutation is created + * for it with the given column family and columns. In the event the value + * in the column is missing (i.e., null), then it is marked for + * {@link Deletion}. Similarly, if the entire value for a key is missing + * (i.e., null), then the entire key is marked for {@link Deletion}. + *

+ * + * @param keybuff + * the key to write. + * @param values + * the values to write. + * @throws IOException + */ + @Override + public void write(Map keys, List values) throws IOException + { + ByteBuffer rowKey = getRowKey(keys); + Range range = ringCache.getRange(rowKey); + + // get the client for the given range, or create a new one + RangeClient client = clients.get(range); + if (client == null) + { + // haven't seen keys for this range: create new client + client = new RangeClient(ringCache.getEndpoint(range)); + client.start(); + clients.put(range, client); + } + + client.put(Pair.create(rowKey, values)); + progressable.progress(); + } + + /** + * A client that runs in a threadpool and connects to the list of endpoints for a particular + * range. Binded variable values for keys in that range are sent to this client via a queue. + */ + public class RangeClient extends AbstractRangeClient> + { + /** + * Constructs an {@link RangeClient} for the given endpoints. + * @param endpoints the possible endpoints to execute the mutations on + */ + public RangeClient(List endpoints) + { + super(endpoints); + } + + /** + * Loops collecting cql binded variable values from the queue and sending to Cassandra + */ + public void run() + { + outer: + while (run || !queue.isEmpty()) + { + Pair> bindVariables; + try + { + bindVariables = queue.take(); + } + catch (InterruptedException e) + { + // re-check loop condition after interrupt + continue; + } + + Iterator iter = endpoints.iterator(); + while (true) + { + // send the mutation to the last-used endpoint. first time through, this will NPE harmlessly. + try + { + int i = 0; + int itemId = preparedStatement(client); + while (bindVariables != null) + { + client.execute_prepared_cql3_query(itemId, bindVariables.right, ConsistencyLevel.ONE); + i++; + + if (i >= batchThreshold) + break; + + bindVariables = queue.poll(); + } + + break; + } + catch (Exception e) + { + closeInternal(); + if (!iter.hasNext()) + { + lastException = new IOException(e); + break outer; + } + } + + // attempt to connect to a different endpoint + try + { + InetAddress address = iter.next(); + String host = address.getHostName(); + int port = ConfigHelper.getOutputRpcPort(conf); + client = ColumnFamilyOutputFormat.createAuthenticatedClient(host, port, conf); + } + catch (Exception e) + { + closeInternal(); + // TException means something unexpected went wrong to that endpoint, so + // we should try again to another. Other exceptions (auth or invalid request) are fatal. + if ((!(e instanceof TException)) || !iter.hasNext()) + { + lastException = new IOException(e); + break outer; + } + } + } + } + } + + /** get prepared statement id from cache, otherwise prepare it from Cassandra server*/ + private int preparedStatement(Cassandra.Client client) + { + Integer itemId = preparedStatements.get(client); + if (itemId == null) + { + CqlPreparedResult result; + try + { + result = client.prepare_cql3_query(ByteBufferUtil.bytes(cql), Compression.NONE); + } + catch (InvalidRequestException e) + { + throw new RuntimeException("failed to prepare cql query " + cql, e); + } + catch (TException e) + { + throw new RuntimeException("failed to prepare cql query " + cql, e); + } + + Integer previousId = preparedStatements.putIfAbsent(client, Integer.valueOf(result.itemId)); + itemId = previousId == null ? result.itemId : previousId; + } + return itemId; + } + } + + private ByteBuffer getRowKey(Map keysMap) + { + //current row key + ByteBuffer rowKey; + if (keyValidator instanceof CompositeType) + { + ByteBuffer[] keys = new ByteBuffer[partitionkeys.length]; + for (int i = 0; i< keys.length; i++) + keys[i] = keysMap.get(partitionkeys[i]); + + rowKey = ((CompositeType) keyValidator).build(keys); + } + else + { + rowKey = keysMap.get(partitionkeys[0]); + } + return rowKey; + } + + /** retrieve the key validator from system.schema_columnfamilies table */ + private void retrievePartitionKeyValidator(Cassandra.Client client) throws Exception + { + String keyspace = ConfigHelper.getOutputKeyspace(conf); + String cfName = ConfigHelper.getOutputColumnFamily(conf); + String query = "SELECT key_validator," + + " key_aliases " + + "FROM system.schema_columnfamilies " + + "WHERE keyspace_name='%s' and columnfamily_name='%s'"; + String formatted = String.format(query, keyspace, cfName); + CqlResult result = client.execute_cql3_query(ByteBufferUtil.bytes(formatted), Compression.NONE, ConsistencyLevel.ONE); + + Column rawKeyValidator = result.rows.get(0).columns.get(0); + String validator = ByteBufferUtil.string(ByteBuffer.wrap(rawKeyValidator.getValue())); + keyValidator = parseType(validator); + + Column rawPartitionKeys = result.rows.get(0).columns.get(1); + String keyString = ByteBufferUtil.string(ByteBuffer.wrap(rawPartitionKeys.getValue())); + logger.debug("partition keys: " + keyString); + + List keys = FBUtilities.fromJsonList(keyString); + partitionkeys = new String [keys.size()]; + int i=0; + for (String key: keys) + { + partitionkeys[i] = key; + i++; + } + } + + private AbstractType parseType(String type) throws IOException + { + try + { + // always treat counters like longs, specifically CCT.compose is not what we need + if (type != null && type.equals("org.apache.cassandra.db.marshal.CounterColumnType")) + return LongType.instance; + return TypeParser.parse(type); + } + catch (ConfigurationException e) + { + throw new IOException(e); + } + catch (SyntaxException e) + { + throw new IOException(e); + } + } + + private String getAnyHost() throws IOException, InvalidRequestException, TException + { + Cassandra.Client client = ConfigHelper.getClientFromOutputAddressList(conf); + List ring = client.describe_ring(ConfigHelper.getOutputKeyspace(conf)); + try + { + for (TokenRange range : ring) + return range.endpoints.get(0); + } + finally + { + if (client != null) + { + TTransport transport = client.getOutputProtocol().getTransport(); + if (transport.isOpen()) + transport.close(); + client = null; + } + } + throw new IOException("There are no endpoints"); + } + +} diff --git a/src/java/org/apache/cassandra/thrift/TClientTransportFactory.java b/src/java/org/apache/cassandra/thrift/TClientTransportFactory.java new file mode 100644 index 0000000000..0a73043f78 --- /dev/null +++ b/src/java/org/apache/cassandra/thrift/TClientTransportFactory.java @@ -0,0 +1,70 @@ +package org.apache.cassandra.thrift; + +/* + * + * 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. + * + */ + +import org.apache.hadoop.conf.Configuration; +import org.apache.thrift.transport.TTransport; + +import java.util.Map; +import java.util.Set; + + +/** + * Transport factory for establishing thrift connections from clients to a remote server. + */ +public interface TClientTransportFactory +{ + static final String PROPERTY_KEY = "cassandra.client.transport.factory"; + static final String LONG_OPTION = "transport-factory"; + static final String SHORT_OPTION = "tr"; + + /** + * Opens a client transport to a thrift server. + * Example: + * + * 
+     * TTransport transport = clientTransportFactory.openTransport(address, port);
+     * Cassandra.Iface client = new Cassandra.Client(new BinaryProtocol(transport));
+     *
+ * + * @param host fully qualified hostname of the server + * @param port RPC port of the server + * @param conf Hadoop configuration + * @return open and ready to use transport + * @throws Exception implementation defined; usually throws TTransportException or IOException + * if the connection cannot be established + */ + TTransport openTransport(String host, int port, Configuration conf) throws Exception; + + /** + * Sets an implementation defined set of options. + * Keys in this map must conform to the set set returned by TClientTransportFactory#supportedOptions. + * @param options option map + */ + void setOptions(Map options); + + /** + * @return set of options supported by this transport factory implementation + */ + Set supportedOptions(); +} + diff --git a/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java b/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java index 792618d5f1..5905f4aebe 100644 --- a/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java +++ b/src/java/org/apache/cassandra/thrift/TFramedTransportFactory.java @@ -21,6 +21,10 @@ package org.apache.cassandra.thrift; * */ +import java.util.Collections; +import java.util.Map; +import java.util.Set; + import org.apache.cassandra.hadoop.ConfigHelper; import org.apache.thrift.transport.TFramedTransport; import org.apache.thrift.transport.TSocket; @@ -29,7 +33,7 @@ import org.apache.thrift.transport.TTransportException; import org.apache.hadoop.conf.Configuration; -public class TFramedTransportFactory implements ITransportFactory +public class TFramedTransportFactory implements ITransportFactory, TClientTransportFactory { public TTransport openTransport(TSocket socket, Configuration conf) throws TTransportException { @@ -37,4 +41,22 @@ public class TFramedTransportFactory implements ITransportFactory transport.open(); return transport; } + + // + public TTransport openTransport(String host, int port, Configuration conf) throws TTransportException + { + TSocket socket = new TSocket(host, port); + TTransport transport = new TFramedTransport(socket, ConfigHelper.getThriftFramedTransportSize(conf)); + transport.open(); + return transport; + } + + public void setOptions(Map options) + { + } + + public Set supportedOptions() + { + return Collections.emptySet(); + } }