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

Remove deprecated legacy Hadoop code 

patch by Philip Thompson; reviewed by Aleksey Yeschenko for
CASSANDRA-9353
This commit is contained in:
Philip Thompson 2015-06-08 22:41:28 +03:00 committed by Aleksey Yeschenko
parent d62cd1bc9e
commit 446e253789
27 changed files with 451 additions and 4597 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0:
* Remove deprecated legacy Hadoop code (CASSANDRA-9353)
* Decommissioned nodes will not rejoin the cluster (CASSANDRA-8801)
* Change gossip stabilization to use endpoit size (CASSANDRA-9401)
* Change default garbage collector to G1 (CASSANDRA-7486)

14
NEWS.txt
View File

@ -13,6 +13,20 @@ restore snapshots created with the previous major version using the
'sstableloader' tool. You can upgrade the file format of your snapshots
using the provided 'sstableupgrade' tool.
3.0
===
Upgrading
---------
- Pig's CassandraStorage has been removed. Use CqlNativeStorage instead.
- Hadoop BulkOutputFormat and BulkRecordWriter have been removed; use
CqlBulkOutputFormat and CqlBulkRecordWriter instead.
- Hadoop ColumnFamilyInputFormat and ColumnFamilyOutputFormat have been removed;
use CqlInputFormat and CqlOutputFormat instead.
- Hadoop ColumnFamilyRecordReader and ColumnFamilyRecordWriter have been removed;
use CqlRecordReader and CqlRecordWriter instead.
2.2
===

50
examples/hadoop_word_count/README.txt
View File

@ -1,50 +0,0 @@
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 examples/word_count.
Running
=======
First build and start a Cassandra server with the default configuration*. Ensure that the Thrift
interface is enabled, either by setting start_rpc:true in cassandra.yaml or by running
`nodetool enablethrift` after startup.
Once Cassandra has started and the Thrift interface is available, 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/cqlsh and
perform the following operations:
$ bin/cqlsh localhost
> use wordcount;
> 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 'wordcount' 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.
*It is recommended to turn off vnodes when running Cassandra with hadoop.
This is done by setting "num_tokens: 1" in cassandra.yaml. If you want to
point wordcount at a real cluster, modify the seed and listenaddress
settings accordingly.
Troubleshooting
===============
word_count uses conf/logback.xml to log to wc.out.

61
examples/hadoop_word_count/bin/word_count
View File

@ -1,61 +0,0 @@
#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
cwd=`dirname $0`
# 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

59
examples/hadoop_word_count/bin/word_count_counters
View File

@ -1,59 +0,0 @@
#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
cwd=`dirname $0`
# 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
#echo $CLASSPATH
"$JAVA" -Xmx1G -ea -cp "$CLASSPATH" WordCountCounters

61
examples/hadoop_word_count/bin/word_count_setup
View File

@ -1,61 +0,0 @@
#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
cwd=`dirname $0`
# 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

113
examples/hadoop_word_count/build.xml
View File

@ -1,113 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
-->
<project default="jar" name="word_count" xmlns:ivy="antlib:org.apache.ivy.ant">
<property name="cassandra.dir" value="../.." />
<property name="cassandra.dir.lib" value="${cassandra.dir}/lib" />
<property name="cassandra.classes" value="${cassandra.dir}/build/classes" />
<property name="build.src" value="${basedir}/src" />
<property name="build.dir" value="${basedir}/build" />
<property name="ivy.lib.dir" value="${build.dir}/lib" />
<property name="build.classes" value="${build.dir}/classes" />
<property name="final.name" value="word_count" />
<property name="ivy.version" value="2.1.0" />
<property name="ivy.url"
value="http://repo2.maven.org/maven2/org/apache/ivy/ivy" />
<condition property="ivy.jar.exists">
<available file="${build.dir}/ivy-${ivy.version}.jar" />
</condition>
<path id="autoivy.classpath">
<fileset dir="${ivy.lib.dir}">
<include name="**/*.jar" />
</fileset>
<pathelement location="${build.dir}/ivy-${ivy.version}.jar"/>
</path>
<path id="wordcount.build.classpath">
<fileset dir="${ivy.lib.dir}">
<include name="**/*.jar" />
</fileset>
<!-- cassandra dependencies -->
<fileset dir="${cassandra.dir.lib}">
<include name="**/*.jar" />
</fileset>
<fileset dir="${cassandra.dir}/build/lib/jars">
<include name="**/*.jar" />
</fileset>
<pathelement location="${cassandra.classes}/main" />
<pathelement location="${cassandra.classes}/thrift" />
</path>
<target name="init">
<mkdir dir="${build.classes}" />
</target>
<target depends="init,ivy-retrieve-build" name="build">
<javac destdir="${build.classes}">
<src path="${build.src}" />
<classpath refid="wordcount.build.classpath" />
</javac>
</target>
<target name="jar" depends="build">
<mkdir dir="${build.classes}/META-INF" />
<jar jarfile="${build.dir}/${final.name}.jar">
<fileset dir="${build.classes}" />
<fileset dir="${cassandra.classes}/main" />
<fileset dir="${cassandra.classes}/thrift" />
<fileset dir="${cassandra.dir}">
<include name="lib/**/*.jar" />
</fileset>
<zipfileset dir="${cassandra.dir}/build/lib/jars/" prefix="lib">
<include name="**/*.jar" />
</zipfileset>
<fileset file="${basedir}/cassandra.yaml" />
</jar>
</target>
<target name="clean">
<delete dir="${build.dir}" />
</target>
<!--
Ivy Specific targets
to fetch Ivy and this project's dependencies
-->
<target name="ivy-download" unless="ivy.jar.exists">
<echo>Downloading Ivy...</echo>
<mkdir dir="${build.dir}" />
<get src="${ivy.url}/${ivy.version}/ivy-${ivy.version}.jar"
dest="${build.dir}/ivy-${ivy.version}.jar" usetimestamp="true" />
</target>
<target name="ivy-init" depends="ivy-download" unless="ivy.initialized">
<mkdir dir="${ivy.lib.dir}"/>
<taskdef resource="org/apache/ivy/ant/antlib.xml"
uri="antlib:org.apache.ivy.ant"
classpathref="autoivy.classpath"/>
<property name="ivy.initialized" value="true"/>
</target>
<target name="ivy-retrieve-build" depends="ivy-init">
<ivy:retrieve type="jar,source" sync="true"
pattern="${ivy.lib.dir}/[type]s/[artifact]-[revision].[ext]" />
</target>
</project>

42
examples/hadoop_word_count/conf/logback.xml
View File

@ -1,42 +0,0 @@
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<configuration scan="true">
<jmxConfigurator />
<appender name="FILE" class="ch.qos.logback.core.FileAppender">
<file>wc.out</file>
<encoder>
<pattern>%-5level [%thread] %date{ISO8601} %F:%L - %msg%n</pattern>
</encoder>
</appender>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
</encoder>
</appender>
<root level="INFO">
<appender-ref ref="FILE" />
<appender-ref ref="STDOUT" />
</root>
</configuration>

24
examples/hadoop_word_count/ivy.xml
View File

@ -1,24 +0,0 @@
<!--
~ Licensed to the Apache Software Foundation (ASF) under one
~ or more contributor license agreements. See the NOTICE file
~ distributed with this work for additional information
~ regarding copyright ownership. The ASF licenses this file
~ to you under the Apache License, Version 2.0 (the
~ "License"); you may not use this file except in compliance
~ with the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing,
~ software distributed under the License is distributed on an
~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
~ KIND, either express or implied. See the License for the
~ specific language governing permissions and limitations
~ under the License.
-->
<ivy-module version="2.0">
<info organisation="apache-cassandra" module="word-count"/>
<dependencies>
<dependency org="org.apache.hadoop" name="hadoop-core" rev="1.0.3"/>
</dependencies>
</ivy-module>

222
examples/hadoop_word_count/src/WordCount.java
View File

@ -1,222 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.hadoop.*;
import org.apache.cassandra.thrift.*;
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.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
* This counts the occurrences of words in ColumnFamily Standard1, that has a single column (that we care about)
* "text" containing a sequence of words.
*
* For each word, we output the total number of occurrences across all texts.
*
* When outputting to Cassandra, we write the word counts as a {word, count} column/value pair,
* with a row key equal to the name of the source column we read the words from.
*/
public class WordCount extends Configured implements Tool
{
private static final Logger logger = LoggerFactory.getLogger(WordCount.class);
static final String KEYSPACE = "wordcount";
static final String COLUMN_FAMILY = "input_words";
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 CONF_COLUMN_NAME = "columnname";
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<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>, 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(ByteBuffer key, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column> columns, Context context) throws IOException, InterruptedException
{
for (ColumnFamilyRecordReader.Column column : columns.values())
{
String name = ByteBufferUtil.string(column.name);
String value = null;
if (name.contains("int"))
value = String.valueOf(ByteBufferUtil.toInt(column.value));
else
value = ByteBufferUtil.string(column.value);
logger.debug("read {}:{}={} from {}",
new Object[] {ByteBufferUtil.string(key), name, value, context.getInputSplit()});
StringTokenizer itr = new StringTokenizer(value);
while (itr.hasMoreTokens())
{
word.set(itr.nextToken());
context.write(word, one);
}
}
}
}
public static class ReducerToFilesystem extends Reducer<Text, IntWritable, Text, IntWritable>
{
public void reduce(Text key, Iterable<IntWritable> 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<Text, IntWritable, ByteBuffer, List<Mutation>>
{
private ByteBuffer outputKey;
protected void setup(org.apache.hadoop.mapreduce.Reducer.Context context)
throws IOException, InterruptedException
{
outputKey = ByteBufferUtil.bytes(context.getConfiguration().get(CONF_COLUMN_NAME));
}
public void reduce(Text word, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
{
int sum = 0;
for (IntWritable val : values)
sum += val.get();
context.write(outputKey, Collections.singletonList(getMutation(word, sum)));
}
private static Mutation getMutation(Text word, int sum)
{
org.apache.cassandra.thrift.Column c = new org.apache.cassandra.thrift.Column();
c.setName(Arrays.copyOf(word.getBytes(), word.getLength()));
c.setValue(ByteBufferUtil.bytes(sum));
c.setTimestamp(System.currentTimeMillis());
Mutation m = new Mutation();
m.setColumn_or_supercolumn(new ColumnOrSuperColumn());
m.column_or_supercolumn.setColumn(c);
return m;
}
}
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);
// use a smaller page size that doesn't divide the row count evenly to exercise the paging logic better
ConfigHelper.setRangeBatchSize(getConf(), 99);
for (int i = 0; i < WordCountSetup.TEST_COUNT; i++)
{
String columnName = "text" + i;
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 + i));
}
else
{
job.setReducerClass(ReducerToCassandra.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(ByteBuffer.class);
job.setOutputValueClass(List.class);
job.setOutputFormatClass(ColumnFamilyOutputFormat.class);
ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE, OUTPUT_COLUMN_FAMILY);
job.getConfiguration().set(CONF_COLUMN_NAME, "sum");
}
job.setInputFormatClass(ColumnFamilyInputFormat.class);
ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(ByteBufferUtil.bytes(columnName)));
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
if (i == 4)
{
IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("int4"), IndexOperator.EQ, ByteBufferUtil.bytes(0));
ConfigHelper.setInputRange(job.getConfiguration(), Arrays.asList(expr));
}
if (i == 5)
{
// this will cause the predicate to be ignored in favor of scanning everything as a wide row
ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY, true);
}
ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
job.waitForCompletion(true);
}
return 0;
}
}

104
examples/hadoop_word_count/src/WordCountCounters.java
View File

@ -1,104 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.SortedMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.ColumnFamilyRecordReader;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
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.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
* This sums the word count stored in the input_words_count ColumnFamily for the key "key-if-verse1".
*
* 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<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>, Text, LongWritable>
{
public void map(ByteBuffer key, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column> columns, Context context) throws IOException, InterruptedException
{
long sum = 0;
for (ColumnFamilyRecordReader.Column column : columns.values())
{
logger.debug("read " + key + ":" + ByteBufferUtil.string(column.name) + " from " + context.getInputSplit());
sum += ByteBufferUtil.toLong(column.value);
}
context.write(new Text(ByteBufferUtil.string(key)), new LongWritable(sum));
}
}
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(), "org.apache.cassandra.dht.Murmur3Partitioner");
ConfigHelper.setInputColumnFamily(job.getConfiguration(), WordCount.KEYSPACE, WordCountCounters.COUNTER_COLUMN_FAMILY);
SlicePredicate predicate = new SlicePredicate().setSlice_range(
new SliceRange().
setStart(ByteBufferUtil.EMPTY_BYTE_BUFFER).
setFinish(ByteBufferUtil.EMPTY_BYTE_BUFFER).
setCount(100));
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
job.waitForCompletion(true);
return 0;
}
}

239
examples/hadoop_word_count/src/WordCountSetup.java
View File

@ -1,239 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.TimeUnit;
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;
import com.google.common.util.concurrent.Uninterruptibles;
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);
Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap;
Column c;
// text0: no rows
// text1: 1 row, 1 word
c = new Column()
.setName(ByteBufferUtil.bytes("text1"))
.setValue(ByteBufferUtil.bytes("word1"))
.setTimestamp(System.currentTimeMillis());
mutationMap = getMutationMap(ByteBufferUtil.bytes("key0"), WordCount.COLUMN_FAMILY, c);
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
logger.info("added text1");
// text1: 1 row, 2 word
c = new Column()
.setName(ByteBufferUtil.bytes("text2"))
.setValue(ByteBufferUtil.bytes("word1 word2"))
.setTimestamp(System.currentTimeMillis());
mutationMap = getMutationMap(ByteBufferUtil.bytes("key0"), WordCount.COLUMN_FAMILY, c);
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
logger.info("added text2");
// text3: 1000 rows, 1 word
mutationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
for (int i = 0; i < 1000; i++)
{
c = new Column()
.setName(ByteBufferUtil.bytes("text3"))
.setValue(ByteBufferUtil.bytes("word1"))
.setTimestamp(System.currentTimeMillis());
addToMutationMap(mutationMap, ByteBufferUtil.bytes("key" + i), WordCount.COLUMN_FAMILY, c);
}
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
logger.info("added text3");
// text4: 1000 rows, 1 word, one column to filter on
mutationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
for (int i = 0; i < 1000; i++)
{
Column c1 = new Column()
.setName(ByteBufferUtil.bytes("text4"))
.setValue(ByteBufferUtil.bytes("word1"))
.setTimestamp(System.currentTimeMillis());
Column c2 = new Column()
.setName(ByteBufferUtil.bytes("int4"))
.setValue(ByteBufferUtil.bytes(i % 4))
.setTimestamp(System.currentTimeMillis());
ByteBuffer key = ByteBufferUtil.bytes("key" + i);
addToMutationMap(mutationMap, key, WordCount.COLUMN_FAMILY, c1);
addToMutationMap(mutationMap, key, WordCount.COLUMN_FAMILY, c2);
}
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
logger.info("added text4");
// sentence data for the counters
final ByteBuffer key = ByteBufferUtil.bytes("key-if-verse1");
final ColumnParent colParent = new ColumnParent(WordCountCounters.COUNTER_COLUMN_FAMILY);
for (String sentence : sentenceData())
{
client.add(key,
colParent,
new CounterColumn(ByteBufferUtil.bytes(sentence),
(long) sentence.split("\\s").length),
ConsistencyLevel.ONE);
}
logger.info("added key-if-verse1");
System.exit(0);
}
private static Map<ByteBuffer, Map<String, List<Mutation>>> getMutationMap(ByteBuffer key, String cf, Column c)
{
Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
addToMutationMap(mutationMap, key, cf, c);
return mutationMap;
}
private static void addToMutationMap(Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap, ByteBuffer key, String cf, Column c)
{
Map<String, List<Mutation>> cfMutation = mutationMap.get(key);
if (cfMutation == null)
{
cfMutation = new HashMap<String, List<Mutation>>();
mutationMap.put(key, cfMutation);
}
List<Mutation> mutationList = cfMutation.get(cf);
if (mutationList == null)
{
mutationList = new ArrayList<Mutation>();
cfMutation.put(cf, mutationList);
}
ColumnOrSuperColumn cc = new ColumnOrSuperColumn();
Mutation m = new Mutation();
cc.setColumn(c);
m.setColumn_or_supercolumn(cc);
mutationList.add(m);
}
private static void setupKeyspace(Cassandra.Iface client) throws TException, InvalidRequestException, SchemaDisagreementException
{
List<CfDef> cfDefList = new ArrayList<CfDef>();
CfDef input = new CfDef(WordCount.KEYSPACE, WordCount.COLUMN_FAMILY);
input.setComparator_type("AsciiType");
input.setColumn_metadata(Arrays.asList(new ColumnDef(ByteBufferUtil.bytes("text1"), "AsciiType"),
new ColumnDef(ByteBufferUtil.bytes("text2"), "AsciiType"),
new ColumnDef(ByteBufferUtil.bytes("text3"), "AsciiType"),
new ColumnDef(ByteBufferUtil.bytes("text4"), "AsciiType"),
new ColumnDef(ByteBufferUtil.bytes("int4"), "Int32Type").setIndex_name("int4idx").setIndex_type(IndexType.KEYS)));
cfDefList.add(input);
CfDef output = new CfDef(WordCount.KEYSPACE, WordCount.OUTPUT_COLUMN_FAMILY);
output.setComparator_type("AsciiType");
output.setDefault_validation_class("Int32Type");
cfDefList.add(output);
CfDef counterInput = new CfDef(WordCount.KEYSPACE, WordCountCounters.COUNTER_COLUMN_FAMILY);
counterInput.setComparator_type("UTF8Type");
counterInput.setDefault_validation_class("CounterColumnType");
cfDefList.add(counterInput);
KsDef ksDef = new KsDef(WordCount.KEYSPACE, "org.apache.cassandra.locator.SimpleStrategy", cfDefList);
ksDef.putToStrategy_options("replication_factor", "1");
client.system_add_keyspace(ksDef);
int magnitude = getNumberOfHosts(client);
Uninterruptibles.sleepUninterruptibly(magnitude, TimeUnit.SECONDS);
}
private static int getNumberOfHosts(Cassandra.Iface client)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
client.set_keyspace("system");
SlicePredicate predicate = new SlicePredicate();
SliceRange sliceRange = new SliceRange();
sliceRange.setStart(new byte[0]);
sliceRange.setFinish(new byte[0]);
predicate.setSlice_range(sliceRange);
KeyRange keyrRange = new KeyRange();
keyrRange.setStart_key(new byte[0]);
keyrRange.setEnd_key(new byte[0]);
//keyrRange.setCount(100);
ColumnParent parent = new ColumnParent("peers");
List<KeySlice> ls = client.get_range_slices(parent, predicate, keyrRange, ConsistencyLevel.ONE);
return ls.size();
}
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 String[] sentenceData()
{ // 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,",
"Or being lied about, don’t deal in lies,",
"Or being hated, don’t give way to hating,",
"And yet don’t look too good, nor talk too wise;"
};
}
}

296
src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
View File

@ -1,296 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.hadoop;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TokenRange;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.OrderPreservingPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.hadoop.cql3.*;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.*;
public abstract class AbstractColumnFamilyInputFormat<K, Y> extends InputFormat<K, Y> implements org.apache.hadoop.mapred.InputFormat<K, Y>
{
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 (instantiated 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 Session session;
protected void validateConfiguration(Configuration conf)
{
if (ConfigHelper.getInputKeyspace(conf) == null || ConfigHelper.getInputColumnFamily(conf) == null)
{
throw new UnsupportedOperationException("you must set the keyspace and table 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 List<InputSplit> getSplits(JobContext context) throws IOException
{
Configuration conf = HadoopCompat.getConfiguration(context);
validateConfiguration(conf);
keyspace = ConfigHelper.getInputKeyspace(conf);
cfName = ConfigHelper.getInputColumnFamily(conf);
partitioner = ConfigHelper.getInputPartitioner(conf);
logger.debug("partitioner is {}", partitioner);
// canonical ranges and nodes holding replicas
Map<TokenRange, Set<Host>> masterRangeNodes = getRangeMap(conf, keyspace);
// canonical ranges, split into pieces, fetching the splits in parallel
ExecutorService executor = new ThreadPoolExecutor(0, 128, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
List<InputSplit> splits = new ArrayList<>();
try
{
List<Future<List<InputSplit>>> splitfutures = new ArrayList<>();
KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf);
Range<Token> jobRange = null;
if (jobKeyRange != null)
{
if (jobKeyRange.start_key != null)
{
if (!partitioner.preservesOrder())
throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving partitioner");
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));
}
else if (jobKeyRange.start_token != null)
{
jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.start_token),
partitioner.getTokenFactory().fromString(jobKeyRange.end_token));
}
else
{
logger.warn("ignoring jobKeyRange specified without start_key or start_token");
}
}
session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect();
Metadata metadata = session.getCluster().getMetadata();
for (TokenRange range : masterRangeNodes.keySet())
{
if (jobRange == null)
{
// for each tokenRange, pick a live owner and ask it to compute bite-sized splits
splitfutures.add(executor.submit(new SplitCallable(range, masterRangeNodes.get(range), conf)));
}
else
{
TokenRange jobTokenRange = rangeToTokenRange(metadata, jobRange);
if (range.intersects(jobTokenRange))
{
for (TokenRange intersection: range.intersectWith(jobTokenRange))
{
// for each tokenRange, pick a live owner and ask it to compute bite-sized splits
splitfutures.add(executor.submit(new SplitCallable(intersection, masterRangeNodes.get(range), conf)));
}
}
}
}
// wait until we have all the results back
for (Future<List<InputSplit>> 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;
}
private TokenRange rangeToTokenRange(Metadata metadata, Range<Token> range)
{
return metadata.newTokenRange(metadata.newToken(partitioner.getTokenFactory().toString(range.left)),
metadata.newToken(partitioner.getTokenFactory().toString(range.right)));
}
/**
* Gets a token tokenRange and splits it up according to the suggested
* size into input splits that Hadoop can use.
*/
class SplitCallable implements Callable<List<InputSplit>>
{
private final TokenRange tokenRange;
private final Set<Host> hosts;
private final Configuration conf;
public SplitCallable(TokenRange tr, Set<Host> hosts, Configuration conf)
{
this.tokenRange = tr;
this.hosts = hosts;
this.conf = conf;
}
public List<InputSplit> call() throws Exception
{
ArrayList<InputSplit> splits = new ArrayList<>();
Map<TokenRange, Long> subSplits;
subSplits = getSubSplits(keyspace, cfName, tokenRange, conf);
// turn the sub-ranges into InputSplits
String[] endpoints = new String[hosts.size()];
// hadoop needs hostname, not ip
int endpointIndex = 0;
for (Host endpoint : hosts)
endpoints[endpointIndex++] = endpoint.getAddress().getHostName();
boolean partitionerIsOpp = partitioner instanceof OrderPreservingPartitioner || partitioner instanceof ByteOrderedPartitioner;
for (TokenRange subSplit : subSplits.keySet())
{
List<TokenRange> ranges = subSplit.unwrap();
for (TokenRange subrange : ranges)
{
ColumnFamilySplit split =
new ColumnFamilySplit(
partitionerIsOpp ?
subrange.getStart().toString().substring(2) : subrange.getStart().toString(),
partitionerIsOpp ?
subrange.getEnd().toString().substring(2) : subrange.getStart().toString(),
subSplits.get(subSplit),
endpoints);
logger.debug("adding {}", split);
splits.add(split);
}
}
return splits;
}
}
private Map<TokenRange, Long> getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException
{
int splitSize = ConfigHelper.getInputSplitSize(conf);
try
{
return describeSplits(keyspace, cfName, range, splitSize);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
private Map<TokenRange, Set<Host>> getRangeMap(Configuration conf, String keyspace)
{
try (Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect())
{
Map<TokenRange, Set<Host>> map = new HashMap<>();
Metadata metadata = session.getCluster().getMetadata();
for (TokenRange tokenRange : metadata.getTokenRanges())
map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange));
return map;
}
}
private Map<TokenRange, Long> describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize)
{
String query = String.format("SELECT mean_partition_size, partitions_count " +
"FROM %s.%s " +
"WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?",
SystemKeyspace.NAME,
SystemKeyspace.SIZE_ESTIMATES);
ResultSet resultSet = session.execute(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString());
Row row = resultSet.one();
// If we have no data on this split, return the full split i.e., do not sub-split
// Assume smallest granularity of partition count available from CASSANDRA-7688
if (row == null)
{
Map<TokenRange, Long> wrappedTokenRange = new HashMap<>();
wrappedTokenRange.put(tokenRange, (long) 128);
return wrappedTokenRange;
}
long meanPartitionSize = row.getLong("mean_partition_size");
long partitionCount = row.getLong("partitions_count");
int splitCount = (int)((meanPartitionSize * partitionCount) / splitSize);
List<TokenRange> splitRanges = tokenRange.splitEvenly(splitCount);
Map<TokenRange, Long> rangesWithLength = new HashMap<>();
for (TokenRange range : splitRanges)
rangesWithLength.put(range, partitionCount/splitCount);
return rangesWithLength;
}
// Old Hadoop API
public org.apache.hadoop.mapred.InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException
{
TaskAttemptContext tac = HadoopCompat.newTaskAttemptContext(jobConf, new TaskAttemptID());
List<org.apache.hadoop.mapreduce.InputSplit> 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;
}
}

91
src/java/org/apache/cassandra/hadoop/BulkOutputFormat.java
View File

@ -1,91 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.hadoop;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.thrift.Mutation;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.*;
@Deprecated
public class BulkOutputFormat extends OutputFormat<ByteBuffer,List<Mutation>>
implements org.apache.hadoop.mapred.OutputFormat<ByteBuffer,List<Mutation>>
{
/** Fills the deprecated OutputFormat interface for streaming. */
@Deprecated
public BulkRecordWriter 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 BulkRecordWriter(job, progress);
}
@Override
public BulkRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
{
return new BulkRecordWriter(context);
}
@Override
public void checkOutputSpecs(JobContext context)
{
checkOutputSpecs(HadoopCompat.getConfiguration(context));
}
private void checkOutputSpecs(Configuration conf)
{
if (ConfigHelper.getOutputKeyspace(conf) == null)
{
throw new UnsupportedOperationException("you must set the keyspace with setColumnFamily()");
}
}
@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);
}
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) { }
}
}

296
src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
View File

@ -1,296 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.hadoop;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.io.sstable.SSTableSimpleUnsortedWriter;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.CounterColumn;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.utils.NativeSSTableLoaderClient;
import org.apache.cassandra.utils.OutputHandler;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.util.Progressable;
@Deprecated
public final class BulkRecordWriter extends RecordWriter<ByteBuffer, List<Mutation>>
implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer, List<Mutation>>
{
public final static String OUTPUT_LOCATION = "mapreduce.output.bulkoutputformat.localdir";
public final static String BUFFER_SIZE_IN_MB = "mapreduce.output.bulkoutputformat.buffersize";
public final static String STREAM_THROTTLE_MBITS = "mapreduce.output.bulkoutputformat.streamthrottlembits";
public final static String MAX_FAILED_HOSTS = "mapreduce.output.bulkoutputformat.maxfailedhosts";
private final Logger logger = LoggerFactory.getLogger(BulkRecordWriter.class);
protected final Configuration conf;
protected final int maxFailures;
protected final int bufferSize;
protected Closeable writer;
protected SSTableLoader loader;
protected Progressable progress;
protected TaskAttemptContext context;
private File outputDir;
private enum CFType
{
NORMAL,
SUPER,
}
private enum ColType
{
NORMAL,
COUNTER
}
private CFType cfType;
private ColType colType;
BulkRecordWriter(TaskAttemptContext context)
{
this(HadoopCompat.getConfiguration(context));
this.context = context;
}
BulkRecordWriter(Configuration conf, Progressable progress)
{
this(conf);
this.progress = progress;
}
BulkRecordWriter(Configuration conf)
{
Config.setOutboundBindAny(true);
this.conf = conf;
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(Integer.parseInt(conf.get(STREAM_THROTTLE_MBITS, "0")));
maxFailures = Integer.parseInt(conf.get(MAX_FAILED_HOSTS, "0"));
bufferSize = Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64"));
}
protected String getOutputLocation() throws IOException
{
String dir = conf.get(OUTPUT_LOCATION, System.getProperty("java.io.tmpdir"));
if (dir == null)
throw new IOException("Output directory not defined, if hadoop is not setting java.io.tmpdir then define " + OUTPUT_LOCATION);
return dir;
}
private void setTypes(Mutation mutation)
{
if (cfType == null)
{
if (mutation.getColumn_or_supercolumn().isSetSuper_column() || mutation.getColumn_or_supercolumn().isSetCounter_super_column())
cfType = CFType.SUPER;
else
cfType = CFType.NORMAL;
if (mutation.getColumn_or_supercolumn().isSetCounter_column() || mutation.getColumn_or_supercolumn().isSetCounter_super_column())
colType = ColType.COUNTER;
else
colType = ColType.NORMAL;
}
}
private void prepareWriter() throws IOException
{
if (outputDir == null)
{
String keyspace = ConfigHelper.getOutputKeyspace(conf);
//dir must be named by ks/cf for the loader
outputDir = new File(getOutputLocation() + File.separator + keyspace + File.separator + ConfigHelper.getOutputColumnFamily(conf));
outputDir.mkdirs();
}
if (writer == null)
{
AbstractType<?> subcomparator = null;
if (cfType == CFType.SUPER)
subcomparator = BytesType.instance;
writer = new SSTableSimpleUnsortedWriter(
outputDir,
ConfigHelper.getOutputPartitioner(conf),
ConfigHelper.getOutputKeyspace(conf),
ConfigHelper.getOutputColumnFamily(conf),
BytesType.instance,
subcomparator,
Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64")),
ConfigHelper.getOutputCompressionParamaters(conf));
this.loader = new SSTableLoader(outputDir, new ExternalClient(conf), new NullOutputHandler());
}
}
@Override
public void close(TaskAttemptContext context) throws IOException
{
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
{
if (writer != null)
{
writer.close();
Future<StreamState> future = loader.stream();
while (true)
{
try
{
future.get(1000, TimeUnit.MILLISECONDS);
break;
}
catch (ExecutionException | TimeoutException te)
{
if (null != progress)
progress.progress();
if (null != context)
HadoopCompat.progress(context);
}
catch (InterruptedException e)
{
throw new IOException(e);
}
}
if (loader.getFailedHosts().size() > 0)
{
if (loader.getFailedHosts().size() > maxFailures)
throw new IOException("Too many hosts failed: " + loader.getFailedHosts());
else
logger.warn("Some hosts failed: {}", loader.getFailedHosts());
}
}
}
@Override
public void write(ByteBuffer keybuff, List<Mutation> value) throws IOException
{
setTypes(value.get(0));
prepareWriter();
SSTableSimpleUnsortedWriter ssWriter = (SSTableSimpleUnsortedWriter) writer;
ssWriter.newRow(keybuff);
for (Mutation mut : value)
{
if (cfType == CFType.SUPER)
{
ssWriter.newSuperColumn(mut.getColumn_or_supercolumn().getSuper_column().name);
if (colType == ColType.COUNTER)
for (CounterColumn column : mut.getColumn_or_supercolumn().getCounter_super_column().columns)
ssWriter.addCounterColumn(column.name, column.value);
else
{
for (Column column : mut.getColumn_or_supercolumn().getSuper_column().columns)
{
if(column.ttl == 0)
ssWriter.addColumn(column.name, column.value, column.timestamp);
else
ssWriter.addExpiringColumn(column.name, column.value, column.timestamp, column.ttl, System.currentTimeMillis() + ((long)column.ttl * 1000));
}
}
}
else
{
if (colType == ColType.COUNTER)
ssWriter.addCounterColumn(mut.getColumn_or_supercolumn().counter_column.name, mut.getColumn_or_supercolumn().counter_column.value);
else
{
if(mut.getColumn_or_supercolumn().column.ttl == 0)
ssWriter.addColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value, mut.getColumn_or_supercolumn().column.timestamp);
else
ssWriter.addExpiringColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value, mut.getColumn_or_supercolumn().column.timestamp, mut.getColumn_or_supercolumn().column.ttl, System.currentTimeMillis() + ((long)(mut.getColumn_or_supercolumn().column.ttl) * 1000));
}
}
if (null != progress)
progress.progress();
if (null != context)
HadoopCompat.progress(context);
}
}
public static class ExternalClient extends NativeSSTableLoaderClient
{
public ExternalClient(Configuration conf)
{
super(resolveHostAddresses(conf),
CqlConfigHelper.getOutputNativePort(conf),
ConfigHelper.getOutputKeyspaceUserName(conf),
ConfigHelper.getOutputKeyspacePassword(conf),
CqlConfigHelper.getSSLOptions(conf).orNull());
}
private static Collection<InetAddress> resolveHostAddresses(Configuration conf)
{
Set<InetAddress> addresses = new HashSet<>();
for (String host : ConfigHelper.getOutputInitialAddress(conf).split(","))
{
try
{
addresses.add(InetAddress.getByName(host));
}
catch (UnknownHostException e)
{
throw new RuntimeException(e);
}
}
return addresses;
}
}
public static class NullOutputHandler implements OutputHandler
{
public void output(String msg) {}
public void debug(String msg) {}
public void warn(String msg) {}
public void warn(String msg, Throwable th) {}
}
}

125
src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java
View File

@ -1,125 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.hadoop;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.auth.PasswordAuthenticator;
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.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
/**
* Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily.
*
* At minimum, you need to set the CF and predicate (description of columns to extract from each row)
* in your Hadoop job Configuration. The ConfigHelper class is provided to make this
* simple:
* ConfigHelper.setInputColumnFamily
* ConfigHelper.setInputSlicePredicate
*
* You can also configure the number of rows per InputSplit with
* ConfigHelper.setInputSplitSize
* This should be "as big as possible, but no bigger." Each InputSplit is read from Cassandra
* with multiple get_slice_range queries, and the per-call overhead of get_slice_range is high,
* so larger split sizes are better -- but if it is too large, you will run out of memory.
*
* The default split size is 64k rows.
*/
@Deprecated
public class ColumnFamilyInputFormat extends AbstractColumnFamilyInputFormat<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>>
{
private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyInputFormat.class);
@SuppressWarnings("resource")
public static Cassandra.Client createAuthenticatedClient(String location, int port, Configuration conf) throws Exception
{
logger.debug("Creating authenticated client for CF input format");
TTransport transport;
try
{
transport = ConfigHelper.getClientTransportFactory(conf).openTransport(location, port);
}
catch (Exception e)
{
throw new TTransportException("Failed to open a transport to " + location + ":" + port + ".", e);
}
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) && (ConfigHelper.getInputKeyspacePassword(conf) != null))
{
Map<String, String> creds = new HashMap<String, String>();
creds.put(PasswordAuthenticator.USERNAME_KEY, ConfigHelper.getInputKeyspaceUserName(conf));
creds.put(PasswordAuthenticator.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 RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException
{
return new ColumnFamilyRecordReader();
}
public org.apache.hadoop.mapred.RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>> getRecordReader(org.apache.hadoop.mapred.InputSplit split, JobConf jobConf, final Reporter reporter) throws IOException
{
TaskAttemptContext tac = HadoopCompat.newMapContext(
jobConf,
TaskAttemptID.forName(jobConf.get(MAPRED_TASK_ID)),
null,
null,
null,
new ReporterWrapper(reporter),
null);
ColumnFamilyRecordReader recordReader = new ColumnFamilyRecordReader(jobConf.getInt(CASSANDRA_HADOOP_MAX_KEY_SIZE, CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT));
recordReader.initialize((org.apache.hadoop.mapreduce.InputSplit)split, tac);
return recordReader;
}
@Override
protected void validateConfiguration(Configuration conf)
{
super.validateConfiguration(conf);
if (ConfigHelper.getInputSlicePredicate(conf) == null)
{
throw new UnsupportedOperationException("you must set the predicate with setInputSlicePredicate");
}
}
}

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

@ -1,183 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.hadoop;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.*;
import org.slf4j.*;
import org.apache.cassandra.auth.*;
import org.apache.cassandra.thrift.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.thrift.protocol.*;
import org.apache.thrift.transport.*;
/**
* The <code>ColumnFamilyOutputFormat</code> 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.
*
* <p>
* 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.
* </p>
*
* <p>
* 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.
* </p>
*/
@Deprecated
public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutation>>
implements org.apache.hadoop.mapred.OutputFormat<ByteBuffer,List<Mutation>>
{
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);
/**
* 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();
}
/**
* Check for validity of the output-specification for the job.
*
* @param context
* information about the job
*/
public void checkOutputSpecs(JobContext context)
{
checkOutputSpecs(HadoopCompat.getConfiguration(context));
}
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);
}
/**
* 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
*/
@SuppressWarnings("resource")
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);
TProtocol binaryProtocol = new TBinaryProtocol(transport, true, true);
Cassandra.Client client = new Cassandra.Client(binaryProtocol);
client.set_keyspace(ConfigHelper.getOutputKeyspace(conf));
String user = ConfigHelper.getOutputKeyspaceUserName(conf);
String password = ConfigHelper.getOutputKeyspacePassword(conf);
if ((user != null) && (password != null))
login(user, password, client);
logger.debug("Authenticated client for CF output format created successfully");
return client;
}
public static void login(String user, String password, Cassandra.Client client) throws Exception
{
Map<String, String> creds = new HashMap<String, String>();
creds.put(PasswordAuthenticator.USERNAME_KEY, user);
creds.put(PasswordAuthenticator.PASSWORD_KEY, password);
AuthenticationRequest authRequest = new AuthenticationRequest(creds);
client.login(authRequest);
}
/** 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)
{
return new ColumnFamilyRecordWriter(job, 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.
*/
public ColumnFamilyRecordWriter getRecordWriter(final TaskAttemptContext context) throws InterruptedException
{
return new ColumnFamilyRecordWriter(context);
}
/**
* 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) { }
}
}

615
src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
View File

@ -1,615 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.hadoop;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.collect.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.schema.LegacySchemaTables;
import org.apache.cassandra.db.SystemKeyspace;
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.exceptions.ConfigurationException;
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;
@Deprecated
public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>>
implements org.apache.hadoop.mapred.RecordReader<ByteBuffer, SortedMap<ByteBuffer, ColumnFamilyRecordReader.Column>>
{
private static final Logger logger = LoggerFactory.getLogger(ColumnFamilyRecordReader.class);
public static final int CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT = 8192;
private ColumnFamilySplit split;
private RowIterator iter;
private Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> currentRow;
private SlicePredicate predicate;
private boolean isEmptyPredicate;
private int totalRowCount; // total number of rows to fetch
private int batchSize; // fetch this many per batch
private String keyspace;
private String cfName;
private Cassandra.Client client;
private ConsistencyLevel consistencyLevel;
private int keyBufferSize = 8192;
private List<IndexExpression> filter;
public ColumnFamilyRecordReader()
{
this(ColumnFamilyRecordReader.CASSANDRA_HADOOP_MAX_KEY_SIZE_DEFAULT);
}
public ColumnFamilyRecordReader(int keyBufferSize)
{
super();
this.keyBufferSize = keyBufferSize;
}
@SuppressWarnings("resource")
public void close()
{
if (client != null)
{
TTransport transport = client.getOutputProtocol().getTransport();
if (transport.isOpen())
transport.close();
}
}
public ByteBuffer getCurrentKey()
{
return currentRow.left;
}
public SortedMap<ByteBuffer, Column> getCurrentValue()
{
return currentRow.right;
}
public float getProgress()
{
if (!iter.hasNext())
return 1.0F;
// the progress is likely to be reported slightly off the actual but close enough
float progress = ((float) iter.rowsRead() / totalRowCount);
return progress > 1.0F ? 1.0F : progress;
}
static boolean isEmptyPredicate(SlicePredicate predicate)
{
if (predicate == null)
return true;
if (predicate.isSetColumn_names() && predicate.getSlice_range() == null)
return false;
if (predicate.getSlice_range() == null)
return true;
byte[] start = predicate.getSlice_range().getStart();
if ((start != null) && (start.length > 0))
return false;
byte[] finish = predicate.getSlice_range().getFinish();
if ((finish != null) && (finish.length > 0))
return false;
return true;
}
public void initialize(InputSplit split, TaskAttemptContext context) throws IOException
{
this.split = (ColumnFamilySplit) split;
Configuration conf = HadoopCompat.getConfiguration(context);
KeyRange jobRange = ConfigHelper.getInputKeyRange(conf);
filter = jobRange == null ? null : jobRange.row_filter;
predicate = ConfigHelper.getInputSlicePredicate(conf);
boolean widerows = ConfigHelper.getInputIsWide(conf);
isEmptyPredicate = isEmptyPredicate(predicate);
totalRowCount = (this.split.getLength() < Long.MAX_VALUE)
? (int) this.split.getLength()
: ConfigHelper.getInputSplitSize(conf);
batchSize = ConfigHelper.getRangeBatchSize(conf);
cfName = ConfigHelper.getInputColumnFamily(conf);
consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getReadConsistencyLevel(conf));
keyspace = ConfigHelper.getInputKeyspace(conf);
if (batchSize < 2)
throw new IllegalArgumentException("Minimum batchSize is 2. Suggested batchSize is 100 or more");
try
{
if (client != null)
return;
// create connection using thrift
String location = getLocation();
int port = ConfigHelper.getInputRpcPort(conf);
client = ColumnFamilyInputFormat.createAuthenticatedClient(location, port, conf);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
iter = widerows ? new WideRowIterator() : new StaticRowIterator();
logger.debug("created {}", iter);
}
public boolean nextKeyValue() throws IOException
{
if (!iter.hasNext())
{
logger.debug("Finished scanning {} rows (estimate was: {})", iter.rowsRead(), totalRowCount);
return false;
}
currentRow = iter.next();
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<InetAddress> localAddresses = FBUtilities.getAllLocalAddresses();
for (InetAddress address : localAddresses)
{
for (String location : split.getLocations())
{
InetAddress locationAddress = null;
try
{
locationAddress = InetAddress.getByName(location);
}
catch (UnknownHostException e)
{
throw new AssertionError(e);
}
if (address.equals(locationAddress))
{
return location;
}
}
}
return split.getLocations()[0];
}
private abstract class RowIterator extends AbstractIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, Column>>>
{
protected List<KeySlice> rows;
protected int totalRead = 0;
protected final boolean isSuper;
protected final AbstractType<?> comparator;
protected final AbstractType<?> subComparator;
protected final IPartitioner partitioner;
private RowIterator()
{
CfDef cfDef = new CfDef();
try
{
partitioner = FBUtilities.newPartitioner(client.describe_partitioner());
// get CF meta data
String query = String.format("SELECT comparator, subcomparator, type " +
"FROM %s.%s " +
"WHERE keyspace_name = '%s' AND columnfamily_name = '%s'",
SystemKeyspace.NAME,
LegacySchemaTables.COLUMNFAMILIES,
keyspace,
cfName);
CqlResult result = client.execute_cql3_query(ByteBufferUtil.bytes(query), Compression.NONE, ConsistencyLevel.ONE);
Iterator<CqlRow> iteraRow = result.rows.iterator();
if (iteraRow.hasNext())
{
CqlRow cqlRow = iteraRow.next();
cfDef.comparator_type = ByteBufferUtil.string(cqlRow.columns.get(0).value);
ByteBuffer subComparator = cqlRow.columns.get(1).value;
if (subComparator != null)
cfDef.subcomparator_type = ByteBufferUtil.string(subComparator);
ByteBuffer type = cqlRow.columns.get(2).value;
if (type != null)
cfDef.column_type = ByteBufferUtil.string(type);
}
comparator = TypeParser.parse(cfDef.comparator_type);
subComparator = cfDef.subcomparator_type == null ? null : TypeParser.parse(cfDef.subcomparator_type);
}
catch (ConfigurationException e)
{
throw new RuntimeException("unable to load sub/comparator", e);
}
catch (TException e)
{
throw new RuntimeException("error communicating via Thrift", e);
}
catch (Exception e)
{
throw new RuntimeException("unable to load keyspace " + keyspace, e);
}
isSuper = "Super".equalsIgnoreCase(cfDef.column_type);
}
/**
* @return total number of rows read by this record reader
*/
public int rowsRead()
{
return totalRead;
}
protected List<Pair<ByteBuffer, Column>> unthriftify(ColumnOrSuperColumn cosc)
{
if (cosc.counter_column != null)
return Collections.singletonList(unthriftifyCounter(cosc.counter_column));
if (cosc.counter_super_column != null)
return unthriftifySuperCounter(cosc.counter_super_column);
if (cosc.super_column != null)
return unthriftifySuper(cosc.super_column);
assert cosc.column != null;
return Collections.singletonList(unthriftifySimple(cosc.column));
}
private List<Pair<ByteBuffer, Column>> unthriftifySuper(SuperColumn super_column)
{
List<Pair<ByteBuffer, Column>> columns = new ArrayList<>(super_column.columns.size());
for (org.apache.cassandra.thrift.Column column : super_column.columns)
{
Pair<ByteBuffer, Column> c = unthriftifySimple(column);
columns.add(Pair.create(CompositeType.build(super_column.name, c.left), c.right));
}
return columns;
}
protected Pair<ByteBuffer, Column> unthriftifySimple(org.apache.cassandra.thrift.Column column)
{
return Pair.create(column.name, Column.fromRegularColumn(column));
}
private Pair<ByteBuffer, Column> unthriftifyCounter(CounterColumn column)
{
return Pair.create(column.name, Column.fromCounterColumn(column));
}
private List<Pair<ByteBuffer, Column>> unthriftifySuperCounter(CounterSuperColumn super_column)
{
List<Pair<ByteBuffer, Column>> columns = new ArrayList<>(super_column.columns.size());
for (CounterColumn column : super_column.columns)
{
Pair<ByteBuffer, Column> c = unthriftifyCounter(column);
columns.add(Pair.create(CompositeType.build(super_column.name, c.left), c.right));
}
return columns;
}
}
private class StaticRowIterator extends RowIterator
{
protected int i = 0;
private void maybeInit()
{
// check if we need another batch
if (rows != null && i < rows.size())
return;
String startToken;
if (totalRead == 0)
{
// first request
startToken = split.getStartToken();
}
else
{
startToken = partitioner.getTokenFactory().toString(partitioner.getToken(Iterables.getLast(rows).key));
if (startToken.equals(split.getEndToken()))
{
// reached end of the split
rows = null;
return;
}
}
KeyRange keyRange = new KeyRange(batchSize)
.setStart_token(startToken)
.setEnd_token(split.getEndToken())
.setRow_filter(filter);
try
{
rows = client.get_range_slices(new ColumnParent(cfName), predicate, keyRange, consistencyLevel);
// nothing new? reached the end
if (rows.isEmpty())
{
rows = null;
return;
}
// remove ghosts when fetching all columns
if (isEmptyPredicate)
{
Iterator<KeySlice> it = rows.iterator();
KeySlice ks;
do
{
ks = it.next();
if (ks.getColumnsSize() == 0)
{
it.remove();
}
} while (it.hasNext());
// all ghosts, spooky
if (rows.isEmpty())
{
// maybeInit assumes it can get the start-with key from the rows collection, so add back the last
rows.add(ks);
maybeInit();
return;
}
}
// reset to iterate through this new batch
i = 0;
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
protected Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> computeNext()
{
maybeInit();
if (rows == null)
return endOfData();
totalRead++;
KeySlice ks = rows.get(i++);
AbstractType<?> comp = isSuper ? CompositeType.getInstance(comparator, subComparator) : comparator;
SortedMap<ByteBuffer, Column> map = new TreeMap<>(comp);
for (ColumnOrSuperColumn cosc : ks.columns)
{
List<Pair<ByteBuffer, Column>> columns = unthriftify(cosc);
for (Pair<ByteBuffer, Column> column : columns)
map.put(column.left, column.right);
}
return Pair.create(ks.key, map);
}
}
private class WideRowIterator extends RowIterator
{
private PeekingIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, Column>>> wideColumns;
private ByteBuffer lastColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER;
private ByteBuffer lastCountedKey = ByteBufferUtil.EMPTY_BYTE_BUFFER;
private void maybeInit()
{
if (wideColumns != null && wideColumns.hasNext())
return;
KeyRange keyRange;
if (totalRead == 0)
{
String startToken = split.getStartToken();
keyRange = new KeyRange(batchSize)
.setStart_token(startToken)
.setEnd_token(split.getEndToken())
.setRow_filter(filter);
}
else
{
KeySlice lastRow = Iterables.getLast(rows);
logger.debug("Starting with last-seen row {}", lastRow.key);
keyRange = new KeyRange(batchSize)
.setStart_key(lastRow.key)
.setEnd_token(split.getEndToken())
.setRow_filter(filter);
}
try
{
rows = client.get_paged_slice(cfName, keyRange, lastColumn, consistencyLevel);
int n = 0;
for (KeySlice row : rows)
n += row.columns.size();
logger.debug("read {} columns in {} rows for {} starting with {}",
new Object[]{ n, rows.size(), keyRange, lastColumn });
wideColumns = Iterators.peekingIterator(new WideColumnIterator(rows));
if (wideColumns.hasNext() && wideColumns.peek().right.keySet().iterator().next().equals(lastColumn))
wideColumns.next();
if (!wideColumns.hasNext())
rows = null;
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
protected Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> computeNext()
{
maybeInit();
if (rows == null)
return endOfData();
Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> next = wideColumns.next();
lastColumn = next.right.keySet().iterator().next().duplicate();
maybeIncreaseRowCounter(next);
return next;
}
/**
* Increases the row counter only if we really moved to the next row.
* @param next just fetched row slice
*/
private void maybeIncreaseRowCounter(Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> next)
{
ByteBuffer currentKey = next.left;
if (!currentKey.equals(lastCountedKey))
{
totalRead++;
lastCountedKey = currentKey;
}
}
private class WideColumnIterator extends AbstractIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, Column>>>
{
private final Iterator<KeySlice> rows;
private Iterator<ColumnOrSuperColumn> columns;
public KeySlice currentRow;
public WideColumnIterator(List<KeySlice> rows)
{
this.rows = rows.iterator();
if (this.rows.hasNext())
nextRow();
else
columns = Iterators.emptyIterator();
}
private void nextRow()
{
currentRow = rows.next();
columns = currentRow.columns.iterator();
}
protected Pair<ByteBuffer, SortedMap<ByteBuffer, Column>> computeNext()
{
AbstractType<?> comp = isSuper ? CompositeType.getInstance(comparator, subComparator) : comparator;
while (true)
{
if (columns.hasNext())
{
ColumnOrSuperColumn cosc = columns.next();
SortedMap<ByteBuffer, Column> map;
List<Pair<ByteBuffer, Column>> columns = unthriftify(cosc);
if (columns.size() == 1)
{
map = ImmutableSortedMap.of(columns.get(0).left, columns.get(0).right);
}
else
{
assert isSuper;
map = new TreeMap<>(comp);
for (Pair<ByteBuffer, Column> column : columns)
map.put(column.left, column.right);
}
return Pair.create(currentRow.key, map);
}
if (!rows.hasNext())
return endOfData();
nextRow();
}
}
}
}
// 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(ByteBuffer key, SortedMap<ByteBuffer, Column> value) throws IOException
{
if (this.nextKeyValue())
{
key.clear();
key.put(this.getCurrentKey().duplicate());
key.flip();
value.clear();
value.putAll(this.getCurrentValue());
return true;
}
return false;
}
public ByteBuffer createKey()
{
return ByteBuffer.wrap(new byte[this.keyBufferSize]);
}
public SortedMap<ByteBuffer, Column> createValue()
{
return new TreeMap<>();
}
public long getPos() throws IOException
{
return iter.rowsRead();
}
public static final class Column
{
public final ByteBuffer name;
public final ByteBuffer value;
public final long timestamp;
private Column(ByteBuffer name, ByteBuffer value, long timestamp)
{
this.name = name;
this.value = value;
this.timestamp = timestamp;
}
static Column fromRegularColumn(org.apache.cassandra.thrift.Column input)
{
return new Column(input.name, input.value, input.timestamp);
}
static Column fromCounterColumn(org.apache.cassandra.thrift.CounterColumn input)
{
return new Column(input.name, ByteBufferUtil.bytes(input.value), 0);
}
}
}

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

@ -1,341 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.hadoop;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import org.apache.cassandra.client.*;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.utils.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.*;
import org.apache.thrift.TException;
import org.apache.hadoop.util.Progressable;
import org.apache.thrift.transport.*;
/**
* The <code>ColumnFamilyRecordWriter</code> maps the output &lt;key, value&gt;
* 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.
*
* <p>
* 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.
* </p>
*
* @see ColumnFamilyOutputFormat
*/
@Deprecated
final class ColumnFamilyRecordWriter extends RecordWriter<ByteBuffer, List<Mutation>> implements
org.apache.hadoop.mapred.RecordWriter<ByteBuffer, List<Mutation>>
{
// The configuration this writer is associated with.
protected final Configuration conf;
// The number of mutations to buffer per endpoint
protected final int queueSize;
protected final long batchThreshold;
protected final ConsistencyLevel consistencyLevel;
protected Progressable progressable;
protected TaskAttemptContext context;
// handles for clients for each range running in the threadpool
private final Map<Range, RangeClient> clients;
// 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;
/**
* 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)
{
this(HadoopCompat.getConfiguration(context));
this.context = context;
}
ColumnFamilyRecordWriter(Configuration conf, Progressable progressable)
{
this(conf);
this.progressable = progressable;
}
ColumnFamilyRecordWriter(Configuration conf)
{
this.conf = conf;
this.queueSize = conf.getInt(ColumnFamilyOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, 32);
consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getWriteConsistencyLevel(conf));
this.ringCache = new RingCache(conf);
this.clients = new HashMap<Range, RangeClient>();
}
/**
* Close this <code>RecordWriter</code> 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
{
close();
}
/** Fills the deprecated RecordWriter interface for streaming. */
@Deprecated
public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException
{
close();
}
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}.
* </p>
*
* @param keybuff
* the key to write.
* @param value
* the value to write.
* @throws IOException
*/
@Override
public void write(ByteBuffer keybuff, List<Mutation> value) throws IOException
{
Range<Token> range = ringCache.getRange(keybuff);
// 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);
}
for (Mutation amut : value)
client.put(Pair.create(keybuff, amut));
if (progressable != null)
progressable.progress();
if (context != null)
HadoopCompat.progress(context);
}
/**
* 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
{
// The list of endpoints for this range
protected final List<InetAddress> endpoints;
// A bounded queue of incoming mutations for this range
protected final BlockingQueue<Pair<ByteBuffer, Mutation>> 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;
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<InetAddress> endpoints)
{
super("client-" + endpoints);
this.endpoints = endpoints;
}
/**
* enqueues the given value to Cassandra
*/
public void put(Pair<ByteBuffer, Mutation> 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;
}
@SuppressWarnings("resource")
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 void run()
{
outer:
while (run || !queue.isEmpty())
{
Pair<ByteBuffer, Mutation> mutation;
try
{
mutation = queue.take();
}
catch (InterruptedException e)
{
// re-check loop condition after interrupt
continue;
}
Map<ByteBuffer, Map<String, List<Mutation>>> batch = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
while (mutation != null)
{
Map<String, List<Mutation>> subBatch = batch.get(mutation.left);
if (subBatch == null)
{
subBatch = Collections.singletonMap(columnFamily, (List<Mutation>) new ArrayList<Mutation>());
batch.put(mutation.left, subBatch);
}
subBatch.get(columnFamily).add(mutation.right);
if (batch.size() >= batchThreshold)
break;
mutation = queue.poll();
}
Iterator<InetAddress> iter = endpoints.iterator();
while (true)
{
// send the mutation to the last-used endpoint. first time through, this will NPE harmlessly.
try
{
client.batch_mutate(batch, consistencyLevel);
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;
}
}
}
}
}
}
}

12
src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
View File

@ -33,7 +33,6 @@ import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.hadoop.BulkRecordWriter;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.io.sstable.CQLSSTableWriter;
@ -41,6 +40,7 @@ import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.utils.NativeSSTableLoaderClient;
import org.apache.cassandra.utils.OutputHandler;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
@ -152,7 +152,7 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
ExternalClient externalClient = new ExternalClient(conf);
externalClient.setTableMetadata(CFMetaData.compile(schema, keyspace));
loader = new SSTableLoader(outputDir, externalClient, new BulkRecordWriter.NullOutputHandler())
loader = new SSTableLoader(outputDir, externalClient, new NullOutputHandler())
{
@Override
public void onSuccess(StreamState finalState)
@ -287,4 +287,12 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
return addresses;
}
}
public static class NullOutputHandler implements OutputHandler
{
public void output(String msg) {}
public void debug(String msg) {}
public void warn(String msg) {}
public void warn(String msg, Throwable th) {}
}
}

277
src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
View File

@ -18,12 +18,44 @@
package org.apache.cassandra.hadoop.cql3;
import java.io.IOException;
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.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.hadoop.AbstractColumnFamilyInputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TokenRange;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.OrderPreservingPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.hadoop.ColumnFamilySplit;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.hadoop.conf.Configuration;
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.JobContext;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
@ -48,8 +80,15 @@ import com.datastax.driver.core.Row;
*
* other native protocol connection parameters in CqlConfigHelper
*/
public class CqlInputFormat extends AbstractColumnFamilyInputFormat<Long, Row>
public class CqlInputFormat extends org.apache.hadoop.mapreduce.InputFormat<Long, Row> implements org.apache.hadoop.mapred.InputFormat<Long, Row>
{
public static final String MAPRED_TASK_ID = "mapred.task.id";
private static final Logger logger = LoggerFactory.getLogger(CqlInputFormat.class);
private String keyspace;
private String cfName;
private IPartitioner partitioner;
private Session session;
public RecordReader<Long, Row> getRecordReader(InputSplit split, JobConf jobConf, final Reporter reporter)
throws IOException
{
@ -75,4 +114,238 @@ public class CqlInputFormat extends AbstractColumnFamilyInputFormat<Long, Row>
return new CqlRecordReader();
}
protected void validateConfiguration(Configuration conf)
{
if (ConfigHelper.getInputKeyspace(conf) == null || ConfigHelper.getInputColumnFamily(conf) == null)
{
throw new UnsupportedOperationException("you must set the keyspace and table 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 List<org.apache.hadoop.mapreduce.InputSplit> getSplits(JobContext context) throws IOException
{
Configuration conf = HadoopCompat.getConfiguration(context);
validateConfiguration(conf);
keyspace = ConfigHelper.getInputKeyspace(conf);
cfName = ConfigHelper.getInputColumnFamily(conf);
partitioner = ConfigHelper.getInputPartitioner(conf);
logger.debug("partitioner is {}", partitioner);
// canonical ranges and nodes holding replicas
Map<TokenRange, Set<Host>> masterRangeNodes = getRangeMap(conf, keyspace);
// canonical ranges, split into pieces, fetching the splits in parallel
ExecutorService executor = new ThreadPoolExecutor(0, 128, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
List<org.apache.hadoop.mapreduce.InputSplit> splits = new ArrayList<>();
try
{
List<Future<List<org.apache.hadoop.mapreduce.InputSplit>>> splitfutures = new ArrayList<>();
KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf);
Range<Token> jobRange = null;
if (jobKeyRange != null)
{
if (jobKeyRange.start_key != null)
{
if (!partitioner.preservesOrder())
throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving partitioner");
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));
}
else if (jobKeyRange.start_token != null)
{
jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.start_token),
partitioner.getTokenFactory().fromString(jobKeyRange.end_token));
}
else
{
logger.warn("ignoring jobKeyRange specified without start_key or start_token");
}
}
session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect();
Metadata metadata = session.getCluster().getMetadata();
for (TokenRange range : masterRangeNodes.keySet())
{
if (jobRange == null)
{
// for each tokenRange, pick a live owner and ask it to compute bite-sized splits
splitfutures.add(executor.submit(new SplitCallable(range, masterRangeNodes.get(range), conf)));
}
else
{
TokenRange jobTokenRange = rangeToTokenRange(metadata, jobRange);
if (range.intersects(jobTokenRange))
{
for (TokenRange intersection: range.intersectWith(jobTokenRange))
{
// for each tokenRange, pick a live owner and ask it to compute bite-sized splits
splitfutures.add(executor.submit(new SplitCallable(intersection, masterRangeNodes.get(range), conf)));
}
}
}
}
// wait until we have all the results back
for (Future<List<org.apache.hadoop.mapreduce.InputSplit>> 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;
}
private TokenRange rangeToTokenRange(Metadata metadata, Range<Token> range)
{
return metadata.newTokenRange(metadata.newToken(partitioner.getTokenFactory().toString(range.left)),
metadata.newToken(partitioner.getTokenFactory().toString(range.right)));
}
private Map<TokenRange, Long> getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException
{
int splitSize = ConfigHelper.getInputSplitSize(conf);
try
{
return describeSplits(keyspace, cfName, range, splitSize);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
private Map<TokenRange, Set<Host>> getRangeMap(Configuration conf, String keyspace)
{
try (Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect())
{
Map<TokenRange, Set<Host>> map = new HashMap<>();
Metadata metadata = session.getCluster().getMetadata();
for (TokenRange tokenRange : metadata.getTokenRanges())
map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange));
return map;
}
}
private Map<TokenRange, Long> describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize)
{
String query = String.format("SELECT mean_partition_size, partitions_count " +
"FROM %s.%s " +
"WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?",
SystemKeyspace.NAME,
SystemKeyspace.SIZE_ESTIMATES);
ResultSet resultSet = session.execute(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString());
Row row = resultSet.one();
// If we have no data on this split, return the full split i.e., do not sub-split
// Assume smallest granularity of partition count available from CASSANDRA-7688
if (row == null)
{
Map<TokenRange, Long> wrappedTokenRange = new HashMap<>();
wrappedTokenRange.put(tokenRange, (long) 128);
return wrappedTokenRange;
}
long meanPartitionSize = row.getLong("mean_partition_size");
long partitionCount = row.getLong("partitions_count");
int splitCount = (int)((meanPartitionSize * partitionCount) / splitSize);
List<TokenRange> splitRanges = tokenRange.splitEvenly(splitCount);
Map<TokenRange, Long> rangesWithLength = new HashMap<>();
for (TokenRange range : splitRanges)
rangesWithLength.put(range, partitionCount/splitCount);
return rangesWithLength;
}
// Old Hadoop API
public InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException
{
TaskAttemptContext tac = HadoopCompat.newTaskAttemptContext(jobConf, new TaskAttemptID());
List<org.apache.hadoop.mapreduce.InputSplit> newInputSplits = this.getSplits(tac);
InputSplit[] oldInputSplits = new InputSplit[newInputSplits.size()];
for (int i = 0; i < newInputSplits.size(); i++)
oldInputSplits[i] = (ColumnFamilySplit)newInputSplits.get(i);
return oldInputSplits;
}
/**
* Gets a token tokenRange and splits it up according to the suggested
* size into input splits that Hadoop can use.
*/
class SplitCallable implements Callable<List<org.apache.hadoop.mapreduce.InputSplit>>
{
private final TokenRange tokenRange;
private final Set<Host> hosts;
private final Configuration conf;
public SplitCallable(TokenRange tr, Set<Host> hosts, Configuration conf)
{
this.tokenRange = tr;
this.hosts = hosts;
this.conf = conf;
}
public List<org.apache.hadoop.mapreduce.InputSplit> call() throws Exception
{
ArrayList<org.apache.hadoop.mapreduce.InputSplit> splits = new ArrayList<>();
Map<TokenRange, Long> subSplits;
subSplits = getSubSplits(keyspace, cfName, tokenRange, conf);
// turn the sub-ranges into InputSplits
String[] endpoints = new String[hosts.size()];
// hadoop needs hostname, not ip
int endpointIndex = 0;
for (Host endpoint : hosts)
endpoints[endpointIndex++] = endpoint.getAddress().getHostName();
boolean partitionerIsOpp = partitioner instanceof OrderPreservingPartitioner || partitioner instanceof ByteOrderedPartitioner;
for (TokenRange subSplit : subSplits.keySet())
{
List<TokenRange> ranges = subSplit.unwrap();
for (TokenRange subrange : ranges)
{
ColumnFamilySplit split =
new ColumnFamilySplit(
partitionerIsOpp ?
subrange.getStart().toString().substring(2) : subrange.getStart().toString(),
partitionerIsOpp ?
subrange.getEnd().toString().substring(2) : subrange.getStart().toString(),
subSplits.get(subSplit),
endpoints);
logger.debug("adding {}", split);
splits.add(split);
}
}
return splits;
}
}
}

3
src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java
View File

@ -55,6 +55,9 @@ import org.apache.hadoop.mapreduce.*;
public class CqlOutputFormat extends OutputFormat<Map<String, ByteBuffer>, List<ByteBuffer>>
implements org.apache.hadoop.mapred.OutputFormat<Map<String, ByteBuffer>, List<ByteBuffer>>
{
public static final String BATCH_THRESHOLD = "mapreduce.output.columnfamilyoutputformat.batch.threshold";
public static final String QUEUE_SIZE = "mapreduce.output.columnfamilyoutputformat.queue.size";
/**
* Check for validity of the output-specification for the job.
*

5
src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
View File

@ -31,7 +31,6 @@ import com.datastax.driver.core.exceptions.*;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.hadoop.ColumnFamilyOutputFormat;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.utils.FBUtilities;
@ -109,8 +108,8 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
CqlRecordWriter(Configuration conf)
{
this.conf = conf;
this.queueSize = conf.getInt(ColumnFamilyOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, 32);
this.queueSize = conf.getInt(CqlOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
batchThreshold = conf.getLong(CqlOutputFormat.BATCH_THRESHOLD, 32);
this.clients = new HashMap<>();
try

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

File diff suppressed because it is too large Load Diff

27
test/pig/org/apache/cassandra/pig/PigTestBase.java
View File

@ -20,6 +20,8 @@ package org.apache.cassandra.pig;
import java.io.IOException;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.marshal.AbstractType;
@ -27,9 +29,6 @@ import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.service.EmbeddedCassandraService;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.pig.ExecType;
@ -37,13 +36,6 @@ import org.apache.pig.PigServer;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.impl.PigContext;
import org.apache.pig.test.MiniCluster;
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.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@ -80,13 +72,10 @@ public class PigTestBase extends SchemaLoader
pig.shutdown();
}
protected static Cassandra.Client getClient() throws TTransportException
protected static Session getClient()
{
TTransport tr = new TFramedTransport(new TSocket("localhost", 9170));
TProtocol proto = new TBinaryProtocol(tr);
Cassandra.Client client = new Cassandra.Client(proto);
tr.open();
return client;
Cluster cluster = Cluster.builder().addContactPoints("localhost").withPort(9042).build();
return cluster.connect();
}
protected static void startCassandra() throws IOException
@ -114,14 +103,14 @@ public class PigTestBase extends SchemaLoader
}
}
protected static void executeCQLStatements(String[] statements) throws TException
protected static void executeCQLStatements(String[] statements)
{
Cassandra.Client client = getClient();
Session client = getClient();
for (String statement : statements)
{
System.out.println("Executing statement: " + statement);
client.execute_cql3_query(ByteBufferUtil.bytes(statement), Compression.NONE, ConsistencyLevel.ONE);
client.execute(statement);
}
}
}

70
test/pig/org/apache/cassandra/pig/ThriftColumnFamilyDataTypeTest.java
View File

@ -24,10 +24,8 @@ import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.utils.Hex;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
import org.apache.thrift.TException;
import org.junit.BeforeClass;
import org.junit.Test;
@ -69,7 +67,7 @@ public class ThriftColumnFamilyDataTypeTest extends PigTestBase
};
@BeforeClass
public static void setup() throws IOException, ConfigurationException, TException
public static void setup() throws IOException, ConfigurationException
{
startCassandra();
executeCQLStatements(statements);
@ -79,76 +77,74 @@ public class ThriftColumnFamilyDataTypeTest extends PigTestBase
@Test
public void testCassandraStorageDataType() throws IOException
{
pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + "' USING CqlNativeStorage();");
Tuple t = pig.openIterator("rows").next();
// key
assertEquals("foo", t.get(0));
// col_ascii
Tuple column = (Tuple) t.get(1);
assertEquals("ascii", column.get(1));
Object column = t.get(1);
assertEquals("ascii", column);
// col_bigint
column = (Tuple) t.get(2);
assertEquals(12345678L, column.get(1));
column = t.get(2);
assertEquals(12345678L, column);
// col_blob
column = (Tuple) t.get(3);
assertEquals(new DataByteArray(Hex.hexToBytes("DEADBEEF")), column.get(1));
column = t.get(3);
assertEquals(new DataByteArray(Hex.hexToBytes("DEADBEEF")), column);
// col_boolean
column = (Tuple) t.get(4);
assertEquals(false, column.get(1));
column = t.get(4);
assertEquals(false, column);
// col_decimal
column = (Tuple) t.get(5);
assertEquals("23.345", column.get(1));
column = t.get(5);
assertEquals("23.345", column);
// col_double
column = (Tuple) t.get(6);
assertEquals(2.7182818284590451d, column.get(1));
column = t.get(6);
assertEquals(2.7182818284590451d, column);
// col_float
column = (Tuple) t.get(7);
assertEquals(23.45f, column.get(1));
column = t.get(7);
assertEquals(23.45f, column);
// col_inet
column = (Tuple) t.get(8);
assertEquals("127.0.0.1", column.get(1));
column = t.get(8);
assertEquals("127.0.0.1", column);
// col_int
column = (Tuple) t.get(9);
assertEquals(23, column.get(1));
column = t.get(9);
assertEquals(23, column);
// col_text
column = (Tuple) t.get(10);
assertEquals("hello", column.get(1));
column = t.get(10);
assertEquals("hello", column);
// col_timestamp
column = (Tuple) t.get(11);
assertEquals(1296705900000L, column.get(1));
column = t.get(11);
assertEquals(1296705900000L, column);
// col_timeuuid
column = (Tuple) t.get(12);
assertEquals(new DataByteArray((TimeUUIDType.instance.fromString("e23f450f-53a6-11e2-7f7f-7f7f7f7f7f7f").array())), column.get(1));
column = t.get(12);
assertEquals(new DataByteArray((TimeUUIDType.instance.fromString("e23f450f-53a6-11e2-7f7f-7f7f7f7f7f7f").array())), column);
// col_uuid
column = (Tuple) t.get(13);
assertEquals(new DataByteArray((UUIDType.instance.fromString("550e8400-e29b-41d4-a716-446655440000").array())), column.get(1));
column = t.get(13);
assertEquals(new DataByteArray((UUIDType.instance.fromString("550e8400-e29b-41d4-a716-446655440000").array())), column);
// col_varint
column = (Tuple) t.get(14);
assertEquals(12345, column.get(1));
column = t.get(14);
assertEquals(12345, column);
pig.registerQuery("cc_rows = LOAD 'cassandra://thrift_ks/cc?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("cc_rows = LOAD 'cql://thrift_ks/cc?" + defaultParameters + "' USING CqlNativeStorage();");
t = pig.openIterator("cc_rows").next();
assertEquals("chuck", t.get(0));
DataBag columns = (DataBag) t.get(1);
column = columns.iterator().next();
assertEquals("kick", column.get(0));
assertEquals(3L, column.get(1));
assertEquals("kick", t.get(1));
assertEquals(3L, t.get(2));
}
}

318
test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
View File

@ -22,17 +22,14 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
import org.apache.thrift.TException;
import org.junit.Assert;
import org.junit.BeforeClass;
@ -171,7 +168,7 @@ public class ThriftColumnFamilyTest extends PigTestBase
};
@BeforeClass
public static void setup() throws IOException, ConfigurationException, TException
public static void setup() throws IOException, ConfigurationException
{
startCassandra();
executeCQLStatements(statements);
@ -300,11 +297,11 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
public void testCassandraStorageSchema() throws IOException
public void testCqlNativeStorageSchema() throws IOException
{
//results: (qux,(atomic_weight,0.660161815846869),(created,1335890877),(name,User Qux),(percent,64.7),
//(rating,2),(score,12000),(vote_type,dislike))
pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + "' USING CqlNativeStorage();");
//schema: {key: chararray,atomic_weight: (name: chararray,value: double),created: (name: chararray,value: long),
//name: (name: chararray,value: chararray),percent: (name: chararray,value: float),
@ -339,12 +336,13 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
public void testCassandraStorageFullCopy() throws IOException, TException
public void testCqlNativeStorageFullCopy() throws IOException
{
pig.setBatchOn();
pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
pig.registerQuery("records = FOREACH rows GENERATE TOTUPLE(TOTUPLE('key', key)),TOTUPLE(atomic_weight, created, name, percent, rating, score, vote_type);");
//full copy
pig.registerQuery("STORE rows INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("STORE records INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+atomic_weight+%3D+%3F,+created+%3D+%3F,+name+%3D+%3F,+percent+%3D+%3F,+rating+%3D+%3F,+score+%3D+%3F,+vote_type+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
Assert.assertEquals("User Qux", getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type"));
Assert.assertEquals("dislike", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type"));
@ -352,158 +350,94 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
public void testCassandraStorageSingleTupleCopy() throws IOException, TException
public void testCqlNativeStorageSingleTupleCopy() throws IOException
{
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
//single tuple
pig.registerQuery("onecol = FOREACH rows GENERATE key, percent;");
pig.registerQuery("STORE onecol INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("onecol = FOREACH rows GENERATE TOTUPLE(TOTUPLE('key', key)), TOTUPLE(percent);");
pig.registerQuery("STORE onecol INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+percent+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
String value = null;
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
String value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
if (value != null)
Assert.fail();
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
if (value != null)
Assert.fail();
Assert.assertEquals("64.7", getColumnValue("thrift_ks", "copy_of_some_app", "percent", "qux", "FloatType"));
}
@Test
public void testCassandraStorageBagOnlyCopy() throws IOException, TException
public void testCqlNativeStorageBagOnlyCopy() throws IOException
{
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
//bag only
pig.registerQuery("other = FOREACH rows GENERATE key, columns;");
pig.registerQuery("STORE other INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("other = FOREACH rows GENERATE TOTUPLE(TOTUPLE('key', key)), TOTUPLE();");
pig.registerQuery("STORE other INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "' USING CqlNativeStorage();");
pig.executeBatch();
String value = null;
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
String value = getColumnValue("thrift_ks", "copy_of_some_app", "name", "qux", "UTF8Type");
if (value != null)
Assert.fail();
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
if (value != null)
Assert.fail();
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "percent", "qux", "FloatType");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
value = getColumnValue("thrift_ks", "copy_of_some_app", "percent", "qux", "FloatType");
if (value != null)
Assert.fail();
}
@Test
public void testCassandraStorageFilter() throws IOException, TException
public void testCqlNativeStorageFilter() throws IOException
{
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
//filter
pig.registerQuery("likes = FILTER rows by vote_type.value eq 'like' and rating.value > 5;");
pig.registerQuery("STORE likes INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("likes = FILTER rows by vote_type eq 'like' and rating > 5;");
pig.registerQuery("records = FOREACH likes GENERATE TOTUPLE(TOTUPLE('key', key)),TOTUPLE(atomic_weight, created, name, percent, rating, score, vote_type);");
pig.registerQuery("STORE records INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+atomic_weight+%3D+%3F,+created+%3D+%3F,+name+%3D+%3F,+percent+%3D+%3F,+rating+%3D+%3F,+score+%3D+%3F,+vote_type+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
Assert.assertEquals("like", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "bar", "UTF8Type"));
Assert.assertEquals("like", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "foo", "UTF8Type"));
String value = null;
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
String value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type");
if (value != null)
Assert.fail();
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "baz", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "baz", "UTF8Type");
if (value != null)
Assert.fail();
executeCQLStatements(deleteCopyOfSomeAppTableData);
pig.setBatchOn();
pig.registerQuery("rows = LOAD 'cassandra://thrift_ks/some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("dislikes_extras = FILTER rows by vote_type.value eq 'dislike';");
pig.registerQuery("STORE dislikes_extras INTO 'cassandra://thrift_ks/copy_of_some_app?" + defaultParameters + "' USING CassandraStorage();");
pig.registerQuery("rows = LOAD 'cql://thrift_ks/some_app?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20some_app%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();");
pig.registerQuery("dislikes_extras = FILTER rows by vote_type eq 'dislike';");
pig.registerQuery("dislikes_records = FOREACH dislikes_extras GENERATE TOTUPLE(TOTUPLE('key', key)),TOTUPLE(atomic_weight, created, name, percent, rating, score, vote_type);");
pig.registerQuery("STORE dislikes_records INTO 'cql://thrift_ks/copy_of_some_app?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.copy_of_some_app+set+atomic_weight+%3D+%3F,+created+%3D+%3F,+name+%3D+%3F,+percent+%3D+%3F,+rating+%3D+%3F,+score+%3D+%3F,+vote_type+%3D+%3F' USING CqlNativeStorage();");
pig.executeBatch();
Assert.assertEquals("dislike", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "baz", "UTF8Type"));
Assert.assertEquals("dislike", getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "qux", "UTF8Type"));
value = null;
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "bar", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "bar", "UTF8Type");
if (value != null)
Assert.fail();
try
{
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "foo", "UTF8Type");
}
catch (NotFoundException e)
{
Assert.assertTrue(true);
}
value = getColumnValue("thrift_ks", "copy_of_some_app", "vote_type", "foo", "UTF8Type");
if (value != null)
Assert.fail();
}
@Test
public void testCassandraStorageJoin() throws IOException
public void testCqlNativeStorageJoin() throws IOException
{
//test key types with a join
pig.registerQuery("U8 = load 'cassandra://thrift_ks/u8?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("Bytes = load 'cassandra://thrift_ks/bytes?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("U8 = load 'cql://thrift_ks/u8?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20u8%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
pig.registerQuery("Bytes = load 'cql://thrift_ks/bytes?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20bytes%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
//cast key to chararray
pig.registerQuery("b = foreach Bytes generate (chararray)key, columns;");
pig.registerQuery("b = foreach Bytes generate (chararray)key, column1, value;");
//key in Bytes is a bytearray, U8 chararray
//(foo,{(x,Z)},foo,{(x,Z)})
@ -512,18 +446,11 @@ public class ThriftColumnFamilyTest extends PigTestBase
if (it.hasNext()) {
Tuple t = it.next();
Assert.assertEquals(t.get(0), new DataByteArray("foo".getBytes()));
DataBag columns = (DataBag) t.get(1);
Iterator<Tuple> iter = columns.iterator();
Tuple t1 = iter.next();
Assert.assertEquals(t1.get(0), "x");
Assert.assertEquals(t1.get(1), new DataByteArray("Z".getBytes()));
String column = (String) t.get(2);
Assert.assertEquals(column, "foo");
columns = (DataBag) t.get(3);
iter = columns.iterator();
Tuple t2 = iter.next();
Assert.assertEquals(t2.get(0), "x");
Assert.assertEquals(t2.get(1), new DataByteArray("Z".getBytes()));
Assert.assertEquals(t.get(1), "x");
Assert.assertEquals(t.get(2), new DataByteArray("Z".getBytes()));
Assert.assertEquals(t.get(3), "foo");
Assert.assertEquals(t.get(4), "x");
Assert.assertEquals(t.get(5), new DataByteArray("Z".getBytes()));
}
//key should now be cast into a chararray
//(foo,{(x,Z)},foo,{(x,Z)})
@ -532,27 +459,22 @@ public class ThriftColumnFamilyTest extends PigTestBase
if (it.hasNext()) {
Tuple t = it.next();
Assert.assertEquals(t.get(0), "foo");
DataBag columns = (DataBag) t.get(1);
Iterator<Tuple> iter = columns.iterator();
Tuple t1 = iter.next();
Assert.assertEquals(t1.get(0), "x");
Assert.assertEquals(t1.get(1), new DataByteArray("Z".getBytes()));
String column = (String) t.get(2);
Assert.assertEquals(column, "foo");
columns = (DataBag) t.get(3);
iter = columns.iterator();
Tuple t2 = iter.next();
Assert.assertEquals(t2.get(0), "x");
Assert.assertEquals(t2.get(1), new DataByteArray("Z".getBytes()));
Assert.assertEquals(t.get(1), "x");
Assert.assertEquals(t.get(2), new DataByteArray("Z".getBytes()));
Assert.assertEquals(t.get(3), "foo");
Assert.assertEquals(t.get(4), "x");
Assert.assertEquals(t.get(5), new DataByteArray("Z".getBytes()));
}
}
@Test
public void testCassandraStorageCounterCF() throws IOException
public void testCqlNativeStorageCounterCF() throws IOException
{
//Test counter column family support
pig.registerQuery("CC = load 'cassandra://thrift_ks/cc?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("total_hits = foreach CC generate key, SUM(columns.value);");
pig.registerQuery("CC = load 'cql://thrift_ks/cc?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20cc%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
pig.registerQuery("A = foreach CC generate key, name, value;");
pig.registerQuery("B = GROUP A BY key;");
pig.registerQuery("total_hits = foreach B generate group, SUM(A.value);");
//(chuck,4)
Tuple t = pig.openIterator("total_hits").next();
Assert.assertEquals(t.get(0), "chuck");
@ -560,12 +482,11 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
public void testCassandraStorageCompositeColumnCF() throws IOException
public void testCqlNativeStorageCompositeColumnCF() throws IOException
{
//Test CompositeType
pig.registerQuery("compo = load 'cassandra://thrift_ks/compo?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("compo = foreach compo generate key as method, flatten(columns);");
pig.registerQuery("lee = filter compo by columns::name == ('bruce','lee');");
pig.registerQuery("compo = load 'cql://thrift_ks/compo?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
pig.registerQuery("lee = filter compo by column1 == 'bruce' AND column2 == 'lee';");
//(kick,(bruce,lee),oww)
//(punch,(bruce,lee),ouch)
@ -574,18 +495,16 @@ public class ThriftColumnFamilyTest extends PigTestBase
while (it.hasNext()) {
count ++;
Tuple t = it.next();
Tuple t1 = (Tuple) t.get(1);
Assert.assertEquals(t1.get(0), "bruce");
Assert.assertEquals(t1.get(1), "lee");
Assert.assertEquals(t.get(1), "bruce");
Assert.assertEquals(t.get(2), "lee");
if ("kick".equals(t.get(0)))
Assert.assertEquals(t.get(2), "oww");
else if ("kick".equals(t.get(0)))
Assert.assertEquals(t.get(2), "ouch");
Assert.assertEquals(t.get(3), "oww");
else
Assert.assertEquals(t.get(3), "ouch");
}
Assert.assertEquals(count, 2);
pig.registerQuery("night = load 'cassandra://thrift_ks/compo_int?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("night = foreach night generate flatten(columns);");
pig.registerQuery("night = foreach night generate (int)columns::name.$0+(double)columns::name.$1/60 as hour, columns::value as noise;");
pig.registerQuery("night = load 'cql://thrift_ks/compo_int?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_int%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
pig.registerQuery("night = foreach night generate (int)column1+(double)column2/60 as hour, value as noise;");
//What happens at the darkest hour?
pig.registerQuery("darkest = filter night by hour > 2 and hour < 5;");
@ -598,10 +517,10 @@ public class ThriftColumnFamilyTest extends PigTestBase
Assert.assertEquals(t.get(1), "daddy?");
}
pig.setBatchOn();
pig.registerQuery("compo_int_rows = LOAD 'cassandra://thrift_ks/compo_int?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("STORE compo_int_rows INTO 'cassandra://thrift_ks/compo_int_copy?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("compo_int_rows = LOAD 'cql://thrift_ks/compo_int?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_int%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
pig.registerQuery("STORE compo_int_rows INTO 'cql://thrift_ks/compo_int_copy?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.compo_int_copy+set+column1+%3D+%3F,+column2+%3D+%3F,+value+%3D+%3F' using CqlNativeStorage();");
pig.executeBatch();
pig.registerQuery("compocopy_int_rows = LOAD 'cassandra://thrift_ks/compo_int_copy?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("compocopy_int_rows = LOAD 'cql://thrift_ks/compo_int_copy?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_int_copy%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();");
//(clock,{((1,0),z),((1,30),zzzz),((2,30),daddy?),((6,30),coffee...)})
it = pig.openIterator("compocopy_int_rows");
count = 0;
@ -627,32 +546,26 @@ public class ThriftColumnFamilyTest extends PigTestBase
}
@Test
public void testCassandraStorageCompositeKeyCF() throws IOException
public void testCqlNativeStorageCompositeKeyCF() throws IOException
{
//Test CompositeKey
pig.registerQuery("compokeys = load 'cassandra://thrift_ks/compo_key?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("compokeys = filter compokeys by key.$1 == 40;");
//((clock,40),{(6,coffee...)})
pig.registerQuery("compokeys = load 'cql://thrift_ks/compo_key?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_key%20where%20token(key,column1)%20%3E%20%3F%20and%20token(key,column1)%20%3C%3D%20%3F' using CqlNativeStorage();");
pig.registerQuery("compokeys = filter compokeys by column1 == 40;");
//(clock,40,6,coffee...)
Iterator<Tuple> it = pig.openIterator("compokeys");
if (it.hasNext()) {
Tuple t = it.next();
Tuple key = (Tuple) t.get(0);
Assert.assertEquals(key.get(0), "clock");
Assert.assertEquals(key.get(1), 40L);
DataBag columns = (DataBag) t.get(1);
Iterator<Tuple> iter = columns.iterator();
if (iter.hasNext())
{
Tuple t1 = iter.next();
Assert.assertEquals(t1.get(0), 6L);
Assert.assertEquals(t1.get(1), "coffee...");
}
Assert.assertEquals(t.get(0), "clock");
Assert.assertEquals(t.get(1), 40L);
Assert.assertEquals(t.get(2), 6L);
Assert.assertEquals(t.get(3), "coffee...");
}
pig.setBatchOn();
pig.registerQuery("compo_key_rows = LOAD 'cassandra://thrift_ks/compo_key?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("STORE compo_key_rows INTO 'cassandra://thrift_ks/compo_key_copy?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("compo_key_rows = LOAD 'cql://thrift_ks/compo_key?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_key%20where%20token(key,column1)%20%3E%20%3F%20and%20token(key,column1)%20%3C%3D%20%3F' using CqlNativeStorage();");
pig.registerQuery("compo_key_rows = FOREACH compo_key_rows GENERATE TOTUPLE(TOTUPLE('key',key),TOTUPLE('column1',column1),TOTUPLE('column2',column2)),TOTUPLE(value);");
pig.registerQuery("STORE compo_key_rows INTO 'cql://thrift_ks/compo_key_copy?" + defaultParameters + nativeParameters + "&output_query=UPDATE+thrift_ks.compo_key_copy+set+value+%3D+%3F' using CqlNativeStorage();");
pig.executeBatch();
pig.registerQuery("compo_key_copy_rows = LOAD 'cassandra://thrift_ks/compo_key_copy?" + defaultParameters + "' using CassandraStorage();");
pig.registerQuery("compo_key_copy_rows = LOAD 'cql://thrift_ks/compo_key_copy?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compo_key_copy%20where%20token(key,column1)%20%3E%20%3F%20and%20token(key,column1)%20%3C%3D%20%3F' using CqlNativeStorage();");
//((clock,10),{(1,z)})
//((clock,20),{(1,zzzz)})
//((clock,30),{(2,daddy?)})
@ -662,66 +575,43 @@ public class ThriftColumnFamilyTest extends PigTestBase
while (it.hasNext()) {
Tuple t = it.next();
count ++;
Tuple key = (Tuple) t.get(0);
if ("clock".equals(key.get(0)) && (Long) key.get(1) == 10L)
if ("clock".equals(t.get(0)) && (Long) t.get(1) == 10L)
{
DataBag columns = (DataBag) t.get(1);
Iterator<Tuple> iter = columns.iterator();
if (iter.hasNext())
{
Tuple t1 = iter.next();
Assert.assertEquals(t1.get(0), 1L);
Assert.assertEquals(t1.get(1), "z");
}
Assert.assertEquals(t.get(2), 1L);
Assert.assertEquals(t.get(3), "z");
}
else if ("clock".equals(key.get(0)) && (Long) key.get(1) == 40L)
else if ("clock".equals(t.get(0)) && (Long) t.get(1) == 40L)
{
DataBag columns = (DataBag) t.get(1);
Iterator<Tuple> iter = columns.iterator();
if (iter.hasNext())
{
Tuple t1 = iter.next();
Assert.assertEquals(t1.get(0), 6L);
Assert.assertEquals(t1.get(1), "coffee...");
}
Assert.assertEquals(t.get(2), 6L);
Assert.assertEquals(t.get(3), "coffee...");
}
else if ("clock".equals(key.get(0)) && (Long) key.get(1) == 20L)
else if ("clock".equals(t.get(0)) && (Long) t.get(1) == 20L)
{
DataBag columns = (DataBag) t.get(1);
Iterator<Tuple> iter = columns.iterator();
if (iter.hasNext())
{
Tuple t1 = iter.next();
Assert.assertEquals(t1.get(0), 1L);
Assert.assertEquals(t1.get(1), "zzzz");
}
Assert.assertEquals(t.get(2), 1L);
Assert.assertEquals(t.get(3), "zzzz");
}
else if ("clock".equals(key.get(0)) && (Long) key.get(1) == 30L)
else if ("clock".equals(t.get(0)) && (Long) t.get(1) == 30L)
{
DataBag columns = (DataBag) t.get(1);
Iterator<Tuple> iter = columns.iterator();
if (iter.hasNext())
{
Tuple t1 = iter.next();
Assert.assertEquals(t1.get(0), 2L);
Assert.assertEquals(t1.get(1), "daddy?");
}
Assert.assertEquals(t.get(2), 2L);
Assert.assertEquals(t.get(3), "daddy?");
}
}
Assert.assertEquals(4, count);
}
private String getColumnValue(String ks, String cf, String colName, String key, String validator) throws TException, IOException
private String getColumnValue(String ks, String cf, String colName, String key, String validator) throws IOException
{
Cassandra.Client client = getClient();
client.set_keyspace(ks);
Session client = getClient();
client.execute("USE " + ks);
ByteBuffer key_user_id = ByteBufferUtil.bytes(key);
ColumnPath cp = new ColumnPath(cf);
cp.column = ByteBufferUtil.bytes(colName);
String query = String.format("SELECT %s FROM %s WHERE key = '%s'", colName, cf, key);
// read
ColumnOrSuperColumn got = client.get(key_user_id, cp, ConsistencyLevel.ONE);
return parseType(validator).getString(got.getColumn().value);
ResultSet rows = client.execute(query);
Row row = rows.one();
if (row == null || row.isNull(0))
return null;
return parseType(validator).getString(row.getBytesUnsafe(0));
}
}