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

Merge branch 'cassandra-3.0' into trunk

This commit is contained in:
Aleksey Yeschenko 2015-09-15 16:16:56 +01:00
parent ef54c6c72f 7ca4db0a11
commit 1944e402cf
6 changed files with 203 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -14,6 +14,7 @@ Merged from 2.2:
* Defer default role manager setup until all nodes are on 2.2+ (CASSANDRA-9761)
* Handle missing RoleManager in config after upgrade to 2.2 (CASSANDRA-10209)
Merged from 2.1:
* (Pig) support BulkOutputFormat as a URL parameter (CASSANDRA-7410)
* BATCH statement is broken in cqlsh (CASSANDRA-10272)
* (cqlsh) Make cqlsh PEP8 Compliant (CASSANDRA-10066)
* (cqlsh) Fix error when starting cqlsh with --debug (CASSANDRA-10282)

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

@ -28,10 +28,11 @@ import java.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
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;
@ -84,6 +85,7 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
private String insertStatement;
private File outputDir;
private boolean deleteSrc;
private IPartitioner partitioner;
CqlBulkRecordWriter(TaskAttemptContext context) throws IOException
{
@ -124,6 +126,14 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
insertStatement = CqlBulkOutputFormat.getTableInsertStatement(conf, table);
outputDir = getTableDirectory();
deleteSrc = CqlBulkOutputFormat.getDeleteSourceOnSuccess(conf);
try
{
partitioner = ConfigHelper.getInputPartitioner(conf);
}
catch (Exception e)
{
partitioner = Murmur3Partitioner.instance;
}
}
protected String getOutputLocation() throws IOException
@ -144,6 +154,7 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
.withPartitioner(ConfigHelper.getOutputPartitioner(conf))
.inDirectory(outputDir)
.withBufferSizeInMB(Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64")))
.withPartitioner(partitioner)
.build();
}

171
src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
View File

@ -34,12 +34,15 @@ import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TableMetadata;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.exceptions.AuthenticationException;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat;
import org.apache.cassandra.hadoop.cql3.CqlBulkRecordWriter;
import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
import org.apache.cassandra.hadoop.cql3.CqlRecordReader;
import org.apache.cassandra.serializers.CollectionSerializer;
@ -82,6 +85,7 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
protected int nativeProtocolVersion = 1;
private static final Logger logger = LoggerFactory.getLogger(CqlNativeStorage.class);
private static String BULK_OUTPUT_FORMAT = "org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat";
private int pageSize = 1000;
private String columns;
private String outputQuery;
@ -109,6 +113,16 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
private String nativeSSLCipherSuites;
private String inputCql;
private boolean bulkOutputFormat = false;
private String bulkCfSchema;
private String bulkInsertStatement;
private String bulkOutputLocation;
private int bulkBuffSize = -1;
private int bulkStreamThrottle = -1;
private int bulkMaxFailedHosts = -1;
private boolean bulkDeleteSourceOnSuccess = true;
private String bulkTableAlias;
public CqlNativeStorage()
{
this(1000);
@ -268,56 +282,21 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
return client.getCluster().getMetadata().getKeyspace(Metadata.quote(keyspace)).getTable(Metadata.quote(column_family));
}
/** output: (((name, value), (name, value)), (value ... value), (value...value)) */
public void putNext(Tuple t) throws IOException
{
if (t.size() < 1)
{
// simply nothing here, we can't even delete without a key
logger.warn("Empty output skipped, filter empty tuples to suppress this warning");
return;
}
if (t.getType(0) == DataType.TUPLE)
{
if (t.getType(1) == DataType.TUPLE)
{
Map<String, ByteBuffer> key = tupleToKeyMap((Tuple)t.get(0));
cqlQueryFromTuple(key, t, 1);
}
else
throw new IOException("Second argument in output must be a tuple");
}
else
throw new IOException("First argument in output must be a tuple");
}
/** convert key tuple to key map */
private Map<String, ByteBuffer> tupleToKeyMap(Tuple t) throws IOException
{
Map<String, ByteBuffer> keys = new HashMap<String, ByteBuffer>();
for (int i = 0; i < t.size(); i++)
{
if (t.getType(i) == DataType.TUPLE)
{
Tuple inner = (Tuple) t.get(i);
if (inner.size() == 2)
{
Object name = inner.get(0);
if (name != null)
{
keys.put(name.toString(), objToBB(inner.get(1)));
}
else
throw new IOException("Key name was empty");
}
else
throw new IOException("Keys were not in name and value pairs");
}
else
{
if (t.getType(i) != DataType.TUPLE)
throw new IOException("keys was not a tuple");
}
Tuple inner = (Tuple) t.get(i);
if (inner.size() != 2)
throw new IOException("Keys were not in name and value pairs");
Object name = inner.get(0);
if (name == null)
throw new IOException("Key name was empty");
keys.put(name.toString(), objToBB(inner.get(1)));
}
return keys;
}
@ -412,21 +391,16 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
{
for (int i = offset; i < t.size(); i++)
{
if (t.getType(i) == DataType.TUPLE)
{
Tuple inner = (Tuple) t.get(i);
if (inner.size() > 0)
{
List<ByteBuffer> bindedVariables = bindedVariablesFromTuple(inner);
if (bindedVariables.size() > 0)
sendCqlQuery(key, bindedVariables);
else
throw new IOException("Missing binded variables");
}
}
else
{
if (t.getType(i) != DataType.TUPLE)
throw new IOException("Output type was not a tuple");
Tuple inner = (Tuple) t.get(i);
if (inner.size() > 0)
{
List<ByteBuffer> bindedVariables = bindedVariablesFromTuple(inner);
if (bindedVariables.size() <= 0)
throw new IOException("Missing binded variables");
sendCqlQuery(key, bindedVariables);
}
}
}
@ -547,6 +521,37 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
return property.getProperty(StorageHelper.PARTITION_FILTER_SIGNATURE);
}
/**
* output: (((name, value), (name, value)), (value ... value), (value...value))
* bulk output: ((value ... value), (value...value))
*
* */
public void putNext(Tuple t) throws IOException
{
if (t.size() < 1)
{
// simply nothing here, we can't even delete without a key
logger.warn("Empty output skipped, filter empty tuples to suppress this warning");
return;
}
if (t.getType(0) != DataType.TUPLE)
throw new IOException("First argument in output must be a tuple");
if (!bulkOutputFormat && t.getType(1) != DataType.TUPLE)
throw new IOException("Second argument in output must be a tuple");
if (bulkOutputFormat)
{
cqlQueryFromTuple(null, t, 0);
}
else
{
Map<String, ByteBuffer> key = tupleToKeyMap((Tuple)t.get(0));
cqlQueryFromTuple(key, t, 1);
}
}
/** set read configuration settings */
public void setLocation(String location, Job job) throws IOException
{
@ -672,6 +677,32 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
CqlConfigHelper.setOutputCql(conf, outputQuery);
if (bulkOutputFormat)
{
DEFAULT_OUTPUT_FORMAT = BULK_OUTPUT_FORMAT;
if (bulkCfSchema != null)
CqlBulkOutputFormat.setTableSchema(conf, column_family, bulkCfSchema);
else
throw new IOException("bulk_cf_schema is missing in input url parameter");
if (bulkInsertStatement != null)
CqlBulkOutputFormat.setTableInsertStatement(conf, column_family, bulkInsertStatement);
else
throw new IOException("bulk_insert_statement is missing in input url parameter");
if (bulkTableAlias != null)
CqlBulkOutputFormat.setTableAlias(conf, bulkTableAlias, column_family);
CqlBulkOutputFormat.setDeleteSourceOnSuccess(conf, bulkDeleteSourceOnSuccess);
if (bulkOutputLocation != null)
conf.set(CqlBulkRecordWriter.OUTPUT_LOCATION, bulkOutputLocation);
if (bulkBuffSize > 0)
conf.set(CqlBulkRecordWriter.BUFFER_SIZE_IN_MB, String.valueOf(bulkBuffSize));
if (bulkStreamThrottle > 0)
conf.set(CqlBulkRecordWriter.STREAM_THROTTLE_MBITS, String.valueOf(bulkStreamThrottle));
if (bulkMaxFailedHosts > 0)
conf.set(CqlBulkRecordWriter.MAX_FAILED_HOSTS, String.valueOf(bulkMaxFailedHosts));
if (partitionerClass!= null)
ConfigHelper.setInputPartitioner(conf, partitionerClass);
}
setConnectionInformation();
if (ConfigHelper.getOutputRpcPort(conf) == 0)
@ -773,6 +804,25 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
if (urlQuery.containsKey("output_query"))
outputQuery = urlQuery.get("output_query");
if (urlQuery.containsKey("bulk_output_format"))
bulkOutputFormat = Boolean.valueOf(urlQuery.get("bulk_output_format"));
if (urlQuery.containsKey("bulk_cf_schema"))
bulkCfSchema = urlQuery.get("bulk_cf_schema");
if (urlQuery.containsKey("bulk_insert_statement"))
bulkInsertStatement = urlQuery.get("bulk_insert_statement");
if (urlQuery.containsKey("bulk_output_location"))
bulkOutputLocation = urlQuery.get("bulk_output_location");
if (urlQuery.containsKey("bulk_buff_size"))
bulkBuffSize = Integer.valueOf(urlQuery.get("bulk_buff_size"));
if (urlQuery.containsKey("bulk_stream_throttle"))
bulkStreamThrottle = Integer.valueOf(urlQuery.get("bulk_stream_throttle"));
if (urlQuery.containsKey("bulk_max_failed_hosts"))
bulkMaxFailedHosts = Integer.valueOf(urlQuery.get("bulk_max_failed_hosts"));
if (urlQuery.containsKey("bulk_delete_source"))
bulkDeleteSourceOnSuccess = Boolean.parseBoolean(urlQuery.get("bulk_delete_source"));
if (urlQuery.containsKey("bulk_table_alias"))
bulkTableAlias = urlQuery.get("bulk_table_alias");
//split size
if (urlQuery.containsKey("split_size"))
splitSize = Integer.parseInt(urlQuery.get("split_size"));
@ -855,8 +905,11 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
"[&keep_alive=<keep_alive>][&auth_provider=<auth_provider>][&trust_store_path=<trust_store_path>]" +
"[&key_store_path=<key_store_path>][&trust_store_password=<trust_store_password>]" +
"[&key_store_password=<key_store_password>][&cipher_suites=<cipher_suites>][&input_cql=<input_cql>]" +
"[columns=<columns>][where_clause=<where_clause>]]': " + e.getMessage());
}
"[columns=<columns>][where_clause=<where_clause>]" +
"[&bulk_cf_schema=bulk_cf_schema][&bulk_insert_statement=bulk_insert_statement][&bulk_table_alias=<bulk_table_alias>]" +
"[&bulk_output_location=<bulk_output_location>][&bulk_buff_size=<bulk_buff_size>][&bulk_delete_source=<bulk_delete_source>]" +
"[&bulk_stream_throttle=<bulk_stream_throttle>][&bulk_max_failed_hosts=<bulk_max_failed_hosts>]]': " + e.getMessage());
}
}
public ByteBuffer nullToBB()

41
test/conf/cassandra_pig.yaml Normal file
View File

@ -0,0 +1,41 @@
#
# Warning!
# Consider the effects on 'o.a.c.i.s.LegacySSTableTest' before changing schemas in this file.
#
cluster_name: Test Cluster
memtable_allocation_type: offheap_objects
commitlog_sync: batch
commitlog_sync_batch_window_in_ms: 1.0
commitlog_segment_size_in_mb: 5
commitlog_directory: build/test/cassandra/commitlog
partitioner: org.apache.cassandra.dht.Murmur3Partitioner
listen_address: 127.0.0.1
storage_port: 7010
rpc_port: 9170
start_native_transport: true
native_transport_port: 9042
column_index_size_in_kb: 4
saved_caches_directory: build/test/cassandra/saved_caches
data_file_directories:
- build/test/cassandra/data
disk_access_mode: mmap
seed_provider:
- class_name: org.apache.cassandra.locator.SimpleSeedProvider
parameters:
- seeds: "127.0.0.1"
endpoint_snitch: org.apache.cassandra.locator.SimpleSnitch
dynamic_snitch: true
request_scheduler: org.apache.cassandra.scheduler.RoundRobinScheduler
request_scheduler_id: keyspace
server_encryption_options:
internode_encryption: none
keystore: conf/.keystore
keystore_password: cassandra
truststore: conf/.truststore
truststore_password: cassandra
incremental_backups: true
concurrent_compactors: 4
compaction_throughput_mb_per_sec: 0
row_cache_class_name: org.apache.cassandra.cache.OHCProvider
row_cache_size_in_mb: 16
enable_user_defined_functions: true

35
test/pig/org/apache/cassandra/pig/CqlTableTest.java
View File

@ -44,6 +44,10 @@ public class CqlTableTest extends PigTestBase
"CREATE INDEX test_b on test (b);",
"CREATE TABLE moredata (x int PRIMARY KEY, y int);",
"CREATE TABLE test_bulk (a int PRIMARY KEY, b int);",
"INSERT INTO test_bulk (a,b) VALUES (1,1);",
"INSERT INTO test_bulk (a,b) VALUES (2,2);",
"INSERT INTO test_bulk (a,b) VALUES (3,3);",
"INSERT INTO test (a,b) VALUES (1,1);",
"INSERT INTO test (a,b) VALUES (2,2);",
"INSERT INTO test (a,b) VALUES (3,3);",
@ -152,10 +156,13 @@ public class CqlTableTest extends PigTestBase
//input_cql=select * from test where token(a) > ? and token(a) <= ?
pig.registerQuery("result= LOAD 'cql://cql3ks/test?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20test%20where%20token(a)%20%3E%20%3F%20and%20token(a)%20%3C%3D%20%3F' USING CqlNativeStorage();");
Iterator<Tuple> it = pig.openIterator("result");
int count = 0;
while (it.hasNext()) {
Tuple t = it.next();
Assert.assertEquals(t.get(0), t.get(1));
count ++;
}
Assert.assertEquals(6, count);
}
@Test
@ -265,4 +272,32 @@ public class CqlTableTest extends PigTestBase
Assert.fail("Can't fetch any data");
}
}
@Test
public void testCqlStorageSingleKeyTableBulkLoad() throws TException, IOException
{
pig.setBatchOn();
//input_cql=select * from moredata where token(x) > ? and token(x) <= ?
pig.registerQuery("moretestvalues= LOAD 'cql://cql3ks/moredata?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20moredata%20where%20token(x)%20%3E%20%3F%20and%20token(x)%20%3C%3D%20%3F' USING CqlNativeStorage();");
pig.registerQuery("insertformat= FOREACH moretestvalues GENERATE TOTUPLE(x, y);");
pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/test_bulk?" + defaultParameters + nativeParameters + "&bulk_output_format=true&bulk_cf_schema=CREATE%20TABLE%20cql3ks.test_bulk%20(a%20int%20PRIMARY%20KEY%2C%20b%20int)&bulk_insert_statement=Insert%20into%20cql3ks.test_bulk(a%2C%20b)%20values(%3F%2C%3F)' USING CqlNativeStorage();");
pig.executeBatch();
//(5,5)
//(6,6)
//(4,4)
//(2,2)
//(3,3)
//(1,1)
//input_cql=select * from test_bulk1 where token(a) > ? and token(a) <= ?
pig.registerQuery("result= LOAD 'cql://cql3ks/test_bulk?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20test_bulk%20where%20token(a)%20%3E%20%3F%20and%20token(a)%20%3C%3D%20%3F' USING CqlNativeStorage();");
Iterator<Tuple> it = pig.openIterator("result");
int count = 0;
while (it.hasNext()) {
Tuple t = it.next();
Assert.assertEquals(t.get(0), t.get(1));
count ++;
}
Assert.assertEquals(6, count);
}
}

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

@ -46,7 +46,7 @@ public class PigTestBase extends SchemaLoader
protected static Configuration conf;
protected static MiniCluster cluster;
protected static PigServer pig;
protected static String defaultParameters= "init_address=localhost&rpc_port=9170&partitioner=org.apache.cassandra.dht.ByteOrderedPartitioner";
protected static String defaultParameters= "init_address=localhost&rpc_port=9170&partitioner=org.apache.cassandra.dht.Murmur3Partitioner";
protected static String nativeParameters = "&core_conns=2&max_conns=10&min_simult_reqs=3&max_simult_reqs=10&native_timeout=10000000" +
"&native_read_timeout=10000000&send_buff_size=4096&receive_buff_size=4096&solinger=3" +
"&tcp_nodelay=true&reuse_address=true&keep_alive=true&native_port=9042";
@ -54,6 +54,7 @@ public class PigTestBase extends SchemaLoader
static
{
System.setProperty("logback.configurationFile", "logback-test.xml");
System.setProperty("cassandra.config", "cassandra_pig.yaml");
}
@AfterClass