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

Remove org.apache.cassandra.hadoop code 

patch by Stefan Miklosovic; reviewed by David Capwell and Brandon Williams for CASSANDRA-18323
This commit is contained in:
Stefan Miklosovic 2023-03-13 12:25:46 +01:00
parent 0136374545
commit 592cbeaab0
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
20 changed files with 17 additions and 4554 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.build/cassandra-deps-template.xml
View File

@ -161,14 +161,6 @@
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-core</artifactId>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minicluster</artifactId>
</dependency>
<dependency>
<groupId>com.datastax.cassandra</groupId>
<artifactId>cassandra-driver-core</artifactId>

60
.build/parent-pom-template.xml
View File

@ -503,66 +503,6 @@
<version>8.40</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-core</artifactId>
<version>1.0.3</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<artifactId>servlet-api</artifactId>
<groupId>org.mortbay.jetty</groupId>
</exclusion>
<exclusion>
<artifactId>commons-logging</artifactId>
<groupId>commons-logging</groupId>
</exclusion>
<exclusion>
<artifactId>commons-lang</artifactId>
<groupId>commons-lang</groupId>
</exclusion>
<exclusion>
<artifactId>core</artifactId>
<groupId>org.eclipse.jdt</groupId>
</exclusion>
<exclusion>
<artifactId>ant</artifactId>
<groupId>ant</groupId>
</exclusion>
<exclusion>
<artifactId>junit</artifactId>
<groupId>junit</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-mapper-asl</artifactId>
<groupId>org.codehaus.jackson</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-api</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minicluster</artifactId>
<version>1.0.3</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<artifactId>asm</artifactId>
<groupId>asm</groupId>
</exclusion>
<exclusion>
<artifactId>jackson-mapper-asl</artifactId>
<groupId>org.codehaus.jackson</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-api</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>net.java.dev.jna</groupId>
<artifactId>jna</artifactId>

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0
* Remove org.apache.cassandra.hadoop code (CASSANDRA-18323)
* Remove deprecated CQL functions dateOf and unixTimestampOf (CASSANDRA-18328)
* Remove DateTieredCompactionStrategy (CASSANDRA-18043)
* Add system_views.max_sstable_size and system_views.max_sstable_duration tables (CASSANDRA-18333)

2
NEWS.txt
View File

@ -150,6 +150,8 @@ Upgrading
to TimeWindowCompactionStrategy before upgrading to this version.
- The deprecated functions `dateOf` and `unixTimestampOf` have been removed. They were deprecated and replaced by
`toTimestamp` and `toUnixTimestamp` in Cassandra 2.2.
- Hadoop integration is no longer available (CASSANDRA-18323). If you want to process Cassandra data by big data frameworks,
please upgrade your infrastructure to use Cassandra Spark connector.
Deprecation
-----------

126
src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java
View File

@ -1,126 +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 org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputSplit;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.EOFException;
import java.io.IOException;
import java.util.Arrays;
public class ColumnFamilySplit extends InputSplit implements Writable, org.apache.hadoop.mapred.InputSplit
{
private String startToken;
private String endToken;
private long length;
private String[] dataNodes;
@Deprecated
public ColumnFamilySplit(String startToken, String endToken, String[] dataNodes)
{
this(startToken, endToken, Long.MAX_VALUE, dataNodes);
}
public ColumnFamilySplit(String startToken, String endToken, long length, String[] dataNodes)
{
assert startToken != null;
assert endToken != null;
this.startToken = startToken;
this.endToken = endToken;
this.length = length;
this.dataNodes = dataNodes;
}
public String getStartToken()
{
return startToken;
}
public String getEndToken()
{
return endToken;
}
// getLength and getLocations satisfy the InputSplit abstraction
public long getLength()
{
return length;
}
public String[] getLocations()
{
return dataNodes;
}
// This should only be used by KeyspaceSplit.read();
protected ColumnFamilySplit() {}
// These three methods are for serializing and deserializing
// KeyspaceSplits as needed by the Writable interface.
public void write(DataOutput out) throws IOException
{
out.writeUTF(startToken);
out.writeUTF(endToken);
out.writeInt(dataNodes.length);
for (String endpoint : dataNodes)
{
out.writeUTF(endpoint);
}
out.writeLong(length);
}
public void readFields(DataInput in) throws IOException
{
startToken = in.readUTF();
endToken = in.readUTF();
int numOfEndpoints = in.readInt();
dataNodes = new String[numOfEndpoints];
for(int i = 0; i < numOfEndpoints; i++)
{
dataNodes[i] = in.readUTF();
}
try
{
length = in.readLong();
}
catch (EOFException e)
{
//We must be deserializing in a mixed-version cluster.
}
}
@Override
public String toString()
{
return "ColumnFamilySplit(" +
"(" + startToken
+ ", '" + endToken + ']'
+ " @" + (dataNodes == null ? null : Arrays.asList(dataNodes)) + ')';
}
public static ColumnFamilySplit read(DataInput in) throws IOException
{
ColumnFamilySplit w = new ColumnFamilySplit();
w.readFields(in);
return w;
}
}

408
src/java/org/apache/cassandra/hadoop/ConfigHelper.java
View File

@ -1,408 +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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.hadoop.conf.Configuration;
public class ConfigHelper
{
private static final String INPUT_PARTITIONER_CONFIG = "cassandra.input.partitioner.class";
private static final String OUTPUT_PARTITIONER_CONFIG = "cassandra.output.partitioner.class";
private static final String INPUT_KEYSPACE_CONFIG = "cassandra.input.keyspace";
private static final String OUTPUT_KEYSPACE_CONFIG = "cassandra.output.keyspace";
private static final String INPUT_KEYSPACE_USERNAME_CONFIG = "cassandra.input.keyspace.username";
private static final String INPUT_KEYSPACE_PASSWD_CONFIG = "cassandra.input.keyspace.passwd";
private static final String OUTPUT_KEYSPACE_USERNAME_CONFIG = "cassandra.output.keyspace.username";
private static final String OUTPUT_KEYSPACE_PASSWD_CONFIG = "cassandra.output.keyspace.passwd";
private static final String INPUT_COLUMNFAMILY_CONFIG = "cassandra.input.columnfamily";
private static final String OUTPUT_COLUMNFAMILY_CONFIG = "mapreduce.output.basename"; //this must == OutputFormat.BASE_OUTPUT_NAME
private static final String INPUT_PREDICATE_CONFIG = "cassandra.input.predicate";
private static final String INPUT_KEYRANGE_CONFIG = "cassandra.input.keyRange";
private static final String INPUT_SPLIT_SIZE_CONFIG = "cassandra.input.split.size";
private static final String INPUT_SPLIT_SIZE_IN_MIB_CONFIG = "cassandra.input.split.size_mb";
private static final String INPUT_WIDEROWS_CONFIG = "cassandra.input.widerows";
private static final int DEFAULT_SPLIT_SIZE = 64 * 1024;
private static final String RANGE_BATCH_SIZE_CONFIG = "cassandra.range.batch.size";
private static final int DEFAULT_RANGE_BATCH_SIZE = 4096;
private static final String INPUT_INITIAL_ADDRESS = "cassandra.input.address";
private static final String OUTPUT_INITIAL_ADDRESS = "cassandra.output.address";
private static final String OUTPUT_INITIAL_PORT = "cassandra.output.port";
private static final String READ_CONSISTENCY_LEVEL = "cassandra.consistencylevel.read";
private static final String WRITE_CONSISTENCY_LEVEL = "cassandra.consistencylevel.write";
private static final String OUTPUT_COMPRESSION_CLASS = "cassandra.output.compression.class";
private static final String OUTPUT_COMPRESSION_CHUNK_LENGTH = "cassandra.output.compression.length";
private static final String OUTPUT_LOCAL_DC_ONLY = "cassandra.output.local.dc.only";
private static final String DEFAULT_CASSANDRA_NATIVE_PORT = "7000";
private static final Logger logger = LoggerFactory.getLogger(ConfigHelper.class);
/**
* Set the keyspace and column family for the input of this job.
*
* @param conf Job configuration you are about to run
* @param keyspace
* @param columnFamily
* @param widerows
*/
public static void setInputColumnFamily(Configuration conf, String keyspace, String columnFamily, boolean widerows)
{
if (keyspace == null)
throw new UnsupportedOperationException("keyspace may not be null");
if (columnFamily == null)
throw new UnsupportedOperationException("table may not be null");
conf.set(INPUT_KEYSPACE_CONFIG, keyspace);
conf.set(INPUT_COLUMNFAMILY_CONFIG, columnFamily);
conf.set(INPUT_WIDEROWS_CONFIG, String.valueOf(widerows));
}
/**
* Set the keyspace and column family for the input of this job.
*
* @param conf Job configuration you are about to run
* @param keyspace
* @param columnFamily
*/
public static void setInputColumnFamily(Configuration conf, String keyspace, String columnFamily)
{
setInputColumnFamily(conf, keyspace, columnFamily, false);
}
/**
* Set the keyspace for the output of this job.
*
* @param conf Job configuration you are about to run
* @param keyspace
*/
public static void setOutputKeyspace(Configuration conf, String keyspace)
{
if (keyspace == null)
throw new UnsupportedOperationException("keyspace may not be null");
conf.set(OUTPUT_KEYSPACE_CONFIG, keyspace);
}
/**
* Set the column family for the output of this job.
*
* @param conf Job configuration you are about to run
* @param columnFamily
*/
public static void setOutputColumnFamily(Configuration conf, String columnFamily)
{
conf.set(OUTPUT_COLUMNFAMILY_CONFIG, columnFamily);
}
/**
* Set the column family for the output of this job.
*
* @param conf Job configuration you are about to run
* @param keyspace
* @param columnFamily
*/
public static void setOutputColumnFamily(Configuration conf, String keyspace, String columnFamily)
{
setOutputKeyspace(conf, keyspace);
setOutputColumnFamily(conf, columnFamily);
}
/**
* The number of rows to request with each get range slices request.
* Too big and you can either get timeouts when it takes Cassandra too
* long to fetch all the data. Too small and the performance
* will be eaten up by the overhead of each request.
*
* @param conf Job configuration you are about to run
* @param batchsize Number of rows to request each time
*/
public static void setRangeBatchSize(Configuration conf, int batchsize)
{
conf.setInt(RANGE_BATCH_SIZE_CONFIG, batchsize);
}
/**
* The number of rows to request with each get range slices request.
* Too big and you can either get timeouts when it takes Cassandra too
* long to fetch all the data. Too small and the performance
* will be eaten up by the overhead of each request.
*
* @param conf Job configuration you are about to run
* @return Number of rows to request each time
*/
public static int getRangeBatchSize(Configuration conf)
{
return conf.getInt(RANGE_BATCH_SIZE_CONFIG, DEFAULT_RANGE_BATCH_SIZE);
}
/**
* Set the size of the input split.
* This affects the number of maps created, if the number is too small
* the overhead of each map will take up the bulk of the job time.
*
* @param conf Job configuration you are about to run
* @param splitsize Number of partitions in the input split
*/
public static void setInputSplitSize(Configuration conf, int splitsize)
{
conf.setInt(INPUT_SPLIT_SIZE_CONFIG, splitsize);
}
public static int getInputSplitSize(Configuration conf)
{
return conf.getInt(INPUT_SPLIT_SIZE_CONFIG, DEFAULT_SPLIT_SIZE);
}
/**
* Set the size of the input split. setInputSplitSize value is used if this is not set.
* This affects the number of maps created, if the number is too small
* the overhead of each map will take up the bulk of the job time.
*
* @param conf Job configuration you are about to run
* @param splitSizeMb Input split size in MiB
*/
public static void setInputSplitSizeInMb(Configuration conf, int splitSizeMb)
{
conf.setInt(INPUT_SPLIT_SIZE_IN_MIB_CONFIG, splitSizeMb);
}
/**
* cassandra.input.split.size will be used if the value is undefined or negative.
* @param conf Job configuration you are about to run
* @return split size in MiB or -1 if it is undefined.
*/
public static int getInputSplitSizeInMb(Configuration conf)
{
return conf.getInt(INPUT_SPLIT_SIZE_IN_MIB_CONFIG, -1);
}
/**
* Set the KeyRange to limit the rows.
* @param conf Job configuration you are about to run
*/
public static void setInputRange(Configuration conf, String startToken, String endToken)
{
conf.set(INPUT_KEYRANGE_CONFIG, startToken + "," + endToken);
}
/**
* The start and end token of the input key range as a pair.
*
* may be null if unset.
*/
public static Pair<String, String> getInputKeyRange(Configuration conf)
{
String str = conf.get(INPUT_KEYRANGE_CONFIG);
if (str == null)
return null;
String[] parts = str.split(",");
assert parts.length == 2;
return Pair.create(parts[0], parts[1]);
}
public static String getInputKeyspace(Configuration conf)
{
return conf.get(INPUT_KEYSPACE_CONFIG);
}
public static String getOutputKeyspace(Configuration conf)
{
return conf.get(OUTPUT_KEYSPACE_CONFIG);
}
public static void setInputKeyspaceUserNameAndPassword(Configuration conf, String username, String password)
{
setInputKeyspaceUserName(conf, username);
setInputKeyspacePassword(conf, password);
}
public static void setInputKeyspaceUserName(Configuration conf, String username)
{
conf.set(INPUT_KEYSPACE_USERNAME_CONFIG, username);
}
public static String getInputKeyspaceUserName(Configuration conf)
{
return conf.get(INPUT_KEYSPACE_USERNAME_CONFIG);
}
public static void setInputKeyspacePassword(Configuration conf, String password)
{
conf.set(INPUT_KEYSPACE_PASSWD_CONFIG, password);
}
public static String getInputKeyspacePassword(Configuration conf)
{
return conf.get(INPUT_KEYSPACE_PASSWD_CONFIG);
}
public static void setOutputKeyspaceUserNameAndPassword(Configuration conf, String username, String password)
{
setOutputKeyspaceUserName(conf, username);
setOutputKeyspacePassword(conf, password);
}
public static void setOutputKeyspaceUserName(Configuration conf, String username)
{
conf.set(OUTPUT_KEYSPACE_USERNAME_CONFIG, username);
}
public static String getOutputKeyspaceUserName(Configuration conf)
{
return conf.get(OUTPUT_KEYSPACE_USERNAME_CONFIG);
}
public static void setOutputKeyspacePassword(Configuration conf, String password)
{
conf.set(OUTPUT_KEYSPACE_PASSWD_CONFIG, password);
}
public static String getOutputKeyspacePassword(Configuration conf)
{
return conf.get(OUTPUT_KEYSPACE_PASSWD_CONFIG);
}
public static String getInputColumnFamily(Configuration conf)
{
return conf.get(INPUT_COLUMNFAMILY_CONFIG);
}
public static String getOutputColumnFamily(Configuration conf)
{
if (conf.get(OUTPUT_COLUMNFAMILY_CONFIG) != null)
return conf.get(OUTPUT_COLUMNFAMILY_CONFIG);
else
throw new UnsupportedOperationException("You must set the output column family using either setOutputColumnFamily or by adding a named output with MultipleOutputs");
}
public static boolean getInputIsWide(Configuration conf)
{
return Boolean.parseBoolean(conf.get(INPUT_WIDEROWS_CONFIG));
}
public static String getReadConsistencyLevel(Configuration conf)
{
return conf.get(READ_CONSISTENCY_LEVEL, "LOCAL_ONE");
}
public static void setReadConsistencyLevel(Configuration conf, String consistencyLevel)
{
conf.set(READ_CONSISTENCY_LEVEL, consistencyLevel);
}
public static String getWriteConsistencyLevel(Configuration conf)
{
return conf.get(WRITE_CONSISTENCY_LEVEL, "LOCAL_ONE");
}
public static void setWriteConsistencyLevel(Configuration conf, String consistencyLevel)
{
conf.set(WRITE_CONSISTENCY_LEVEL, consistencyLevel);
}
public static String getInputInitialAddress(Configuration conf)
{
return conf.get(INPUT_INITIAL_ADDRESS);
}
public static void setInputInitialAddress(Configuration conf, String address)
{
conf.set(INPUT_INITIAL_ADDRESS, address);
}
public static void setInputPartitioner(Configuration conf, String classname)
{
conf.set(INPUT_PARTITIONER_CONFIG, classname);
}
public static IPartitioner getInputPartitioner(Configuration conf)
{
return FBUtilities.newPartitioner(conf.get(INPUT_PARTITIONER_CONFIG));
}
public static String getOutputInitialAddress(Configuration conf)
{
return conf.get(OUTPUT_INITIAL_ADDRESS);
}
public static void setOutputInitialPort(Configuration conf, Integer port)
{
conf.set(OUTPUT_INITIAL_PORT, port.toString());
}
public static Integer getOutputInitialPort(Configuration conf)
{
return Integer.valueOf(conf.get(OUTPUT_INITIAL_PORT, DEFAULT_CASSANDRA_NATIVE_PORT));
}
public static void setOutputInitialAddress(Configuration conf, String address)
{
conf.set(OUTPUT_INITIAL_ADDRESS, address);
}
public static void setOutputPartitioner(Configuration conf, String classname)
{
conf.set(OUTPUT_PARTITIONER_CONFIG, classname);
}
public static IPartitioner getOutputPartitioner(Configuration conf)
{
return FBUtilities.newPartitioner(conf.get(OUTPUT_PARTITIONER_CONFIG));
}
public static String getOutputCompressionClass(Configuration conf)
{
return conf.get(OUTPUT_COMPRESSION_CLASS);
}
public static String getOutputCompressionChunkLength(Configuration conf)
{
return conf.get(OUTPUT_COMPRESSION_CHUNK_LENGTH, String.valueOf(CompressionParams.DEFAULT_CHUNK_LENGTH));
}
public static void setOutputCompressionClass(Configuration conf, String classname)
{
conf.set(OUTPUT_COMPRESSION_CLASS, classname);
}
public static void setOutputCompressionChunkLength(Configuration conf, String length)
{
conf.set(OUTPUT_COMPRESSION_CHUNK_LENGTH, length);
}
public static boolean getOutputLocalDCOnly(Configuration conf)
{
return Boolean.parseBoolean(conf.get(OUTPUT_LOCAL_DC_ONLY, "false"));
}
public static void setOutputLocalDCOnly(Configuration conf, boolean localDCOnly)
{
conf.set(OUTPUT_LOCAL_DC_ONLY, Boolean.toString(localDCOnly));
}
}

350
src/java/org/apache/cassandra/hadoop/HadoopCompat.java
View File

@ -1,350 +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.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.JobID;
import org.apache.hadoop.mapreduce.MapContext;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.StatusReporter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;
/*
* This is based on ContextFactory.java from hadoop-2.0.x sources.
*/
/**
* Utility methods to allow applications to deal with inconsistencies between
* MapReduce Context Objects API between Hadoop 1.x and 2.x.
*/
public class HadoopCompat
{
private static final boolean useV21;
private static final Constructor<?> JOB_CONTEXT_CONSTRUCTOR;
private static final Constructor<?> TASK_CONTEXT_CONSTRUCTOR;
private static final Constructor<?> MAP_CONTEXT_CONSTRUCTOR;
private static final Constructor<?> GENERIC_COUNTER_CONSTRUCTOR;
private static final Field READER_FIELD;
private static final Field WRITER_FIELD;
private static final Method GET_CONFIGURATION_METHOD;
private static final Method SET_STATUS_METHOD;
private static final Method GET_COUNTER_METHOD;
private static final Method INCREMENT_COUNTER_METHOD;
private static final Method GET_TASK_ATTEMPT_ID;
private static final Method PROGRESS_METHOD;
static
{
boolean v21 = true;
final String PACKAGE = "org.apache.hadoop.mapreduce";
try
{
Class.forName(PACKAGE + ".task.JobContextImpl");
} catch (ClassNotFoundException cnfe)
{
v21 = false;
}
useV21 = v21;
Class<?> jobContextCls;
Class<?> taskContextCls;
Class<?> taskIOContextCls;
Class<?> mapContextCls;
Class<?> genericCounterCls;
try
{
if (v21)
{
jobContextCls =
Class.forName(PACKAGE+".task.JobContextImpl");
taskContextCls =
Class.forName(PACKAGE+".task.TaskAttemptContextImpl");
taskIOContextCls =
Class.forName(PACKAGE+".task.TaskInputOutputContextImpl");
mapContextCls = Class.forName(PACKAGE + ".task.MapContextImpl");
genericCounterCls = Class.forName(PACKAGE+".counters.GenericCounter");
}
else
{
jobContextCls =
Class.forName(PACKAGE+".JobContext");
taskContextCls =
Class.forName(PACKAGE+".TaskAttemptContext");
taskIOContextCls =
Class.forName(PACKAGE+".TaskInputOutputContext");
mapContextCls = Class.forName(PACKAGE + ".MapContext");
genericCounterCls =
Class.forName("org.apache.hadoop.mapred.Counters$Counter");
}
} catch (ClassNotFoundException e)
{
throw new IllegalArgumentException("Can't find class", e);
}
try
{
JOB_CONTEXT_CONSTRUCTOR =
jobContextCls.getConstructor(Configuration.class, JobID.class);
JOB_CONTEXT_CONSTRUCTOR.setAccessible(true);
TASK_CONTEXT_CONSTRUCTOR =
taskContextCls.getConstructor(Configuration.class,
TaskAttemptID.class);
TASK_CONTEXT_CONSTRUCTOR.setAccessible(true);
GENERIC_COUNTER_CONSTRUCTOR =
genericCounterCls.getDeclaredConstructor(String.class,
String.class,
Long.TYPE);
GENERIC_COUNTER_CONSTRUCTOR.setAccessible(true);
if (useV21)
{
MAP_CONTEXT_CONSTRUCTOR =
mapContextCls.getDeclaredConstructor(Configuration.class,
TaskAttemptID.class,
RecordReader.class,
RecordWriter.class,
OutputCommitter.class,
StatusReporter.class,
InputSplit.class);
Method get_counter;
try
{
get_counter = Class.forName(PACKAGE + ".TaskAttemptContext").getMethod("getCounter", String.class,
String.class);
}
catch (Exception e)
{
get_counter = Class.forName(PACKAGE + ".TaskInputOutputContext").getMethod("getCounter",
String.class, String.class);
}
GET_COUNTER_METHOD = get_counter;
}
else
{
MAP_CONTEXT_CONSTRUCTOR =
mapContextCls.getConstructor(Configuration.class,
TaskAttemptID.class,
RecordReader.class,
RecordWriter.class,
OutputCommitter.class,
StatusReporter.class,
InputSplit.class);
GET_COUNTER_METHOD = Class.forName(PACKAGE+".TaskInputOutputContext")
.getMethod("getCounter", String.class, String.class);
}
MAP_CONTEXT_CONSTRUCTOR.setAccessible(true);
READER_FIELD = mapContextCls.getDeclaredField("reader");
READER_FIELD.setAccessible(true);
WRITER_FIELD = taskIOContextCls.getDeclaredField("output");
WRITER_FIELD.setAccessible(true);
GET_CONFIGURATION_METHOD = Class.forName(PACKAGE+".JobContext")
.getMethod("getConfiguration");
SET_STATUS_METHOD = Class.forName(PACKAGE+".TaskAttemptContext")
.getMethod("setStatus", String.class);
GET_TASK_ATTEMPT_ID = Class.forName(PACKAGE+".TaskAttemptContext")
.getMethod("getTaskAttemptID");
INCREMENT_COUNTER_METHOD = Class.forName(PACKAGE+".Counter")
.getMethod("increment", Long.TYPE);
PROGRESS_METHOD = Class.forName(PACKAGE+".TaskAttemptContext")
.getMethod("progress");
}
catch (SecurityException e)
{
throw new IllegalArgumentException("Can't run constructor ", e);
}
catch (NoSuchMethodException e)
{
throw new IllegalArgumentException("Can't find constructor ", e);
}
catch (NoSuchFieldException e)
{
throw new IllegalArgumentException("Can't find field ", e);
}
catch (ClassNotFoundException e)
{
throw new IllegalArgumentException("Can't find class", e);
}
}
/**
* True if runtime Hadoop version is 2.x, false otherwise.
*/
public static boolean isVersion2x()
{
return useV21;
}
private static Object newInstance(Constructor<?> constructor, Object...args)
{
try
{
return constructor.newInstance(args);
}
catch (InstantiationException e)
{
throw new IllegalArgumentException("Can't instantiate " + constructor, e);
}
catch (IllegalAccessException e)
{
throw new IllegalArgumentException("Can't instantiate " + constructor, e);
}
catch (InvocationTargetException e)
{
throw new IllegalArgumentException("Can't instantiate " + constructor, e);
}
}
/**
* Creates JobContext from a JobConf and jobId using the correct constructor
* for based on Hadoop version. <code>jobId</code> could be null.
*/
public static JobContext newJobContext(Configuration conf, JobID jobId) {
return (JobContext) newInstance(JOB_CONTEXT_CONSTRUCTOR, conf, jobId);
}
/**
* Creates TaskAttempContext from a JobConf and jobId using the correct
* constructor for based on Hadoop version.
*/
public static TaskAttemptContext newTaskAttemptContext(
Configuration conf, TaskAttemptID taskAttemptId) {
return (TaskAttemptContext)
newInstance(TASK_CONTEXT_CONSTRUCTOR, conf, taskAttemptId);
}
/**
* Instantiates MapContext under Hadoop 1 and MapContextImpl under Hadoop 2.
*/
public static MapContext newMapContext(Configuration conf,
TaskAttemptID taskAttemptID,
RecordReader recordReader,
RecordWriter recordWriter,
OutputCommitter outputCommitter,
StatusReporter statusReporter,
InputSplit inputSplit) {
return (MapContext) newInstance(MAP_CONTEXT_CONSTRUCTOR,
conf, taskAttemptID, recordReader, recordWriter, outputCommitter,
statusReporter, inputSplit);
}
/**
* @return with Hadoop 2 : <code>new GenericCounter(args)</code>,<br>
* with Hadoop 1 : <code>new Counter(args)</code>
*/
public static Counter newGenericCounter(String name, String displayName, long value)
{
try
{
return (Counter)
GENERIC_COUNTER_CONSTRUCTOR.newInstance(name, displayName, value);
}
catch (InstantiationException | IllegalAccessException | InvocationTargetException e)
{
throw new IllegalArgumentException("Can't instantiate Counter", e);
}
}
/**
* Invokes a method and rethrows any exception as runtime excetpions.
*/
private static Object invoke(Method method, Object obj, Object... args)
{
try
{
return method.invoke(obj, args);
}
catch (IllegalAccessException | InvocationTargetException e)
{
throw new IllegalArgumentException("Can't invoke method " + method.getName(), e);
}
}
/**
* Invoke getConfiguration() on JobContext. Works with both
* Hadoop 1 and 2.
*/
public static Configuration getConfiguration(JobContext context)
{
return (Configuration) invoke(GET_CONFIGURATION_METHOD, context);
}
/**
* Invoke setStatus() on TaskAttemptContext. Works with both
* Hadoop 1 and 2.
*/
public static void setStatus(TaskAttemptContext context, String status)
{
invoke(SET_STATUS_METHOD, context, status);
}
/**
* returns TaskAttemptContext.getTaskAttemptID(). Works with both
* Hadoop 1 and 2.
*/
public static TaskAttemptID getTaskAttemptID(TaskAttemptContext taskContext)
{
return (TaskAttemptID) invoke(GET_TASK_ATTEMPT_ID, taskContext);
}
/**
* Invoke getCounter() on TaskInputOutputContext. Works with both
* Hadoop 1 and 2.
*/
public static Counter getCounter(TaskInputOutputContext context,
String groupName, String counterName)
{
return (Counter) invoke(GET_COUNTER_METHOD, context, groupName, counterName);
}
/**
* Invoke TaskAttemptContext.progress(). Works with both
* Hadoop 1 and 2.
*/
public static void progress(TaskAttemptContext context)
{
invoke(PROGRESS_METHOD, context);
}
/**
* Increment the counter. Works with both Hadoop 1 and 2
*/
public static void incrementCounter(Counter counter, long increment)
{
// incrementing a count might be called often. Might be affected by
// cost of invoke(). might be good candidate to handle in a shim.
// (TODO Raghu) figure out how achieve such a build with maven
invoke(INCREMENT_COUNTER_METHOD, counter, increment);
}
}

85
src/java/org/apache/cassandra/hadoop/ReporterWrapper.java
View File

@ -1,85 +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 org.apache.hadoop.mapred.Counters;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.StatusReporter;
/**
* A reporter that works with both mapred and mapreduce APIs.
*/
public class ReporterWrapper extends StatusReporter implements Reporter
{
private Reporter wrappedReporter;
public ReporterWrapper(Reporter reporter)
{
wrappedReporter = reporter;
}
@Override
public Counters.Counter getCounter(Enum<?> anEnum)
{
return wrappedReporter.getCounter(anEnum);
}
@Override
public Counters.Counter getCounter(String s, String s1)
{
return wrappedReporter.getCounter(s, s1);
}
@Override
public void incrCounter(Enum<?> anEnum, long l)
{
wrappedReporter.incrCounter(anEnum, l);
}
@Override
public void incrCounter(String s, String s1, long l)
{
wrappedReporter.incrCounter(s, s1, l);
}
@Override
public InputSplit getInputSplit() throws UnsupportedOperationException
{
return wrappedReporter.getInputSplit();
}
@Override
public void progress()
{
wrappedReporter.progress();
}
// @Override
public float getProgress()
{
throw new UnsupportedOperationException();
}
@Override
public void setStatus(String s)
{
wrappedReporter.setStatus(s);
}
}

209
src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java
View File

@ -1,209 +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.cql3;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.util.Progressable;
/**
* The <code>CqlBulkOutputFormat</code> acts as a Hadoop-specific
* OutputFormat that allows reduce tasks to store keys (and corresponding
* bound variable values) as CQL rows (and respective columns) in a given
* table.
*
* <p>
* As is the case with the {@link org.apache.cassandra.hadoop.cql3.CqlOutputFormat},
* you need to set the prepared statement in your
* Hadoop job Configuration. The {@link CqlConfigHelper} class, through its
* {@link org.apache.cassandra.hadoop.ConfigHelper#setOutputPreparedStatement} method, is provided to make this
* simple.
* you need to set the Keyspace. The {@link ConfigHelper} class, through its
* {@link org.apache.cassandra.hadoop.ConfigHelper#setOutputColumnFamily} method, is provided to make this
* simple.
* </p>
*/
public class CqlBulkOutputFormat extends OutputFormat<Object, List<ByteBuffer>>
implements org.apache.hadoop.mapred.OutputFormat<Object, List<ByteBuffer>>
{
private static final String OUTPUT_CQL_SCHEMA_PREFIX = "cassandra.table.schema.";
private static final String OUTPUT_CQL_INSERT_PREFIX = "cassandra.table.insert.";
private static final String DELETE_SOURCE = "cassandra.output.delete.source";
private static final String TABLE_ALIAS_PREFIX = "cqlbulkoutputformat.table.alias.";
/** Fills the deprecated OutputFormat interface for streaming. */
@Deprecated
public CqlBulkRecordWriter getRecordWriter(FileSystem filesystem, JobConf job, String name, Progressable progress) throws IOException
{
return new CqlBulkRecordWriter(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.
* @throws IOException
*/
public CqlBulkRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
{
return new CqlBulkRecordWriter(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 setTable()");
}
}
/** 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);
}
@Override
public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException
{
return new NullOutputCommitter();
}
public static void setTableSchema(Configuration conf, String columnFamily, String schema)
{
conf.set(OUTPUT_CQL_SCHEMA_PREFIX + columnFamily, schema);
}
public static void setTableInsertStatement(Configuration conf, String columnFamily, String insertStatement)
{
conf.set(OUTPUT_CQL_INSERT_PREFIX + columnFamily, insertStatement);
}
public static String getTableSchema(Configuration conf, String columnFamily)
{
String schema = conf.get(OUTPUT_CQL_SCHEMA_PREFIX + columnFamily);
if (schema == null)
{
throw new UnsupportedOperationException("You must set the Table schema using setTableSchema.");
}
return schema;
}
public static String getTableInsertStatement(Configuration conf, String columnFamily)
{
String insert = conf.get(OUTPUT_CQL_INSERT_PREFIX + columnFamily);
if (insert == null)
{
throw new UnsupportedOperationException("You must set the Table insert statement using setTableSchema.");
}
return insert;
}
public static void setDeleteSourceOnSuccess(Configuration conf, boolean deleteSrc)
{
conf.setBoolean(DELETE_SOURCE, deleteSrc);
}
public static boolean getDeleteSourceOnSuccess(Configuration conf)
{
return conf.getBoolean(DELETE_SOURCE, false);
}
public static void setTableAlias(Configuration conf, String alias, String columnFamily)
{
conf.set(TABLE_ALIAS_PREFIX + alias, columnFamily);
}
public static String getTableForAlias(Configuration conf, String alias)
{
return conf.get(TABLE_ALIAS_PREFIX + alias);
}
/**
* Set the hosts to ignore as comma delimited values.
* Data will not be bulk loaded onto the ignored nodes.
* @param conf job configuration
* @param ignoreNodesCsv a comma delimited list of nodes to ignore
*/
public static void setIgnoreHosts(Configuration conf, String ignoreNodesCsv)
{
conf.set(CqlBulkRecordWriter.IGNORE_HOSTS, ignoreNodesCsv);
}
/**
* Set the hosts to ignore. Data will not be bulk loaded onto the ignored nodes.
* @param conf job configuration
* @param ignoreNodes the nodes to ignore
*/
public static void setIgnoreHosts(Configuration conf, String... ignoreNodes)
{
conf.setStrings(CqlBulkRecordWriter.IGNORE_HOSTS, ignoreNodes);
}
/**
* Get the hosts to ignore as a collection of strings
* @param conf job configuration
* @return the nodes to ignore as a collection of stirngs
*/
public static Collection<String> getIgnoreHosts(Configuration conf)
{
return conf.getStringCollection(CqlBulkRecordWriter.IGNORE_HOSTS);
}
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) { }
}
}

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

@ -1,333 +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.cql3;
import java.io.Closeable;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.google.common.net.HostAndPort;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.io.sstable.CQLSSTableWriter;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.TableMetadataRef;
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;
import org.apache.hadoop.util.Progressable;
import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR;
/**
* The <code>CqlBulkRecordWriter</code> maps the output &lt;key, value&gt;
* pairs to a Cassandra column family. In particular, it applies the binded variables
* in the value to the prepared statement, which it associates with the key, and in
* turn the responsible endpoint.
*
* <p>
* Furthermore, this writer groups the cql queries by the endpoint responsible for
* the rows being affected. This allows the cql queries to be executed in parallel,
* directly to a responsible endpoint.
* </p>
*
* @see CqlBulkOutputFormat
*/
public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
implements org.apache.hadoop.mapred.RecordWriter<Object, List<ByteBuffer>>
{
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";
public final static String IGNORE_HOSTS = "mapreduce.output.bulkoutputformat.ignorehosts";
private final Logger logger = LoggerFactory.getLogger(CqlBulkRecordWriter.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;
protected final Set<InetAddressAndPort> ignores = new HashSet<>();
private String keyspace;
private String table;
private String schema;
private String insertStatement;
private File outputDir;
private boolean deleteSrc;
private IPartitioner partitioner;
CqlBulkRecordWriter(TaskAttemptContext context) throws IOException
{
this(HadoopCompat.getConfiguration(context));
this.context = context;
setConfigs();
}
CqlBulkRecordWriter(Configuration conf, Progressable progress) throws IOException
{
this(conf);
this.progress = progress;
setConfigs();
}
CqlBulkRecordWriter(Configuration conf) throws IOException
{
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"));
setConfigs();
}
private void setConfigs() throws IOException
{
// if anything is missing, exceptions will be thrown here, instead of on write()
keyspace = ConfigHelper.getOutputKeyspace(conf);
table = ConfigHelper.getOutputColumnFamily(conf);
// check if table is aliased
String aliasedCf = CqlBulkOutputFormat.getTableForAlias(conf, table);
if (aliasedCf != null)
table = aliasedCf;
schema = CqlBulkOutputFormat.getTableSchema(conf, table);
insertStatement = CqlBulkOutputFormat.getTableInsertStatement(conf, table);
outputDir = getTableDirectory();
deleteSrc = CqlBulkOutputFormat.getDeleteSourceOnSuccess(conf);
try
{
partitioner = ConfigHelper.getInputPartitioner(conf);
}
catch (Exception e)
{
partitioner = Murmur3Partitioner.instance;
}
try
{
for (String hostToIgnore : CqlBulkOutputFormat.getIgnoreHosts(conf))
ignores.add(InetAddressAndPort.getByName(hostToIgnore));
}
catch (UnknownHostException e)
{
throw new RuntimeException(("Unknown host: " + e.getMessage()));
}
}
protected String getOutputLocation() throws IOException
{
String dir = conf.get(OUTPUT_LOCATION, JAVA_IO_TMPDIR.getString());
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 prepareWriter() throws IOException
{
if (writer == null)
{
writer = CQLSSTableWriter.builder()
.forTable(schema)
.using(insertStatement)
.withPartitioner(ConfigHelper.getOutputPartitioner(conf))
.inDirectory(outputDir)
.withBufferSizeInMiB(Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64")))
.withPartitioner(partitioner)
.build();
}
if (loader == null)
{
ExternalClient externalClient = new ExternalClient(conf);
externalClient.setTableMetadata(TableMetadataRef.forOfflineTools(CreateTableStatement.parse(schema, keyspace).build()));
loader = new SSTableLoader(outputDir, externalClient, new NullOutputHandler())
{
@Override
public void onSuccess(StreamState finalState)
{
if (deleteSrc)
FileUtils.deleteRecursive(outputDir);
}
};
}
}
/**
* <p>
* The column values must correspond to the order in which
* they appear in the insert stored procedure.
*
* Key is not used, so it can be null or any object.
* </p>
*
* @param key
* any object or null.
* @param values
* the values to write.
* @throws IOException
*/
@Override
public void write(Object key, List<ByteBuffer> values) throws IOException
{
prepareWriter();
try
{
((CQLSSTableWriter) writer).rawAddRow(values);
if (null != progress)
progress.progress();
if (null != context)
HadoopCompat.progress(context);
}
catch (InvalidRequestException e)
{
throw new IOException("Error adding row with key: " + key, e);
}
}
private File getTableDirectory() throws IOException
{
File dir = new File(String.format("%s%s%s%s%s-%s", getOutputLocation(), File.pathSeparator(), keyspace, File.pathSeparator(), table, UUID.randomUUID().toString()));
if (!dir.exists() && !dir.tryCreateDirectories())
{
throw new IOException("Failed to created output directory: " + dir);
}
return dir;
}
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException
{
close();
}
/** Fills the deprecated RecordWriter interface for streaming. */
@Deprecated
public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException
{
close();
}
private void close() throws IOException
{
if (writer != null)
{
writer.close();
Future<StreamState> future = loader.stream(ignores);
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());
}
}
}
public static class ExternalClient extends NativeSSTableLoaderClient
{
public ExternalClient(Configuration conf)
{
super(resolveHostAddresses(conf),
ConfigHelper.getOutputInitialPort(conf),
ConfigHelper.getOutputKeyspaceUserName(conf),
ConfigHelper.getOutputKeyspacePassword(conf),
CqlConfigHelper.getSSLOptions(conf).orNull());
}
private static Collection<InetSocketAddress> resolveHostAddresses(Configuration conf)
{
Set<InetSocketAddress> addresses = new HashSet<>();
int port = CqlConfigHelper.getOutputNativePort(conf);
for (String host : ConfigHelper.getOutputInitialAddress(conf).split(","))
{
try
{
HostAndPort hap = HostAndPort.fromString(host);
addresses.add(new InetSocketAddress(InetAddress.getByName(hap.getHost()), hap.getPortOrDefault(port)));
}
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(Throwable th, String msg) {}
}
}

109
src/java/org/apache/cassandra/hadoop/cql3/CqlClientHelper.java
View File

@ -1,109 +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.cql3;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.Token;
import com.datastax.driver.core.TokenRange;
public class CqlClientHelper
{
private CqlClientHelper()
{
}
public static Map<TokenRange, List<Host>> getLocalPrimaryRangeForDC(String keyspace, Metadata metadata, String targetDC)
{
Objects.requireNonNull(keyspace, "keyspace");
Objects.requireNonNull(metadata, "metadata");
Objects.requireNonNull(targetDC, "targetDC");
// In 2.1 the logic was to have a set of nodes used as a seed, they were used to query
// client.describe_local_ring(keyspace) -> List<TokenRange>; this should include all nodes in the local dc.
// TokenRange contained the endpoints in order, so .endpoints.get(0) is the primary owner
// Client does not have a similar API, instead it returns Set<Host>. To replicate this we first need
// to compute the primary owners, then add in the replicas
List<Token> tokens = new ArrayList<>();
Map<Token, Host> tokenToHost = new HashMap<>();
for (Host host : metadata.getAllHosts())
{
if (!targetDC.equals(host.getDatacenter()))
continue;
for (Token token : host.getTokens())
{
Host previous = tokenToHost.putIfAbsent(token, host);
if (previous != null)
throw new IllegalStateException("Two hosts share the same token; hosts " + host.getHostId() + ":"
+ host.getTokens() + ", " + previous.getHostId() + ":" + previous.getTokens());
tokens.add(token);
}
}
Collections.sort(tokens);
Map<TokenRange, List<Host>> rangeToReplicas = new HashMap<>();
// The first token in the ring uses the last token as its 'start', handle this here to simplify the loop
Token start = tokens.get(tokens.size() - 1);
Token end = tokens.get(0);
addRange(keyspace, metadata, tokenToHost, rangeToReplicas, start, end);
for (int i = 1; i < tokens.size(); i++)
{
start = tokens.get(i - 1);
end = tokens.get(i);
addRange(keyspace, metadata, tokenToHost, rangeToReplicas, start, end);
}
return rangeToReplicas;
}
private static void addRange(String keyspace,
Metadata metadata,
Map<Token, Host> tokenToHost,
Map<TokenRange, List<Host>> rangeToReplicas,
Token start, Token end)
{
Host host = tokenToHost.get(end);
String dc = host.getDatacenter();
TokenRange range = metadata.newTokenRange(start, end);
List<Host> replicas = new ArrayList<>();
replicas.add(host);
// get all the replicas for the specific DC
for (Host replica : metadata.getReplicas(keyspace, range))
{
if (dc.equals(replica.getDatacenter()) && !host.equals(replica))
replicas.add(replica);
}
List<Host> previous = rangeToReplicas.put(range, replicas);
if (previous != null)
throw new IllegalStateException("Two hosts (" + host + ", " + previous + ") map to the same token range: " + range);
}
}

654
src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java
View File

@ -1,654 +0,0 @@
package org.apache.cassandra.hadoop.cql3;
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
import java.nio.file.Files;
import java.io.InputStream;
import java.io.IOException;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.SecureRandom;
import java.security.UnrecoverableKeyException;
import java.security.cert.CertificateException;
import java.util.Arrays;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
import com.google.common.base.Optional;
import org.apache.commons.lang3.StringUtils;
import com.datastax.driver.core.AuthProvider;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.JdkSSLOptions;
import com.datastax.driver.core.PlainTextAuthProvider;
import com.datastax.driver.core.ProtocolVersion;
import com.datastax.driver.core.policies.LoadBalancingPolicy;
import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.ProtocolOptions;
import com.datastax.driver.core.QueryOptions;
import com.datastax.driver.core.SSLOptions;
import com.datastax.driver.core.SocketOptions;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.io.util.File;
import org.apache.hadoop.conf.Configuration;
public class CqlConfigHelper
{
private static final String INPUT_CQL_COLUMNS_CONFIG = "cassandra.input.columnfamily.columns";
private static final String INPUT_CQL_PAGE_ROW_SIZE_CONFIG = "cassandra.input.page.row.size";
private static final String INPUT_CQL_WHERE_CLAUSE_CONFIG = "cassandra.input.where.clause";
private static final String INPUT_CQL = "cassandra.input.cql";
private static final String USERNAME = "cassandra.username";
private static final String PASSWORD = "cassandra.password";
private static final String INPUT_NATIVE_PORT = "cassandra.input.native.port";
private static final String INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST = "cassandra.input.native.core.connections.per.host";
private static final String INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST = "cassandra.input.native.max.connections.per.host";
private static final String INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION = "cassandra.input.native.max.simult.reqs.per.connection";
private static final String INPUT_NATIVE_CONNECTION_TIMEOUT = "cassandra.input.native.connection.timeout";
private static final String INPUT_NATIVE_READ_CONNECTION_TIMEOUT = "cassandra.input.native.read.connection.timeout";
private static final String INPUT_NATIVE_RECEIVE_BUFFER_SIZE = "cassandra.input.native.receive.buffer.size";
private static final String INPUT_NATIVE_SEND_BUFFER_SIZE = "cassandra.input.native.send.buffer.size";
private static final String INPUT_NATIVE_SOLINGER = "cassandra.input.native.solinger";
private static final String INPUT_NATIVE_TCP_NODELAY = "cassandra.input.native.tcp.nodelay";
private static final String INPUT_NATIVE_REUSE_ADDRESS = "cassandra.input.native.reuse.address";
private static final String INPUT_NATIVE_KEEP_ALIVE = "cassandra.input.native.keep.alive";
private static final String INPUT_NATIVE_AUTH_PROVIDER = "cassandra.input.native.auth.provider";
private static final String INPUT_NATIVE_SSL_TRUST_STORE_PATH = "cassandra.input.native.ssl.trust.store.path";
private static final String INPUT_NATIVE_SSL_KEY_STORE_PATH = "cassandra.input.native.ssl.key.store.path";
private static final String INPUT_NATIVE_SSL_TRUST_STORE_PASSWARD = "cassandra.input.native.ssl.trust.store.password";
private static final String INPUT_NATIVE_SSL_KEY_STORE_PASSWARD = "cassandra.input.native.ssl.key.store.password";
private static final String INPUT_NATIVE_SSL_CIPHER_SUITES = "cassandra.input.native.ssl.cipher.suites";
private static final String INPUT_NATIVE_PROTOCOL_VERSION = "cassandra.input.native.protocol.version";
private static final String OUTPUT_CQL = "cassandra.output.cql";
private static final String OUTPUT_NATIVE_PORT = "cassandra.output.native.port";
/**
* Set the CQL columns for the input of this job.
*
* @param conf Job configuration you are about to run
* @param columns
*/
public static void setInputColumns(Configuration conf, String columns)
{
if (columns == null || columns.isEmpty())
return;
conf.set(INPUT_CQL_COLUMNS_CONFIG, columns);
}
/**
* Set the CQL query Limit for the input of this job.
*
* @param conf Job configuration you are about to run
* @param cqlPageRowSize
*/
public static void setInputCQLPageRowSize(Configuration conf, String cqlPageRowSize)
{
if (cqlPageRowSize == null)
{
throw new UnsupportedOperationException("cql page row size may not be null");
}
conf.set(INPUT_CQL_PAGE_ROW_SIZE_CONFIG, cqlPageRowSize);
}
/**
* Set the CQL user defined where clauses for the input of this job.
*
* @param conf Job configuration you are about to run
* @param clauses
*/
public static void setInputWhereClauses(Configuration conf, String clauses)
{
if (clauses == null || clauses.isEmpty())
return;
conf.set(INPUT_CQL_WHERE_CLAUSE_CONFIG, clauses);
}
/**
* Set the CQL prepared statement for the output of this job.
*
* @param conf Job configuration you are about to run
* @param cql
*/
public static void setOutputCql(Configuration conf, String cql)
{
if (cql == null || cql.isEmpty())
return;
conf.set(OUTPUT_CQL, cql);
}
public static void setInputCql(Configuration conf, String cql)
{
if (cql == null || cql.isEmpty())
return;
conf.set(INPUT_CQL, cql);
}
public static void setUserNameAndPassword(Configuration conf, String username, String password)
{
if (StringUtils.isNotBlank(username))
{
conf.set(INPUT_NATIVE_AUTH_PROVIDER, PlainTextAuthProvider.class.getName());
conf.set(USERNAME, username);
conf.set(PASSWORD, password);
}
}
public static Optional<Integer> getInputCoreConnections(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST, conf);
}
public static Optional<Integer> getInputMaxConnections(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST, conf);
}
public static int getInputNativePort(Configuration conf)
{
return Integer.parseInt(conf.get(INPUT_NATIVE_PORT, "9042"));
}
public static int getOutputNativePort(Configuration conf)
{
return Integer.parseInt(conf.get(OUTPUT_NATIVE_PORT, "9042"));
}
public static Optional<Integer> getInputMaxSimultReqPerConnections(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION, conf);
}
public static Optional<Integer> getInputNativeConnectionTimeout(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_CONNECTION_TIMEOUT, conf);
}
public static Optional<Integer> getInputNativeReadConnectionTimeout(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_READ_CONNECTION_TIMEOUT, conf);
}
public static Optional<Integer> getInputNativeReceiveBufferSize(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_RECEIVE_BUFFER_SIZE, conf);
}
public static Optional<Integer> getInputNativeSendBufferSize(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_SEND_BUFFER_SIZE, conf);
}
public static Optional<Integer> getInputNativeSolinger(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_SOLINGER, conf);
}
public static Optional<Boolean> getInputNativeTcpNodelay(Configuration conf)
{
return getBooleanSetting(INPUT_NATIVE_TCP_NODELAY, conf);
}
public static Optional<Boolean> getInputNativeReuseAddress(Configuration conf)
{
return getBooleanSetting(INPUT_NATIVE_REUSE_ADDRESS, conf);
}
public static Optional<String> getInputNativeAuthProvider(Configuration conf)
{
return getStringSetting(INPUT_NATIVE_AUTH_PROVIDER, conf);
}
public static Optional<String> getInputNativeSSLTruststorePath(Configuration conf)
{
return getStringSetting(INPUT_NATIVE_SSL_TRUST_STORE_PATH, conf);
}
public static Optional<String> getInputNativeSSLKeystorePath(Configuration conf)
{
return getStringSetting(INPUT_NATIVE_SSL_KEY_STORE_PATH, conf);
}
public static Optional<String> getInputNativeSSLKeystorePassword(Configuration conf)
{
return getStringSetting(INPUT_NATIVE_SSL_KEY_STORE_PASSWARD, conf);
}
public static Optional<String> getInputNativeSSLTruststorePassword(Configuration conf)
{
return getStringSetting(INPUT_NATIVE_SSL_TRUST_STORE_PASSWARD, conf);
}
public static Optional<String> getInputNativeSSLCipherSuites(Configuration conf)
{
return getStringSetting(INPUT_NATIVE_SSL_CIPHER_SUITES, conf);
}
public static Optional<Boolean> getInputNativeKeepAlive(Configuration conf)
{
return getBooleanSetting(INPUT_NATIVE_KEEP_ALIVE, conf);
}
public static String getInputcolumns(Configuration conf)
{
return conf.get(INPUT_CQL_COLUMNS_CONFIG);
}
public static Optional<Integer> getInputPageRowSize(Configuration conf)
{
return getIntSetting(INPUT_CQL_PAGE_ROW_SIZE_CONFIG, conf);
}
public static String getInputWhereClauses(Configuration conf)
{
return conf.get(INPUT_CQL_WHERE_CLAUSE_CONFIG);
}
public static String getInputCql(Configuration conf)
{
return conf.get(INPUT_CQL);
}
public static String getOutputCql(Configuration conf)
{
return conf.get(OUTPUT_CQL);
}
private static Optional<Integer> getProtocolVersion(Configuration conf)
{
return getIntSetting(INPUT_NATIVE_PROTOCOL_VERSION, conf);
}
public static Cluster getInputCluster(String host, Configuration conf)
{
// this method has been left for backward compatibility
return getInputCluster(new String[] {host}, conf);
}
public static Cluster getInputCluster(String[] hosts, Configuration conf)
{
int port = getInputNativePort(conf);
return getCluster(hosts, conf, port);
}
public static Cluster getOutputCluster(String host, Configuration conf)
{
return getOutputCluster(new String[]{host}, conf);
}
public static Cluster getOutputCluster(String[] hosts, Configuration conf)
{
int port = getOutputNativePort(conf);
return getCluster(hosts, conf, port);
}
public static Cluster getCluster(String[] hosts, Configuration conf, int port)
{
Optional<AuthProvider> authProvider = getAuthProvider(conf);
Optional<SSLOptions> sslOptions = getSSLOptions(conf);
Optional<Integer> protocolVersion = getProtocolVersion(conf);
LoadBalancingPolicy loadBalancingPolicy = getReadLoadBalancingPolicy(hosts);
SocketOptions socketOptions = getReadSocketOptions(conf);
QueryOptions queryOptions = getReadQueryOptions(conf);
PoolingOptions poolingOptions = getReadPoolingOptions(conf);
Cluster.Builder builder = Cluster.builder()
.addContactPoints(hosts)
.withPort(port)
.withCompression(ProtocolOptions.Compression.NONE);
if (authProvider.isPresent())
builder.withAuthProvider(authProvider.get());
if (sslOptions.isPresent())
builder.withSSL(sslOptions.get());
if (protocolVersion.isPresent())
{
builder.withProtocolVersion(ProtocolVersion.fromInt(protocolVersion.get()));
}
builder.withLoadBalancingPolicy(loadBalancingPolicy)
.withSocketOptions(socketOptions)
.withQueryOptions(queryOptions)
.withPoolingOptions(poolingOptions);
return builder.build();
}
public static void setInputCoreConnections(Configuration conf, String connections)
{
conf.set(INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST, connections);
}
public static void setInputMaxConnections(Configuration conf, String connections)
{
conf.set(INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST, connections);
}
public static void setInputMaxSimultReqPerConnections(Configuration conf, String reqs)
{
conf.set(INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION, reqs);
}
public static void setInputNativeConnectionTimeout(Configuration conf, String timeout)
{
conf.set(INPUT_NATIVE_CONNECTION_TIMEOUT, timeout);
}
public static void setInputNativeReadConnectionTimeout(Configuration conf, String timeout)
{
conf.set(INPUT_NATIVE_READ_CONNECTION_TIMEOUT, timeout);
}
public static void setInputNativeReceiveBufferSize(Configuration conf, String size)
{
conf.set(INPUT_NATIVE_RECEIVE_BUFFER_SIZE, size);
}
public static void setInputNativeSendBufferSize(Configuration conf, String size)
{
conf.set(INPUT_NATIVE_SEND_BUFFER_SIZE, size);
}
public static void setInputNativeSolinger(Configuration conf, String solinger)
{
conf.set(INPUT_NATIVE_SOLINGER, solinger);
}
public static void setInputNativeTcpNodelay(Configuration conf, String tcpNodelay)
{
conf.set(INPUT_NATIVE_TCP_NODELAY, tcpNodelay);
}
public static void setInputNativeAuthProvider(Configuration conf, String authProvider)
{
conf.set(INPUT_NATIVE_AUTH_PROVIDER, authProvider);
}
public static void setInputNativeSSLTruststorePath(Configuration conf, String path)
{
conf.set(INPUT_NATIVE_SSL_TRUST_STORE_PATH, path);
}
public static void setInputNativeSSLKeystorePath(Configuration conf, String path)
{
conf.set(INPUT_NATIVE_SSL_KEY_STORE_PATH, path);
}
public static void setInputNativeSSLKeystorePassword(Configuration conf, String pass)
{
conf.set(INPUT_NATIVE_SSL_KEY_STORE_PASSWARD, pass);
}
public static void setInputNativeSSLTruststorePassword(Configuration conf, String pass)
{
conf.set(INPUT_NATIVE_SSL_TRUST_STORE_PASSWARD, pass);
}
public static void setInputNativeSSLCipherSuites(Configuration conf, String suites)
{
conf.set(INPUT_NATIVE_SSL_CIPHER_SUITES, suites);
}
public static void setInputNativeReuseAddress(Configuration conf, String reuseAddress)
{
conf.set(INPUT_NATIVE_REUSE_ADDRESS, reuseAddress);
}
public static void setInputNativeKeepAlive(Configuration conf, String keepAlive)
{
conf.set(INPUT_NATIVE_KEEP_ALIVE, keepAlive);
}
public static void setInputNativePort(Configuration conf, String port)
{
conf.set(INPUT_NATIVE_PORT, port);
}
private static PoolingOptions getReadPoolingOptions(Configuration conf)
{
Optional<Integer> coreConnections = getInputCoreConnections(conf);
Optional<Integer> maxConnections = getInputMaxConnections(conf);
Optional<Integer> maxSimultaneousRequests = getInputMaxSimultReqPerConnections(conf);
PoolingOptions poolingOptions = new PoolingOptions();
for (HostDistance hostDistance : Arrays.asList(HostDistance.LOCAL, HostDistance.REMOTE))
{
if (coreConnections.isPresent())
poolingOptions.setCoreConnectionsPerHost(hostDistance, coreConnections.get());
if (maxConnections.isPresent())
poolingOptions.setMaxConnectionsPerHost(hostDistance, maxConnections.get());
if (maxSimultaneousRequests.isPresent())
poolingOptions.setNewConnectionThreshold(hostDistance, maxSimultaneousRequests.get());
}
return poolingOptions;
}
private static QueryOptions getReadQueryOptions(Configuration conf)
{
String CL = ConfigHelper.getReadConsistencyLevel(conf);
Optional<Integer> fetchSize = getInputPageRowSize(conf);
QueryOptions queryOptions = new QueryOptions();
if (CL != null && !CL.isEmpty())
queryOptions.setConsistencyLevel(com.datastax.driver.core.ConsistencyLevel.valueOf(CL));
if (fetchSize.isPresent())
queryOptions.setFetchSize(fetchSize.get());
return queryOptions;
}
private static SocketOptions getReadSocketOptions(Configuration conf)
{
SocketOptions socketOptions = new SocketOptions();
Optional<Integer> connectTimeoutMillis = getInputNativeConnectionTimeout(conf);
Optional<Integer> readTimeoutMillis = getInputNativeReadConnectionTimeout(conf);
Optional<Integer> receiveBufferSize = getInputNativeReceiveBufferSize(conf);
Optional<Integer> sendBufferSize = getInputNativeSendBufferSize(conf);
Optional<Integer> soLinger = getInputNativeSolinger(conf);
Optional<Boolean> tcpNoDelay = getInputNativeTcpNodelay(conf);
Optional<Boolean> reuseAddress = getInputNativeReuseAddress(conf);
Optional<Boolean> keepAlive = getInputNativeKeepAlive(conf);
if (connectTimeoutMillis.isPresent())
socketOptions.setConnectTimeoutMillis(connectTimeoutMillis.get());
if (readTimeoutMillis.isPresent())
socketOptions.setReadTimeoutMillis(readTimeoutMillis.get());
if (receiveBufferSize.isPresent())
socketOptions.setReceiveBufferSize(receiveBufferSize.get());
if (sendBufferSize.isPresent())
socketOptions.setSendBufferSize(sendBufferSize.get());
if (soLinger.isPresent())
socketOptions.setSoLinger(soLinger.get());
if (tcpNoDelay.isPresent())
socketOptions.setTcpNoDelay(tcpNoDelay.get());
if (reuseAddress.isPresent())
socketOptions.setReuseAddress(reuseAddress.get());
if (keepAlive.isPresent())
socketOptions.setKeepAlive(keepAlive.get());
return socketOptions;
}
private static LoadBalancingPolicy getReadLoadBalancingPolicy(final String[] stickHosts)
{
return new LimitedLocalNodeFirstLocalBalancingPolicy(stickHosts);
}
private static Optional<AuthProvider> getDefaultAuthProvider(Configuration conf)
{
Optional<String> username = getStringSetting(USERNAME, conf);
Optional<String> password = getStringSetting(PASSWORD, conf);
if (username.isPresent() && password.isPresent())
{
return Optional.of(new PlainTextAuthProvider(username.get(), password.get()));
}
else
{
return Optional.absent();
}
}
private static Optional<AuthProvider> getAuthProvider(Configuration conf)
{
Optional<String> authProvider = getInputNativeAuthProvider(conf);
if (!authProvider.isPresent())
return getDefaultAuthProvider(conf);
return Optional.of(getClientAuthProvider(authProvider.get(), conf));
}
public static Optional<SSLOptions> getSSLOptions(Configuration conf)
{
Optional<String> truststorePath = getInputNativeSSLTruststorePath(conf);
if (truststorePath.isPresent())
{
Optional<String> keystorePath = getInputNativeSSLKeystorePath(conf);
Optional<String> truststorePassword = getInputNativeSSLTruststorePassword(conf);
Optional<String> keystorePassword = getInputNativeSSLKeystorePassword(conf);
Optional<String> cipherSuites = getInputNativeSSLCipherSuites(conf);
SSLContext context;
try
{
context = getSSLContext(truststorePath, truststorePassword, keystorePath, keystorePassword);
}
catch (UnrecoverableKeyException | KeyManagementException |
NoSuchAlgorithmException | KeyStoreException | CertificateException | IOException e)
{
throw new RuntimeException(e);
}
String[] css = null;
if (cipherSuites.isPresent())
css = cipherSuites.get().split(",");
return Optional.of(JdkSSLOptions.builder()
.withSSLContext(context)
.withCipherSuites(css)
.build());
}
return Optional.absent();
}
private static Optional<Integer> getIntSetting(String parameter, Configuration conf)
{
String setting = conf.get(parameter);
if (setting == null)
return Optional.absent();
return Optional.of(Integer.valueOf(setting));
}
private static Optional<Boolean> getBooleanSetting(String parameter, Configuration conf)
{
String setting = conf.get(parameter);
if (setting == null)
return Optional.absent();
return Optional.of(Boolean.valueOf(setting));
}
private static Optional<String> getStringSetting(String parameter, Configuration conf)
{
String setting = conf.get(parameter);
if (setting == null)
return Optional.absent();
return Optional.of(setting);
}
private static AuthProvider getClientAuthProvider(String factoryClassName, Configuration conf)
{
try
{
Class<?> c = Class.forName(factoryClassName);
if (PlainTextAuthProvider.class.equals(c))
{
String username = getStringSetting(USERNAME, conf).or("");
String password = getStringSetting(PASSWORD, conf).or("");
return (AuthProvider) c.getConstructor(String.class, String.class)
.newInstance(username, password);
}
else
{
return (AuthProvider) c.newInstance();
}
}
catch (Exception e)
{
throw new RuntimeException("Failed to instantiate auth provider:" + factoryClassName, e);
}
}
private static SSLContext getSSLContext(Optional<String> truststorePath,
Optional<String> truststorePassword,
Optional<String> keystorePath,
Optional<String> keystorePassword)
throws NoSuchAlgorithmException,
KeyStoreException,
CertificateException,
IOException,
UnrecoverableKeyException,
KeyManagementException
{
SSLContext ctx = SSLContext.getInstance("SSL");
TrustManagerFactory tmf = null;
if (truststorePath.isPresent())
{
try (InputStream tsf = Files.newInputStream(File.getPath(truststorePath.get())))
{
KeyStore ts = KeyStore.getInstance("JKS");
ts.load(tsf, truststorePassword.isPresent() ? truststorePassword.get().toCharArray() : null);
tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(ts);
}
}
KeyManagerFactory kmf = null;
if (keystorePath.isPresent())
{
try (InputStream ksf = Files.newInputStream(File.getPath(keystorePath.get())))
{
KeyStore ks = KeyStore.getInstance("JKS");
ks.load(ksf, keystorePassword.isPresent() ? keystorePassword.get().toCharArray() : null);
kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(ks, keystorePassword.isPresent() ? keystorePassword.get().toCharArray() : null);
}
}
ctx.init(kmf != null ? kmf.getKeyManagers() : null,
tmf != null ? tmf.getTrustManagers() : null,
new SecureRandom());
return ctx;
}
}

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

@ -1,502 +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.cql3;
import java.io.IOException;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.*;
import com.datastax.driver.core.Cluster;
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.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.TokenRange;
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import org.apache.cassandra.schema.SchemaConstants;
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;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.hadoop.*;
import org.apache.cassandra.utils.*;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily.
*
* At minimum, you need to set the KS and CF in your Hadoop job Configuration.
* The ConfigHelper class is provided to make this
* simple:
* ConfigHelper.setInputColumnFamily
*
* You can also configure the number of rows per InputSplit with
* 1: ConfigHelper.setInputSplitSize. The default split size is 64k rows.
* or
* 2: ConfigHelper.setInputSplitSizeInMb. InputSplit size in MB with new, more precise method
* If no value is provided for InputSplitSizeInMb, we default to using InputSplitSize.
*
* CQLConfigHelper.setInputCQLPageRowSize. The default page row size is 1000. You
* should set it to "as big as possible, but no bigger." It set the LIMIT for the CQL
* query, so you need set it big enough to minimize the network overhead, and also
* not too big to avoid out of memory issue.
*
* other native protocol connection parameters in CqlConfigHelper
*/
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;
public RecordReader<Long, Row> getRecordReader(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);
CqlRecordReader recordReader = new CqlRecordReader();
recordReader.initialize((org.apache.hadoop.mapreduce.InputSplit)split, tac);
return recordReader;
}
@Override
public org.apache.hadoop.mapreduce.RecordReader<Long, Row> createRecordReader(
org.apache.hadoop.mapreduce.InputSplit arg0, TaskAttemptContext arg1) throws IOException,
InterruptedException
{
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.trace("partitioner is {}", partitioner);
// canonical ranges, split into pieces, fetching the splits in parallel
ExecutorService executor = executorFactory().pooled("HadoopInput", 128);
List<org.apache.hadoop.mapreduce.InputSplit> splits = new ArrayList<>();
String[] inputInitialAddress = ConfigHelper.getInputInitialAddress(conf).split(",");
try (Cluster cluster = CqlConfigHelper.getInputCluster(inputInitialAddress, conf);
Session session = cluster.connect())
{
List<SplitFuture> splitfutures = new ArrayList<>();
//TODO if the job range is defined and does perfectly match tokens, then the logic will be unable to get estimates since they are pre-computed
// tokens: [0, 10, 20]
// job range: [0, 10) - able to get estimate
// job range: [5, 15) - unable to get estimate
Pair<String, String> jobKeyRange = ConfigHelper.getInputKeyRange(conf);
Range<Token> jobRange = null;
if (jobKeyRange != null)
{
jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.left),
partitioner.getTokenFactory().fromString(jobKeyRange.right));
}
Metadata metadata = cluster.getMetadata();
// canonical ranges and nodes holding replicas
Map<TokenRange, List<Host>> masterRangeNodes = getRangeMap(keyspace, metadata, getTargetDC(metadata, inputInitialAddress));
for (TokenRange range : masterRangeNodes.keySet())
{
if (jobRange == null)
{
for (TokenRange unwrapped : range.unwrap())
{
// for each tokenRange, pick a live owner and ask it for the byte-sized splits
SplitFuture task = new SplitFuture(new SplitCallable(unwrapped, masterRangeNodes.get(range), conf, session));
executor.submit(task);
splitfutures.add(task);
}
}
else
{
TokenRange jobTokenRange = rangeToTokenRange(metadata, jobRange);
if (range.intersects(jobTokenRange))
{
for (TokenRange intersection: range.intersectWith(jobTokenRange))
{
for (TokenRange unwrapped : intersection.unwrap())
{
// for each tokenRange, pick a live owner and ask it for the byte-sized splits
SplitFuture task = new SplitFuture(new SplitCallable(unwrapped, masterRangeNodes.get(range), conf, session));
executor.submit(task);
splitfutures.add(task);
}
}
}
}
}
// wait until we have all the results back
List<SplitFuture> failedTasks = new ArrayList<>();
int maxSplits = 0;
long expectedPartionsForFailedRanges = 0;
for (SplitFuture task : splitfutures)
{
try
{
List<ColumnFamilySplit> tokenRangeSplits = task.get();
if (tokenRangeSplits.size() > maxSplits)
{
maxSplits = tokenRangeSplits.size();
expectedPartionsForFailedRanges = tokenRangeSplits.get(0).getLength();
}
splits.addAll(tokenRangeSplits);
}
catch (Exception e)
{
failedTasks.add(task);
}
}
// The estimate is only stored on a single host, if that host is down then can not get the estimate
// its more than likely that a single host could be "too large" for one split but there is no way of
// knowning!
// This logic attempts to guess the estimate from all the successful ranges
if (!failedTasks.isEmpty())
{
// if every split failed this will be 0
if (maxSplits == 0)
throwAllSplitsFailed(failedTasks);
for (SplitFuture task : failedTasks)
{
try
{
// the task failed, so this should throw
task.get();
}
catch (Exception cause)
{
logger.warn("Unable to get estimate for {}, the host {} had a exception; falling back to default estimate", task.splitCallable.tokenRange, task.splitCallable.hosts.get(0), cause);
}
}
for (SplitFuture task : failedTasks)
splits.addAll(toSplit(task.splitCallable.hosts, splitTokenRange(task.splitCallable.tokenRange, maxSplits, expectedPartionsForFailedRanges)));
}
}
finally
{
executor.shutdownNow();
}
assert splits.size() > 0;
Collections.shuffle(splits, new Random(nanoTime()));
return splits;
}
private static IllegalStateException throwAllSplitsFailed(List<SplitFuture> failedTasks)
{
IllegalStateException exception = new IllegalStateException("No successful tasks found");
for (SplitFuture task : failedTasks)
{
try
{
// the task failed, so this should throw
task.get();
}
catch (Exception cause)
{
exception.addSuppressed(cause);
}
}
throw exception;
}
private static String getTargetDC(Metadata metadata, String[] inputInitialAddress)
{
BiMultiValMap<InetAddress, String> addressToDc = new BiMultiValMap<>();
Multimap<String, InetAddress> dcToAddresses = addressToDc.inverse();
// only way to match is off the broadcast addresses, so for all hosts do a existence check
Set<InetAddress> addresses = new HashSet<>(inputInitialAddress.length);
for (String inputAddress : inputInitialAddress)
addresses.addAll(parseAddress(inputAddress));
for (Host host : metadata.getAllHosts())
{
InetAddress address = host.getBroadcastAddress();
if (addresses.contains(address))
addressToDc.put(address, host.getDatacenter());
}
switch (dcToAddresses.keySet().size())
{
case 1:
return Iterables.getOnlyElement(dcToAddresses.keySet());
case 0:
throw new IllegalStateException("Input addresses could not be used to find DC; non match client metadata");
default:
// Mutliple DCs found, attempt to pick the first based off address list. This is to mimic the 2.1
// behavior which would connect in order and the first node successfully able to connect to was the
// local DC to use; since client abstracts this, we rely on existence as a proxy for connect.
for (String inputAddress : inputInitialAddress)
{
for (InetAddress add : parseAddress(inputAddress))
{
String dc = addressToDc.get(add);
// possible the address isn't in the cluster and the client dropped, so ignore null
if (dc != null)
return dc;
}
}
// some how we were able to connect to the cluster, find multiple DCs using matching, and yet couldn't
// match again...
throw new AssertionError("Unable to infer datacenter from initial addresses; multiple datacenters found "
+ dcToAddresses.keySet() + ", should only use addresses from one datacenter");
}
}
private static List<InetAddress> parseAddress(String str)
{
try
{
return Arrays.asList(InetAddress.getAllByName(str));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
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, Host host, Configuration conf, Session session)
{
int splitSize = ConfigHelper.getInputSplitSize(conf);
int splitSizeMiB = ConfigHelper.getInputSplitSizeInMb(conf);
return describeSplits(keyspace, cfName, range, host, splitSize, splitSizeMiB, session);
}
private static Map<TokenRange, List<Host>> getRangeMap(String keyspace, Metadata metadata, String targetDC)
{
return CqlClientHelper.getLocalPrimaryRangeForDC(keyspace, metadata, targetDC);
}
private Map<TokenRange, Long> describeSplits(String keyspace, String table, TokenRange tokenRange, Host host, int splitSize, int splitSizeMb, Session session)
{
// In 2.1 the host list was walked in-order (only move to next if IOException) and calls
// org.apache.cassandra.service.StorageService.getSplits(java.lang.String, java.lang.String, org.apache.cassandra.dht.Range<org.apache.cassandra.dht.Token>, int)
// that call computes totalRowCountEstimate (used to compute #splits) then splits the ring based off those estimates
//
// The main difference is that the estimates in 2.1 were computed based off the data, so replicas could answer the estimates
// In 3.0 we rely on the below CQL query which is local and only computes estimates for the primary range; this
// puts us in a sticky spot to answer, if the node fails what do we do? 3.0 behavior only matches 2.1 IFF all
// nodes are up and healthy
ResultSet resultSet = queryTableEstimates(session, host, keyspace, table, tokenRange);
Row row = resultSet.one();
long meanPartitionSize = 0;
long partitionCount = 0;
int splitCount = 0;
if (row != null)
{
meanPartitionSize = row.getLong("mean_partition_size");
partitionCount = row.getLong("partitions_count");
splitCount = splitSizeMb > 0
? (int)(meanPartitionSize * partitionCount / splitSizeMb / 1024 / 1024)
: (int)(partitionCount / splitSize);
}
// If we have no data on this split or the size estimate is 0,
// return the full split i.e., do not sub-split
// Assume smallest granularity of partition count available from CASSANDRA-7688
if (splitCount == 0)
{
Map<TokenRange, Long> wrappedTokenRange = new HashMap<>();
wrappedTokenRange.put(tokenRange, partitionCount == 0 ? 128L : partitionCount);
return wrappedTokenRange;
}
return splitTokenRange(tokenRange, splitCount, partitionCount / splitCount);
}
private static ResultSet queryTableEstimates(Session session, Host host, String keyspace, String table, TokenRange tokenRange)
{
try
{
String query = String.format("SELECT mean_partition_size, partitions_count " +
"FROM %s.%s " +
"WHERE keyspace_name = ? AND table_name = ? AND range_type = '%s' AND range_start = ? AND range_end = ?",
SchemaConstants.SYSTEM_KEYSPACE_NAME,
SystemKeyspace.TABLE_ESTIMATES,
SystemKeyspace.TABLE_ESTIMATES_TYPE_LOCAL_PRIMARY);
Statement stmt = new SimpleStatement(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString()).setHost(host);
return session.execute(stmt);
}
catch (InvalidQueryException e)
{
// if the table doesn't exist, fall back to old table. This is likely to return no records in a multi
// DC setup, but should work fine in a single DC setup.
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 = ?",
SchemaConstants.SYSTEM_KEYSPACE_NAME,
SystemKeyspace.LEGACY_SIZE_ESTIMATES);
Statement stmt = new SimpleStatement(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString()).setHost(host);
return session.execute(stmt);
}
}
private static Map<TokenRange, Long> splitTokenRange(TokenRange tokenRange, int splitCount, long partitionCount)
{
List<TokenRange> splitRanges = tokenRange.splitEvenly(splitCount);
Map<TokenRange, Long> rangesWithLength = Maps.newHashMapWithExpectedSize(splitRanges.size());
for (TokenRange range : splitRanges)
rangesWithLength.put(range, partitionCount);
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<ColumnFamilySplit>>
{
private final TokenRange tokenRange;
private final List<Host> hosts;
private final Configuration conf;
private final Session session;
public SplitCallable(TokenRange tokenRange, List<Host> hosts, Configuration conf, Session session)
{
Preconditions.checkArgument(!hosts.isEmpty(), "hosts list requires at least 1 host but was empty");
this.tokenRange = tokenRange;
this.hosts = hosts;
this.conf = conf;
this.session = session;
}
public List<ColumnFamilySplit> call() throws Exception
{
Map<TokenRange, Long> subSplits = getSubSplits(keyspace, cfName, tokenRange, hosts.get(0), conf, session);
return toSplit(hosts, subSplits);
}
}
private static class SplitFuture extends FutureTask<List<ColumnFamilySplit>>
{
private final SplitCallable splitCallable;
SplitFuture(SplitCallable splitCallable)
{
super(splitCallable);
this.splitCallable = splitCallable;
}
}
private List<ColumnFamilySplit> toSplit(List<Host> hosts, Map<TokenRange, Long> subSplits)
{
// 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;
ArrayList<ColumnFamilySplit> splits = new ArrayList<>();
for (Map.Entry<TokenRange, Long> subSplitEntry : subSplits.entrySet())
{
TokenRange subrange = subSplitEntry.getKey();
ColumnFamilySplit split =
new ColumnFamilySplit(
partitionerIsOpp ?
subrange.getStart().toString().substring(2) : subrange.getStart().toString(),
partitionerIsOpp ?
subrange.getEnd().toString().substring(2) : subrange.getEnd().toString(),
subSplitEntry.getValue(),
endpoints);
logger.trace("adding {}", split);
splits.add(split);
}
return splits;
}
}

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

@ -1,143 +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.cql3;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.hadoop.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.mapreduce.*;
/**
* The <code>CqlOutputFormat</code> acts as a Hadoop-specific
* OutputFormat that allows reduce tasks to store keys (and corresponding
* bound variable values) as CQL rows (and respective columns) in a given
* table.
*
* <p>
* As is the case with the {@link org.apache.cassandra.hadoop.ColumnFamilyInputFormat},
* you need to set the prepared statement in your
* Hadoop job Configuration. The {@link CqlConfigHelper} class, through its
* {@link CqlConfigHelper#setOutputCql} method, is provided to make this
* simple.
* you need to set the Keyspace. The {@link ConfigHelper} class, through its
* {@link ConfigHelper#setOutputColumnFamily} method, is provided to make this
* simple.
* </p>
*
* <p>
* For the sake of performance, this class employs a lazy write-back caching
* mechanism, where its record writer prepared statement binded variable values
* created based on the reduce's inputs (in a task-specific map), and periodically
* makes the changes official by sending a execution of prepared statement request
* to Cassandra.
* </p>
*/
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.
*
* @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);
}
/**
* 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();
}
/** Fills the deprecated OutputFormat interface for streaming. */
@Deprecated
public CqlRecordWriter getRecordWriter(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job, String name, org.apache.hadoop.util.Progressable progress) throws IOException
{
return new CqlRecordWriter(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.
* @throws IOException
*/
public CqlRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException
{
return new CqlRecordWriter(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) { }
}
}

784
src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java
View File

@ -1,784 +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.cql3;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.datastax.driver.core.TypeCodec;
import org.apache.cassandra.utils.AbstractIterator;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ColumnMetadata;
import com.datastax.driver.core.LocalDate;
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.TableMetadata;
import com.datastax.driver.core.Token;
import com.datastax.driver.core.TupleValue;
import com.datastax.driver.core.UDTValue;
import com.google.common.reflect.TypeToken;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.hadoop.ColumnFamilySplit;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.utils.ByteBufferUtil;
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;
/**
* <p>
* CqlRecordReader reads the rows return from the CQL query
* It uses CQL auto-paging.
* </p>
* <p>
* Return a Long as a local CQL row key starts from 0;
* </p>
* {@code
* Row as C* java driver CQL result set row
* 1) select clause must include partition key columns (to calculate the progress based on the actual CF row processed)
* 2) where clause must include token(partition_key1, ... , partition_keyn) > ? and
* token(partition_key1, ... , partition_keyn) <= ? (in the right order)
* }
*/
public class CqlRecordReader extends RecordReader<Long, Row>
implements org.apache.hadoop.mapred.RecordReader<Long, Row>, AutoCloseable
{
private static final Logger logger = LoggerFactory.getLogger(CqlRecordReader.class);
private ColumnFamilySplit split;
private RowIterator rowIterator;
private Pair<Long, Row> currentRow;
private int totalRowCount; // total number of rows to fetch
private String keyspace;
private String cfName;
private String cqlQuery;
private Cluster cluster;
private Session session;
private IPartitioner partitioner;
private String inputColumns;
private String userDefinedWhereClauses;
private List<String> partitionKeys = new ArrayList<>();
// partition keys -- key aliases
private LinkedHashMap<String, Boolean> partitionBoundColumns = Maps.newLinkedHashMap();
protected int nativeProtocolVersion = 1;
public CqlRecordReader()
{
super();
}
@Override
public void initialize(InputSplit split, TaskAttemptContext context) throws IOException
{
this.split = (ColumnFamilySplit) split;
Configuration conf = HadoopCompat.getConfiguration(context);
totalRowCount = (this.split.getLength() < Long.MAX_VALUE)
? (int) this.split.getLength()
: ConfigHelper.getInputSplitSize(conf);
cfName = ConfigHelper.getInputColumnFamily(conf);
keyspace = ConfigHelper.getInputKeyspace(conf);
partitioner = ConfigHelper.getInputPartitioner(conf);
inputColumns = CqlConfigHelper.getInputcolumns(conf);
userDefinedWhereClauses = CqlConfigHelper.getInputWhereClauses(conf);
try
{
if (cluster != null)
return;
// create a Cluster instance
String[] locations = split.getLocations();
cluster = CqlConfigHelper.getInputCluster(locations, conf);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
if (cluster != null)
session = cluster.connect(quote(keyspace));
if (session == null)
throw new RuntimeException("Can't create connection session");
//get negotiated serialization protocol
nativeProtocolVersion = cluster.getConfiguration().getProtocolOptions().getProtocolVersion().toInt();
// If the user provides a CQL query then we will use it without validation
// otherwise we will fall back to building a query using the:
// inputColumns
// whereClauses
cqlQuery = CqlConfigHelper.getInputCql(conf);
// validate that the user hasn't tried to give us a custom query along with input columns
// and where clauses
if (StringUtils.isNotEmpty(cqlQuery) && (StringUtils.isNotEmpty(inputColumns) ||
StringUtils.isNotEmpty(userDefinedWhereClauses)))
{
throw new AssertionError("Cannot define a custom query with input columns and / or where clauses");
}
if (StringUtils.isEmpty(cqlQuery))
cqlQuery = buildQuery();
logger.trace("cqlQuery {}", cqlQuery);
rowIterator = new RowIterator();
logger.trace("created {}", rowIterator);
}
public void close()
{
if (session != null)
session.close();
if (cluster != null)
cluster.close();
}
public Long getCurrentKey()
{
return currentRow.left;
}
public Row getCurrentValue()
{
return currentRow.right;
}
public float getProgress()
{
if (!rowIterator.hasNext())
return 1.0F;
// the progress is likely to be reported slightly off the actual but close enough
float progress = ((float) rowIterator.totalRead / totalRowCount);
return progress > 1.0F ? 1.0F : progress;
}
public boolean nextKeyValue() throws IOException
{
if (!rowIterator.hasNext())
{
logger.trace("Finished scanning {} rows (estimate was: {})", rowIterator.totalRead, totalRowCount);
return false;
}
try
{
currentRow = rowIterator.next();
}
catch (Exception e)
{
// throw it as IOException, so client can catch it and handle it at client side
IOException ioe = new IOException(e.getMessage());
ioe.initCause(ioe.getCause());
throw ioe;
}
return true;
}
// 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(Long key, Row value) throws IOException
{
if (nextKeyValue())
{
((WrappedRow)value).setRow(getCurrentValue());
return true;
}
return false;
}
public long getPos() throws IOException
{
return rowIterator.totalRead;
}
public Long createKey()
{
return Long.valueOf(0L);
}
public Row createValue()
{
return new WrappedRow();
}
/**
* Return native version protocol of the cluster connection
* @return serialization protocol version.
*/
public int getNativeProtocolVersion()
{
return nativeProtocolVersion;
}
/** CQL row iterator
* Input cql query
* 1) select clause must include key columns (if we use partition key based row count)
* 2) where clause must include token(partition_key1 ... partition_keyn) > ? and
* token(partition_key1 ... partition_keyn) <= ?
*/
private class RowIterator extends AbstractIterator<Pair<Long, Row>>
{
private long keyId = 0L;
protected int totalRead = 0; // total number of cf rows read
protected Iterator<Row> rows;
private Map<String, ByteBuffer> previousRowKey = new HashMap<String, ByteBuffer>(); // previous CF row key
public RowIterator()
{
AbstractType type = partitioner.getTokenValidator();
ResultSet rs = session.execute(cqlQuery, type.compose(type.fromString(split.getStartToken())), type.compose(type.fromString(split.getEndToken())) );
for (ColumnMetadata meta : cluster.getMetadata().getKeyspace(quote(keyspace)).getTable(quote(cfName)).getPartitionKey())
partitionBoundColumns.put(meta.getName(), Boolean.TRUE);
rows = rs.iterator();
}
protected Pair<Long, Row> computeNext()
{
if (rows == null || !rows.hasNext())
return endOfData();
Row row = rows.next();
Map<String, ByteBuffer> keyColumns = new HashMap<String, ByteBuffer>(partitionBoundColumns.size());
for (String column : partitionBoundColumns.keySet())
keyColumns.put(column, row.getBytesUnsafe(column));
// increase total CF row read
if (previousRowKey.isEmpty() && !keyColumns.isEmpty())
{
previousRowKey = keyColumns;
totalRead++;
}
else
{
for (String column : partitionBoundColumns.keySet())
{
// this is not correct - but we don't seem to have easy access to better type information here
if (ByteBufferUtil.compareUnsigned(keyColumns.get(column), previousRowKey.get(column)) != 0)
{
previousRowKey = keyColumns;
totalRead++;
break;
}
}
}
keyId ++;
return Pair.create(keyId, row);
}
}
private static class WrappedRow implements Row
{
private Row row;
public void setRow(Row row)
{
this.row = row;
}
@Override
public ColumnDefinitions getColumnDefinitions()
{
return row.getColumnDefinitions();
}
@Override
public boolean isNull(int i)
{
return row.isNull(i);
}
@Override
public boolean isNull(String name)
{
return row.isNull(name);
}
@Override
public Object getObject(int i)
{
return row.getObject(i);
}
@Override
public <T> T get(int i, Class<T> aClass)
{
return row.get(i, aClass);
}
@Override
public <T> T get(int i, TypeToken<T> typeToken)
{
return row.get(i, typeToken);
}
@Override
public <T> T get(int i, TypeCodec<T> typeCodec)
{
return row.get(i, typeCodec);
}
@Override
public Object getObject(String s)
{
return row.getObject(s);
}
@Override
public <T> T get(String s, Class<T> aClass)
{
return row.get(s, aClass);
}
@Override
public <T> T get(String s, TypeToken<T> typeToken)
{
return row.get(s, typeToken);
}
@Override
public <T> T get(String s, TypeCodec<T> typeCodec)
{
return row.get(s, typeCodec);
}
@Override
public boolean getBool(int i)
{
return row.getBool(i);
}
@Override
public boolean getBool(String name)
{
return row.getBool(name);
}
@Override
public short getShort(int i)
{
return row.getShort(i);
}
@Override
public short getShort(String s)
{
return row.getShort(s);
}
@Override
public byte getByte(int i)
{
return row.getByte(i);
}
@Override
public byte getByte(String s)
{
return row.getByte(s);
}
@Override
public int getInt(int i)
{
return row.getInt(i);
}
@Override
public int getInt(String name)
{
return row.getInt(name);
}
@Override
public long getLong(int i)
{
return row.getLong(i);
}
@Override
public long getLong(String name)
{
return row.getLong(name);
}
@Override
public Date getTimestamp(int i)
{
return row.getTimestamp(i);
}
@Override
public Date getTimestamp(String s)
{
return row.getTimestamp(s);
}
@Override
public LocalDate getDate(int i)
{
return row.getDate(i);
}
@Override
public LocalDate getDate(String s)
{
return row.getDate(s);
}
@Override
public long getTime(int i)
{
return row.getTime(i);
}
@Override
public long getTime(String s)
{
return row.getTime(s);
}
@Override
public float getFloat(int i)
{
return row.getFloat(i);
}
@Override
public float getFloat(String name)
{
return row.getFloat(name);
}
@Override
public double getDouble(int i)
{
return row.getDouble(i);
}
@Override
public double getDouble(String name)
{
return row.getDouble(name);
}
@Override
public ByteBuffer getBytesUnsafe(int i)
{
return row.getBytesUnsafe(i);
}
@Override
public ByteBuffer getBytesUnsafe(String name)
{
return row.getBytesUnsafe(name);
}
@Override
public ByteBuffer getBytes(int i)
{
return row.getBytes(i);
}
@Override
public ByteBuffer getBytes(String name)
{
return row.getBytes(name);
}
@Override
public String getString(int i)
{
return row.getString(i);
}
@Override
public String getString(String name)
{
return row.getString(name);
}
@Override
public BigInteger getVarint(int i)
{
return row.getVarint(i);
}
@Override
public BigInteger getVarint(String name)
{
return row.getVarint(name);
}
@Override
public BigDecimal getDecimal(int i)
{
return row.getDecimal(i);
}
@Override
public BigDecimal getDecimal(String name)
{
return row.getDecimal(name);
}
@Override
public UUID getUUID(int i)
{
return row.getUUID(i);
}
@Override
public UUID getUUID(String name)
{
return row.getUUID(name);
}
@Override
public InetAddress getInet(int i)
{
return row.getInet(i);
}
@Override
public InetAddress getInet(String name)
{
return row.getInet(name);
}
@Override
public <T> List<T> getList(int i, Class<T> elementsClass)
{
return row.getList(i, elementsClass);
}
@Override
public <T> List<T> getList(int i, TypeToken<T> typeToken)
{
return row.getList(i, typeToken);
}
@Override
public <T> List<T> getList(String name, Class<T> elementsClass)
{
return row.getList(name, elementsClass);
}
@Override
public <T> List<T> getList(String s, TypeToken<T> typeToken)
{
return row.getList(s, typeToken);
}
@Override
public <T> Set<T> getSet(int i, Class<T> elementsClass)
{
return row.getSet(i, elementsClass);
}
@Override
public <T> Set<T> getSet(int i, TypeToken<T> typeToken)
{
return row.getSet(i, typeToken);
}
@Override
public <T> Set<T> getSet(String name, Class<T> elementsClass)
{
return row.getSet(name, elementsClass);
}
@Override
public <T> Set<T> getSet(String s, TypeToken<T> typeToken)
{
return row.getSet(s, typeToken);
}
@Override
public <K, V> Map<K, V> getMap(int i, Class<K> keysClass, Class<V> valuesClass)
{
return row.getMap(i, keysClass, valuesClass);
}
@Override
public <K, V> Map<K, V> getMap(int i, TypeToken<K> typeToken, TypeToken<V> typeToken1)
{
return row.getMap(i, typeToken, typeToken1);
}
@Override
public <K, V> Map<K, V> getMap(String name, Class<K> keysClass, Class<V> valuesClass)
{
return row.getMap(name, keysClass, valuesClass);
}
@Override
public <K, V> Map<K, V> getMap(String s, TypeToken<K> typeToken, TypeToken<V> typeToken1)
{
return row.getMap(s, typeToken, typeToken1);
}
@Override
public UDTValue getUDTValue(int i)
{
return row.getUDTValue(i);
}
@Override
public UDTValue getUDTValue(String name)
{
return row.getUDTValue(name);
}
@Override
public TupleValue getTupleValue(int i)
{
return row.getTupleValue(i);
}
@Override
public TupleValue getTupleValue(String name)
{
return row.getTupleValue(name);
}
@Override
public Token getToken(int i)
{
return row.getToken(i);
}
@Override
public Token getToken(String name)
{
return row.getToken(name);
}
@Override
public Token getPartitionKeyToken()
{
return row.getPartitionKeyToken();
}
}
/**
* Build a query for the reader of the form:
*
* SELECT * FROM ks>cf token(pk1,...pkn)>? AND token(pk1,...pkn)<=? [AND user where clauses] [ALLOW FILTERING]
*/
private String buildQuery()
{
fetchKeys();
List<String> columns = getSelectColumns();
String selectColumnList = columns.size() == 0 ? "*" : makeColumnList(columns);
String partitionKeyList = makeColumnList(partitionKeys);
return String.format("SELECT %s FROM %s.%s WHERE token(%s)>? AND token(%s)<=?" + getAdditionalWhereClauses(),
selectColumnList, quote(keyspace), quote(cfName), partitionKeyList, partitionKeyList);
}
private String getAdditionalWhereClauses()
{
String whereClause = "";
if (StringUtils.isNotEmpty(userDefinedWhereClauses))
whereClause += " AND " + userDefinedWhereClauses;
if (StringUtils.isNotEmpty(userDefinedWhereClauses))
whereClause += " ALLOW FILTERING";
return whereClause;
}
private List<String> getSelectColumns()
{
List<String> selectColumns = new ArrayList<>();
if (StringUtils.isNotEmpty(inputColumns))
{
// We must select all the partition keys plus any other columns the user wants
selectColumns.addAll(partitionKeys);
for (String column : Splitter.on(',').split(inputColumns))
{
if (!partitionKeys.contains(column))
selectColumns.add(column);
}
}
return selectColumns;
}
private String makeColumnList(Collection<String> columns)
{
return Joiner.on(',').join(Iterables.transform(columns, new Function<String, String>()
{
public String apply(String column)
{
return quote(column);
}
}));
}
private void fetchKeys()
{
// get CF meta data
TableMetadata tableMetadata = session.getCluster()
.getMetadata()
.getKeyspace(Metadata.quote(keyspace))
.getTable(Metadata.quote(cfName));
if (tableMetadata == null)
{
throw new RuntimeException("No table metadata found for " + keyspace + "." + cfName);
}
//Here we assume that tableMetadata.getPartitionKey() always
//returns the list of columns in order of component_index
for (ColumnMetadata partitionKey : tableMetadata.getPartitionKey())
{
partitionKeys.add(partitionKey.getName());
}
}
private String quote(String identifier)
{
return "\"" + identifier.replaceAll("\"", "\"\"") + "\"";
}
}

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

@ -1,536 +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.cql3;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.*;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.util.Progressable;
import static java.util.stream.Collectors.toMap;
/**
* The <code>CqlRecordWriter</code> maps the output &lt;key, value&gt;
* pairs to a Cassandra table. In particular, it applies the binded variables
* in the value to the prepared statement, which it associates with the key, and in
* turn the responsible endpoint.
*
* <p>
* Furthermore, this writer groups the cql queries by the endpoint responsible for
* the rows being affected. This allows the cql queries to be executed in parallel,
* directly to a responsible endpoint.
* </p>
*
* @see CqlOutputFormat
*/
class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuffer>> implements
org.apache.hadoop.mapred.RecordWriter<Map<String, ByteBuffer>, List<ByteBuffer>>, AutoCloseable
{
private static final Logger logger = LoggerFactory.getLogger(CqlRecordWriter.class);
// 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 Progressable progressable;
protected TaskAttemptContext context;
// 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 NativeRingCache ringCache;
// handles for clients for each range running in the threadpool
protected final Map<InetAddress, RangeClient> clients;
// host to prepared statement id mappings
protected final ConcurrentHashMap<Session, PreparedStatement> preparedStatements = new ConcurrentHashMap<Session, PreparedStatement>();
protected final String cql;
protected List<ColumnMetadata> partitionKeyColumns;
protected List<ColumnMetadata> clusterColumns;
/**
* 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
*/
CqlRecordWriter(TaskAttemptContext context) throws IOException
{
this(HadoopCompat.getConfiguration(context));
this.context = context;
}
CqlRecordWriter(Configuration conf, Progressable progressable)
{
this(conf);
this.progressable = progressable;
}
CqlRecordWriter(Configuration conf)
{
this.conf = conf;
this.queueSize = conf.getInt(CqlOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
batchThreshold = conf.getLong(CqlOutputFormat.BATCH_THRESHOLD, 32);
this.clients = new HashMap<>();
String keyspace = ConfigHelper.getOutputKeyspace(conf);
try (Cluster cluster = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf))
{
Metadata metadata = cluster.getMetadata();
ringCache = new NativeRingCache(conf, metadata);
TableMetadata tableMetadata = metadata.getKeyspace(Metadata.quote(keyspace)).getTable(ConfigHelper.getOutputColumnFamily(conf));
clusterColumns = tableMetadata.getClusteringColumns();
partitionKeyColumns = tableMetadata.getPartitionKey();
String cqlQuery = CqlConfigHelper.getOutputCql(conf).trim();
if (cqlQuery.toLowerCase(Locale.ENGLISH).startsWith("insert"))
throw new UnsupportedOperationException("INSERT with CqlRecordWriter is not supported, please use UPDATE/DELETE statement");
cql = appendKeyWhereClauses(cqlQuery);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
/**
* 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, InterruptedException
{
close();
}
/** Fills the deprecated RecordWriter interface for streaming. */
@Deprecated
public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException
{
close();
}
@Override
public void close() throws IOException
{
// close all the clients before throwing anything
IOException clientException = null;
for (RangeClient client : clients.values())
{
try
{
client.close();
}
catch (IOException e)
{
clientException = e;
}
}
if (clientException != null)
throw clientException;
}
/**
* If the key is to be associated with a valid value, a mutation is created
* for it with the given table 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 keyColumns
* the key to write.
* @param values
* the values to write.
* @throws IOException
*/
@Override
public void write(Map<String, ByteBuffer> keyColumns, List<ByteBuffer> values) throws IOException
{
TokenRange range = ringCache.getRange(getPartitionKey(keyColumns));
// get the client for the given range, or create a new one
final InetAddress address = ringCache.getEndpoints(range).get(0);
RangeClient client = clients.get(address);
if (client == null)
{
// haven't seen keys for this range: create new client
client = new RangeClient(ringCache.getEndpoints(range));
client.start();
clients.put(address, client);
}
// add primary key columns to the bind variables
List<ByteBuffer> allValues = new ArrayList<ByteBuffer>(values);
for (ColumnMetadata column : partitionKeyColumns)
allValues.add(keyColumns.get(column.getName()));
for (ColumnMetadata column : clusterColumns)
allValues.add(keyColumns.get(column.getName()));
client.put(allValues);
if (progressable != null)
progressable.progress();
if (context != null)
HadoopCompat.progress(context);
}
private static void closeSession(Session session)
{
//Close the session to satisfy to avoid warnings for the resource not being closed
try
{
if (session != null)
session.getCluster().closeAsync();
}
catch (Throwable t)
{
logger.warn("Error closing connection", t);
}
}
/**
* A client that runs in a threadpool and connects to the list of endpoints for a particular
* range. Bound variables 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;
protected Cluster cluster = null;
// A bounded queue of incoming mutations for this range
protected final BlockingQueue<List<ByteBuffer>> queue = new ArrayBlockingQueue<List<ByteBuffer>>(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;
/**
* 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(List<ByteBuffer> 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);
}
}
}
/**
* Loops collecting cql binded variable values from the queue and sending to Cassandra
*/
@SuppressWarnings("resource")
public void run()
{
Session session = null;
try
{
outer:
while (run || !queue.isEmpty())
{
List<ByteBuffer> bindVariables;
try
{
bindVariables = queue.take();
}
catch (InterruptedException e)
{
// re-check loop condition after interrupt
continue;
}
ListIterator<InetAddress> iter = endpoints.listIterator();
while (true)
{
// send the mutation to the last-used endpoint. first time through, this will NPE harmlessly.
if (session != null)
{
try
{
int i = 0;
PreparedStatement statement = preparedStatement(session);
while (bindVariables != null)
{
BoundStatement boundStatement = new BoundStatement(statement);
for (int columnPosition = 0; columnPosition < bindVariables.size(); columnPosition++)
{
boundStatement.setBytesUnsafe(columnPosition, bindVariables.get(columnPosition));
}
session.execute(boundStatement);
i++;
if (i >= batchThreshold)
break;
bindVariables = queue.poll();
}
break;
}
catch (Exception e)
{
closeInternal();
if (!iter.hasNext())
{
lastException = new IOException(e);
break outer;
}
}
}
// attempt to connect to a different endpoint
try
{
InetAddress address = iter.next();
String host = address.getHostName();
cluster = CqlConfigHelper.getOutputCluster(host, conf);
closeSession(session);
session = cluster.connect();
}
catch (Exception e)
{
//If connection died due to Interrupt, just try connecting to the endpoint again.
//There are too many ways for the Thread.interrupted() state to be cleared, so
//we can't rely on that here. Until the java driver gives us a better way of knowing
//that this exception came from an InterruptedException, this is the best solution.
if (canRetryDriverConnection(e))
{
iter.previous();
}
closeInternal();
// Most exceptions mean 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 AuthenticationException || e instanceof InvalidQueryException) || !iter.hasNext())
{
lastException = new IOException(e);
break outer;
}
}
}
}
}
finally
{
closeSession(session);
// close all our connections once we are done.
closeInternal();
}
}
/** get prepared statement id from cache, otherwise prepare it from Cassandra server*/
private PreparedStatement preparedStatement(Session client)
{
PreparedStatement statement = preparedStatements.get(client);
if (statement == null)
{
PreparedStatement result;
try
{
result = client.prepare(cql);
}
catch (NoHostAvailableException e)
{
throw new RuntimeException("failed to prepare cql query " + cql, e);
}
PreparedStatement previousId = preparedStatements.putIfAbsent(client, result);
statement = previousId == null ? result : previousId;
}
return statement;
}
public void close() throws IOException
{
// stop the run loop. this will result in closeInternal being called by the time join() finishes.
run = false;
interrupt();
try
{
this.join();
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
if (lastException != null)
throw lastException;
}
protected void closeInternal()
{
if (cluster != null)
{
cluster.close();
}
}
private boolean canRetryDriverConnection(Exception e)
{
if (e instanceof DriverException && e.getMessage().contains("Connection thread interrupted"))
return true;
if (e instanceof NoHostAvailableException)
{
if (((NoHostAvailableException) e).getErrors().size() == 1)
{
Throwable cause = ((NoHostAvailableException) e).getErrors().values().iterator().next();
if (cause != null && cause.getCause() instanceof java.nio.channels.ClosedByInterruptException)
{
return true;
}
}
}
return false;
}
}
private ByteBuffer getPartitionKey(Map<String, ByteBuffer> keyColumns)
{
ByteBuffer partitionKey;
if (partitionKeyColumns.size() > 1)
{
ByteBuffer[] keys = new ByteBuffer[partitionKeyColumns.size()];
for (int i = 0; i< keys.length; i++)
keys[i] = keyColumns.get(partitionKeyColumns.get(i).getName());
partitionKey = CompositeType.build(ByteBufferAccessor.instance, keys);
}
else
{
partitionKey = keyColumns.get(partitionKeyColumns.get(0).getName());
}
return partitionKey;
}
/**
* add where clauses for partition keys and cluster columns
*/
private String appendKeyWhereClauses(String cqlQuery)
{
String keyWhereClause = "";
for (ColumnMetadata partitionKey : partitionKeyColumns)
keyWhereClause += String.format("%s = ?", keyWhereClause.isEmpty() ? quote(partitionKey.getName()) : (" AND " + quote(partitionKey.getName())));
for (ColumnMetadata clusterColumn : clusterColumns)
keyWhereClause += " AND " + quote(clusterColumn.getName()) + " = ?";
return cqlQuery + " WHERE " + keyWhereClause;
}
/** Quoting for working with uppercase */
private String quote(String identifier)
{
return "\"" + identifier.replaceAll("\"", "\"\"") + "\"";
}
static class NativeRingCache
{
private final Map<TokenRange, Set<Host>> rangeMap;
private final Metadata metadata;
private final IPartitioner partitioner;
public NativeRingCache(Configuration conf, Metadata metadata)
{
this.partitioner = ConfigHelper.getOutputPartitioner(conf);
this.metadata = metadata;
String keyspace = ConfigHelper.getOutputKeyspace(conf);
this.rangeMap = metadata.getTokenRanges()
.stream()
.collect(toMap(p -> p, p -> metadata.getReplicas('"' + keyspace + '"', p)));
}
public TokenRange getRange(ByteBuffer key)
{
Token t = partitioner.getToken(key);
com.datastax.driver.core.Token driverToken = metadata.newToken(partitioner.getTokenFactory().toString(t));
for (TokenRange range : rangeMap.keySet())
{
if (range.contains(driverToken))
{
return range;
}
}
throw new RuntimeException("Invalid token information returned by describe_ring: " + rangeMap);
}
public List<InetAddress> getEndpoints(TokenRange range)
{
Set<Host> hostSet = rangeMap.get(range);
List<InetAddress> addresses = new ArrayList<>(hostSet.size());
for (Host host: hostSet)
{
addresses.add(host.getAddress());
}
return addresses;
}
}
}

216
src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java
View File

@ -1,216 +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.cql3;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.HostDistance;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.policies.LoadBalancingPolicy;
import com.google.common.base.Function;
import com.google.common.collect.Iterators;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.CopyOnWriteArraySet;
/**
* This load balancing policy is intended to be used only for CqlRecordReader when it fetches a particular split.
* <p/>
* It chooses alive hosts only from the set of the given replicas - because the connection is used to load the data from
* the particular split, with a strictly defined list of the replicas, it is pointless to try the other nodes.
* The policy tracks which of the replicas are alive, and when a new query plan is requested, it returns those replicas
* in the following order:
* <ul>
* <li>the local node</li>
* <li>the collection of the remaining hosts (which is shuffled on each request)</li>
* </ul>
*/
class LimitedLocalNodeFirstLocalBalancingPolicy implements LoadBalancingPolicy
{
private final static Logger logger = LoggerFactory.getLogger(LimitedLocalNodeFirstLocalBalancingPolicy.class);
private final static Set<InetAddress> localAddresses = Collections.unmodifiableSet(getLocalInetAddresses());
private final CopyOnWriteArraySet<Host> liveReplicaHosts = new CopyOnWriteArraySet<>();
private final Set<InetAddress> replicaAddresses = new HashSet<>();
private final Set<String> allowedDCs = new CopyOnWriteArraySet<>();
public LimitedLocalNodeFirstLocalBalancingPolicy(String[] replicas)
{
for (String replica : replicas)
{
try
{
InetAddress[] addresses = InetAddress.getAllByName(replica);
Collections.addAll(replicaAddresses, addresses);
}
catch (UnknownHostException e)
{
logger.warn("Invalid replica host name: {}, skipping it", replica);
}
}
if (logger.isTraceEnabled())
logger.trace("Created instance with the following replicas: {}", Arrays.asList(replicas));
}
@Override
public void init(Cluster cluster, Collection<Host> hosts)
{
// first find which DCs the user defined
Set<String> dcs = new HashSet<>();
for (Host host : hosts)
{
if (replicaAddresses.contains(host.getAddress()))
dcs.add(host.getDatacenter());
}
// filter to all nodes within the targeted DCs
List<Host> replicaHosts = new ArrayList<>();
for (Host host : hosts)
{
if (dcs.contains(host.getDatacenter()))
replicaHosts.add(host);
}
liveReplicaHosts.addAll(replicaHosts);
allowedDCs.addAll(dcs);
logger.trace("Initialized with replica hosts: {}", replicaHosts);
}
@Override
public void close()
{
//
}
@Override
public HostDistance distance(Host host)
{
if (isLocalHost(host))
{
return HostDistance.LOCAL;
}
else
{
return HostDistance.REMOTE;
}
}
@Override
public Iterator<Host> newQueryPlan(String keyspace, Statement statement)
{
List<Host> local = new ArrayList<>(1);
List<Host> remote = new ArrayList<>(liveReplicaHosts.size());
for (Host liveReplicaHost : liveReplicaHosts)
{
if (isLocalHost(liveReplicaHost))
{
local.add(liveReplicaHost);
}
else
{
remote.add(liveReplicaHost);
}
}
Collections.shuffle(remote);
logger.trace("Using the following hosts order for the new query plan: {} | {}", local, remote);
return Iterators.concat(local.iterator(), remote.iterator());
}
@Override
public void onAdd(Host host)
{
if (liveReplicaHosts.contains(host))
{
liveReplicaHosts.add(host);
logger.trace("Added a new host {}", host);
}
}
@Override
public void onUp(Host host)
{
if (liveReplicaHosts.contains(host))
{
liveReplicaHosts.add(host);
logger.trace("The host {} is now up", host);
}
}
@Override
public void onDown(Host host)
{
if (liveReplicaHosts.remove(host))
{
logger.trace("The host {} is now down", host);
}
}
@Override
public void onRemove(Host host)
{
if (liveReplicaHosts.remove(host))
{
logger.trace("Removed the host {}", host);
}
}
public void onSuspected(Host host)
{
// not supported by this load balancing policy
}
private static boolean isLocalHost(Host host)
{
InetAddress hostAddress = host.getAddress();
return hostAddress.isLoopbackAddress() || localAddresses.contains(hostAddress);
}
private static Set<InetAddress> getLocalInetAddresses()
{
try
{
return Sets.newHashSet(Iterators.concat(
Iterators.transform(
Iterators.forEnumeration(NetworkInterface.getNetworkInterfaces()),
new Function<NetworkInterface, Iterator<InetAddress>>()
{
@Override
public Iterator<InetAddress> apply(NetworkInterface netIface)
{
return Iterators.forEnumeration(netIface.getInetAddresses());
}
})));
}
catch (SocketException e)
{
logger.warn("Could not retrieve local network interfaces.", e);
return Collections.emptySet();
}
}
}

23
src/java/org/apache/cassandra/hadoop/package-info.java
View File

@ -1,23 +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.
*/
/**
* This package was deprecated. See CASSANDRA-16984.
*/
@Deprecated
package org.apache.cassandra.hadoop;

2
test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java
View File

@ -30,7 +30,7 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.commons.math.stat.descriptive.DescriptiveStatistics;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

20
test/simulator/main/org/apache/cassandra/simulator/debug/SelfReconcilingRandom.java
View File

@ -20,24 +20,29 @@ package org.apache.cassandra.simulator.debug;
import java.util.function.Supplier;
import org.agrona.collections.Long2LongHashMap;
import org.apache.cassandra.simulator.RandomSource;
import org.hsqldb.lib.IntKeyLongValueHashMap;
import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM;
public class SelfReconcilingRandom implements Supplier<RandomSource>
{
static class Map extends IntKeyLongValueHashMap
static class Map extends Long2LongHashMap
{
public Map(long missingValue)
{
super(missingValue);
}
public boolean put(int i, long v)
{
int size = this.size();
super.addOrRemove((long)i, (long)v, (Object)null, (Object)null, false);
super.put(i, v);
return size != this.size();
}
}
final Map map = new Map();
long[] tmp = new long[1];
final Map map = new Map(Long.MIN_VALUE);
boolean isNextPrimary = true;
static abstract class AbstractVerifying extends RandomSource.Abstract
@ -113,10 +118,11 @@ public class SelfReconcilingRandom implements Supplier<RandomSource>
{
void next(long v)
{
if (!map.get(++cur, tmp))
long value = map.get(++cur);
if (value == Long.MIN_VALUE)
throw failWithOOM();
map.remove(cur);
if (tmp[0] != v)
if (value != v)
throw failWithOOM();
}