diff --git a/CHANGES.txt b/CHANGES.txt index cf982ddcce..54d4ddf26a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -10,6 +10,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) diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java index 3e69c2d7db..4552d87e33 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java +++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java @@ -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> 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> 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> .withPartitioner(ConfigHelper.getOutputPartitioner(conf)) .inDirectory(outputDir) .withBufferSizeInMB(Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64"))) + .withPartitioner(partitioner) .build(); } diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java index dc3c174194..223a848b1a 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java @@ -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 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 tupleToKeyMap(Tuple t) throws IOException { Map keys = new HashMap(); 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 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 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 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=][&auth_provider=][&trust_store_path=]" + "[&key_store_path=][&trust_store_password=]" + "[&key_store_password=][&cipher_suites=][&input_cql=]" + - "[columns=][where_clause=]]': " + e.getMessage()); - } + "[columns=][where_clause=]" + + "[&bulk_cf_schema=bulk_cf_schema][&bulk_insert_statement=bulk_insert_statement][&bulk_table_alias=]" + + "[&bulk_output_location=][&bulk_buff_size=][&bulk_delete_source=]" + + "[&bulk_stream_throttle=][&bulk_max_failed_hosts=]]': " + e.getMessage()); + } } public ByteBuffer nullToBB() diff --git a/test/conf/cassandra_pig.yaml b/test/conf/cassandra_pig.yaml new file mode 100644 index 0000000000..68615cfaa0 --- /dev/null +++ b/test/conf/cassandra_pig.yaml @@ -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 diff --git a/test/pig/org/apache/cassandra/pig/CqlTableTest.java b/test/pig/org/apache/cassandra/pig/CqlTableTest.java index 72fdd5a26a..3902fce7d9 100644 --- a/test/pig/org/apache/cassandra/pig/CqlTableTest.java +++ b/test/pig/org/apache/cassandra/pig/CqlTableTest.java @@ -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 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 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); + } } diff --git a/test/pig/org/apache/cassandra/pig/PigTestBase.java b/test/pig/org/apache/cassandra/pig/PigTestBase.java index f556e66cb6..067397bd6b 100644 --- a/test/pig/org/apache/cassandra/pig/PigTestBase.java +++ b/test/pig/org/apache/cassandra/pig/PigTestBase.java @@ -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