cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From alek...@apache.org
Subject cassandra git commit: (Pig) support BulkOutputFormat as a URL parameter
Date Tue, 15 Sep 2015 15:09:15 GMT
Repository: cassandra
Updated Branches:
  refs/heads/cassandra-2.1 ae5108645 -> c7b407357


(Pig) support BulkOutputFormat as a URL parameter

patch by Alex Liu; reviewed by Piotr Kołaczkowski for CASSANDRA-7410


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/c7b40735
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/c7b40735
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/c7b40735

Branch: refs/heads/cassandra-2.1
Commit: c7b40735789c840529002eb3c11d8731f460d61c
Parents: ae51086
Author: Alex Liu <alex_liu68@yahoo.com>
Authored: Tue Sep 15 16:06:18 2015 +0100
Committer: Aleksey Yeschenko <aleksey@apache.org>
Committed: Tue Sep 15 16:08:54 2015 +0100

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../hadoop/cql3/CqlBulkOutputFormat.java        |  93 +++++++-
 .../hadoop/cql3/CqlBulkRecordWriter.java        |  87 ++++----
 .../cassandra/hadoop/pig/CqlNativeStorage.java  | 213 +++++++++++++------
 .../apache/cassandra/hadoop/pig/CqlStorage.java |   1 -
 .../org/apache/cassandra/tools/BulkLoader.java  |   2 +-
 test/conf/cassandra.yaml                        |   1 +
 .../org/apache/cassandra/pig/CqlTableTest.java  |  36 ++++
 .../org/apache/cassandra/pig/PigTestBase.java   |   3 +-
 9 files changed, 336 insertions(+), 101 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index dff47fc..5f11049 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.1.10
+ * (Pig) support BulkOutputFormat as a URL parameter (CASSANDRA-7410)
  * BATCH statement is broken in cqlsh (CASSANDRA-10272)
  * Added configurable warning threshold for GC duration (CASSANDRA-8907)
  * (cqlsh) Make cqlsh PEP8 compliant (CASSANDRA-10066)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java
index 887fe8e..7fedb41 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java
@@ -22,6 +22,7 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
 
+import org.apache.cassandra.config.EncryptionOptions;
 import org.apache.cassandra.hadoop.AbstractBulkOutputFormat;
 import org.apache.cassandra.hadoop.ConfigHelper;
 import org.apache.hadoop.conf.Configuration;
@@ -54,6 +55,16 @@ public class CqlBulkOutputFormat extends AbstractBulkOutputFormat<Object,
List<B
     private static final String OUTPUT_CQL_SCHEMA_PREFIX = "cassandra.columnfamily.schema.";
     private static final String OUTPUT_CQL_INSERT_PREFIX = "cassandra.columnfamily.insert.";
     private static final String DELETE_SOURCE = "cassandra.output.delete.source";
+    private static final String OUTPUT_CQL_STORAGE_PORT = "cassandra.storage.port";
+    private static final String OUTPUT_CQL_SSL_STORAGE_PORT = "cassandra.ssl.storage.port";
+    private static final String INTERNODE_ENCRYPTION = "cassandra.internode.encryption";
+    private static final String SERVER_KEYSTORE = "cassandra.server.keystore";
+    private static final String SERVER_KEYSTORE_PASSWORD = "cassandra.server.keystore.password";
+    private static final String SERVER_TRUSTSTORE = "cassandra.server.truststore";
+    private static final String SERVER_TRUSTSTORE_PASSWORD = "cassandra.server.truststore.password";
+    private static final String SERVER_CIPHER_SUITES = "cassandra.server.truststore.password";
+    public static final int DEFAULT_STORAGE_PORT = 7000;
+    public static final int DEFAULT_SSL_STORAGE_PORT = 7001;
   
     /** Fills the deprecated OutputFormat interface for streaming. */
     @Deprecated
@@ -84,7 +95,87 @@ public class CqlBulkOutputFormat extends AbstractBulkOutputFormat<Object,
List<B
     {
         conf.set(OUTPUT_CQL_INSERT_PREFIX + columnFamily, insertStatement);
     }
-    
+
+    public static void setStoragePort(Configuration conf, int port)
+    {
+        conf.set(OUTPUT_CQL_STORAGE_PORT, "" + port);
+    }
+
+    public static void setSSLStoragePort(Configuration conf, int port)
+    {
+        conf.set(OUTPUT_CQL_SSL_STORAGE_PORT, "" + port);
+    }
+
+    public static void setInternodeEncryption(Configuration conf, String encrypt)
+    {
+        conf.set(INTERNODE_ENCRYPTION, encrypt);
+    }
+
+    public static void setServerKeystore(Configuration conf, String keystore)
+    {
+        conf.set(SERVER_KEYSTORE, keystore);
+    }
+
+    public static void setServerKeystorePassword(Configuration conf, String keystorePass)
+    {
+        conf.set(SERVER_KEYSTORE_PASSWORD, keystorePass);
+    }
+
+    public static void setServerTruststore(Configuration conf, String truststore)
+    {
+        conf.set(SERVER_TRUSTSTORE, truststore);
+    }
+
+    public static void setServerTruststorePassword(Configuration conf, String truststorePass)
+    {
+        conf.set(SERVER_TRUSTSTORE_PASSWORD, truststorePass);
+    }
+
+    public static void setServerCipherSuites(Configuration conf, String cipherSuites)
+    {
+        conf.set(SERVER_CIPHER_SUITES, cipherSuites);
+    }
+
+    public static int getStoragePort(Configuration conf)
+    {
+        return conf.getInt(OUTPUT_CQL_STORAGE_PORT, DEFAULT_STORAGE_PORT);
+    }
+
+    public static int getSSLStoragePort(Configuration conf)
+    {
+        return conf.getInt(OUTPUT_CQL_SSL_STORAGE_PORT, DEFAULT_SSL_STORAGE_PORT);
+    }
+
+    public static String getInternodeEncryption(Configuration conf)
+    {
+        return conf.get(INTERNODE_ENCRYPTION, EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none.name());
+    }
+
+    public static String getServerKeystore(Configuration conf)
+    {
+        return conf.get(SERVER_KEYSTORE);
+    }
+
+    public static String getServerTruststore(Configuration conf)
+    {
+        return conf.get(SERVER_TRUSTSTORE);
+    }
+
+    public static String getServerKeystorePassword(Configuration conf)
+    {
+        return conf.get(SERVER_KEYSTORE_PASSWORD);
+    }
+
+    public static String getServerTruststorePassword(Configuration conf)
+    {
+        return conf.get(SERVER_TRUSTSTORE_PASSWORD);
+    }
+
+    public static String getServerCipherSuites(Configuration conf)
+    {
+        return conf.get(SERVER_CIPHER_SUITES);
+    }
+
     public static String getColumnFamilySchema(Configuration conf, String columnFamily)
     {
         String schema = conf.get(OUTPUT_CQL_SCHEMA_PREFIX + columnFamily);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
index e60a240..ced8aa9 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
@@ -19,13 +19,16 @@ package org.apache.cassandra.hadoop.cql3;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
-import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 
-import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.hadoop.AbstractBulkRecordWriter;
 import org.apache.cassandra.hadoop.BulkRecordWriter;
@@ -35,6 +38,9 @@ import org.apache.cassandra.io.sstable.CQLSSTableWriter;
 import org.apache.cassandra.io.sstable.SSTableLoader;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.streaming.StreamState;
+import org.apache.cassandra.thrift.ITransportFactory;
+import org.apache.cassandra.tools.BulkLoader;
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.util.Progressable;
@@ -108,10 +114,7 @@ public class CqlBulkRecordWriter extends AbstractBulkRecordWriter<Object,
List<B
             }
             if (loader == null)
             {
-                ExternalClient externalClient = new ExternalClient(conf);
-                
-                externalClient.addKnownCfs(keyspace, schema);
-
+                BulkLoader.ExternalClient externalClient = getExternalClient(conf);
                 this.loader = new SSTableLoader(outputDir, externalClient, new BulkRecordWriter.NullOutputHandler())
{
                     @Override
                     public void onSuccess(StreamState finalState)
@@ -171,41 +174,53 @@ public class CqlBulkRecordWriter extends AbstractBulkRecordWriter<Object,
List<B
         
         return dir;
     }
-    
-    public static class ExternalClient extends AbstractBulkRecordWriter.ExternalClient
-    {
-        private Map<String, Map<String, CFMetaData>> knownCqlCfs = new HashMap<>();
-        
-        public ExternalClient(Configuration conf)
-        {
-            super(conf);
-        }
 
-        public void addKnownCfs(String keyspace, String cql)
+    private BulkLoader.ExternalClient getExternalClient(Configuration conf)
+    {
+        Set<InetAddress> hosts = new HashSet<InetAddress>();
+        String outputAddress = ConfigHelper.getOutputInitialAddress(conf);
+        if (outputAddress == null) outputAddress = "localhost";
+        String[] nodes = outputAddress.split(",");
+        for (String node : nodes)
         {
-            Map<String, CFMetaData> cfs = knownCqlCfs.get(keyspace);
-            
-            if (cfs == null)
+            try
             {
-                cfs = new HashMap<>();
-                knownCqlCfs.put(keyspace, cfs);
+                hosts.add(InetAddress.getByName(node));
             }
-            
-            CFMetaData metadata = CFMetaData.compile(cql, keyspace);
-            cfs.put(metadata.cfName, metadata);
-        }
-        
-        @Override
-        public CFMetaData getCFMetaData(String keyspace, String cfName)
-        {
-            CFMetaData metadata = super.getCFMetaData(keyspace, cfName);
-            if (metadata != null)
+            catch (UnknownHostException e)
             {
-                return metadata;
+                throw new RuntimeException(e);
             }
-            
-            Map<String, CFMetaData> cfs = knownCqlCfs.get(keyspace);
-            return cfs != null ? cfs.get(cfName) : null;
         }
+        int rpcPort = ConfigHelper.getOutputRpcPort(conf);
+        String username = ConfigHelper.getOutputKeyspaceUserName(conf);
+        String password = ConfigHelper.getOutputKeyspacePassword(conf);
+        ITransportFactory transportFactory = ConfigHelper.getClientTransportFactory(conf);
+        return new BulkLoader.ExternalClient(hosts,
+                rpcPort,
+                username,
+                password,
+                transportFactory,
+                CqlBulkOutputFormat.getStoragePort(conf),
+                CqlBulkOutputFormat.getSSLStoragePort(conf),
+                getServerEncryptOpt(conf));
+    }
+
+    private ServerEncryptionOptions getServerEncryptOpt(Configuration conf)
+    {
+        ServerEncryptionOptions encryptOpt = new ServerEncryptionOptions();
+        String internodeEncrypt = CqlBulkOutputFormat.getInternodeEncryption(conf);
+        if (StringUtils.isEmpty(internodeEncrypt))
+            return encryptOpt;
+
+        encryptOpt.internode_encryption = EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.valueOf(internodeEncrypt);
+        encryptOpt.keystore = CqlBulkOutputFormat.getServerKeystore(conf);
+        encryptOpt.truststore = CqlBulkOutputFormat.getServerTruststore(conf);
+        encryptOpt.keystore_password = CqlBulkOutputFormat.getServerKeystorePassword(conf);
+        encryptOpt.truststore_password = CqlBulkOutputFormat.getServerTruststorePassword(conf);
+        String cipherSuites = CqlBulkOutputFormat.getServerCipherSuites(conf);
+        if (!StringUtils.isEmpty(cipherSuites))
+            encryptOpt.cipher_suites = cipherSuites.replace(" ", "").split(",");
+        return encryptOpt;
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
index 7887085..5287bf5 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
@@ -24,18 +24,21 @@ import java.util.*;
 
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption;
 import org.apache.cassandra.db.BufferCell;
 import org.apache.cassandra.db.Cell;
 import org.apache.cassandra.db.composites.CellNames;
 import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.hadoop.AbstractBulkRecordWriter;
 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.CqlConfigHelper;
 import org.apache.cassandra.hadoop.cql3.CqlRecordReader;
 import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.thrift.ConsistencyLevel;
 import org.apache.cassandra.utils.*;
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.mapreduce.*;
 import org.apache.pig.Expression;
 import org.apache.pig.ResourceSchema;
@@ -54,6 +57,7 @@ import com.datastax.driver.core.Row;
 public class CqlNativeStorage extends AbstractCassandraStorage
 {
     private static final Logger logger = LoggerFactory.getLogger(CqlNativeStorage.class);
+    public static String BULK_OUTPUT_FORMAT = "org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat";
     private int pageSize = 1000;
     private String columns;
     private String outputQuery;
@@ -83,6 +87,22 @@ public class CqlNativeStorage extends AbstractCassandraStorage
     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 int storagePort = CqlBulkOutputFormat.DEFAULT_STORAGE_PORT;
+    private int sslStoragePort = CqlBulkOutputFormat.DEFAULT_SSL_STORAGE_PORT;
+    private String serverKeystore;
+    private String serverKeystorePassword;
+    private String serverTruststore;
+    private String serverTruststorePassword;
+    private String serverCipherSuites;
+    private String internodeEncrypt;
+
     public CqlNativeStorage()
     {
         this(1000);
@@ -386,57 +406,22 @@ public class CqlNativeStorage extends AbstractCassandraStorage
         return keys;
     }
 
-
-    /** 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;
     }
@@ -446,21 +431,16 @@ public class CqlNativeStorage extends AbstractCassandraStorage
     {
         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);
             }
         }
     }
@@ -561,6 +541,37 @@ public class CqlNativeStorage extends AbstractCassandraStorage
         return property.getProperty(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
     {
@@ -688,6 +699,42 @@ public class CqlNativeStorage extends AbstractCassandraStorage
         ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
         CqlConfigHelper.setOutputCql(conf, outputQuery);
 
+        if (bulkOutputFormat)
+        {
+            DEFAULT_OUTPUT_FORMAT = BULK_OUTPUT_FORMAT;
+            if (bulkCfSchema != null)
+                CqlBulkOutputFormat.setColumnFamilySchema(conf, column_family, bulkCfSchema);
+            else
+                throw new IOException("bulk_cf_schema is missing in input url parameter");
+            if (bulkInsertStatement != null)
+                CqlBulkOutputFormat.setColumnFamilyInsertStatement(conf, column_family, bulkInsertStatement);
+            else
+                throw new IOException("bulk_insert_statement is missing in input url parameter");
+            if (bulkOutputLocation != null)
+                conf.set(AbstractBulkRecordWriter.OUTPUT_LOCATION, bulkOutputLocation);
+            if (bulkBuffSize > 0)
+                conf.set(AbstractBulkRecordWriter.BUFFER_SIZE_IN_MB, String.valueOf(bulkBuffSize));
+            if (bulkStreamThrottle > 0)
+                conf.set(AbstractBulkRecordWriter.STREAM_THROTTLE_MBITS, String.valueOf(bulkStreamThrottle));
+            if (bulkMaxFailedHosts > 0)
+                conf.set(AbstractBulkRecordWriter.MAX_FAILED_HOSTS, String.valueOf(bulkMaxFailedHosts));
+            CqlBulkOutputFormat.setSSLStoragePort(conf, sslStoragePort);
+            CqlBulkOutputFormat.setStoragePort(conf, storagePort);
+            if (serverEncrypted())
+            {
+                if (!StringUtils.isEmpty(serverKeystore))
+                    CqlBulkOutputFormat.setServerKeystore(conf, serverKeystore);
+                if (!StringUtils.isEmpty(serverTruststore))
+                    CqlBulkOutputFormat.setServerTruststore(conf, serverTruststore);
+                if (!StringUtils.isEmpty(serverKeystorePassword))
+                    CqlBulkOutputFormat.setServerKeystorePassword(conf, serverKeystorePassword);
+                if (!StringUtils.isEmpty(serverTruststorePassword))
+                    CqlBulkOutputFormat.setServerTruststorePassword(conf, serverTruststorePassword);
+                if (!StringUtils.isEmpty(serverCipherSuites))
+                    CqlBulkOutputFormat.setServerCipherSuites(conf, serverCipherSuites);
+            }
+        }
+
         setConnectionInformation();
 
         if (ConfigHelper.getOutputRpcPort(conf) == 0)
@@ -700,6 +747,12 @@ public class CqlNativeStorage extends AbstractCassandraStorage
         initSchema(storeSignature);
     }
 
+    private boolean serverEncrypted()
+    {
+        return !StringUtils.isEmpty(internodeEncrypt) && 
+                InternodeEncryption.none != InternodeEncryption.valueOf(internodeEncrypt.toLowerCase());
+    }
+
     private void setLocationFromUri(String location) throws IOException
     {
         try
@@ -720,6 +773,37 @@ public class CqlNativeStorage extends AbstractCassandraStorage
                 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("storage_port"))
+                    storagePort = Integer.valueOf(urlQuery.get("storage_port"));
+                if (urlQuery.containsKey("ssl_storage_port"))
+                    sslStoragePort = Integer.valueOf(urlQuery.get("ssl_storage_port"));
+                if (urlQuery.containsKey("internode_encrypt"))
+                    internodeEncrypt = urlQuery.get("internode_encrypt");
+                if (urlQuery.containsKey("server_keystore"))
+                    serverKeystore = urlQuery.get("server_keystore");
+                if (urlQuery.containsKey("server_truststore"))
+                    serverTruststore = urlQuery.get("server_truststore");
+                if (urlQuery.containsKey("server_keystore_pass"))
+                    serverKeystorePassword = urlQuery.get("server_keystore_pass");
+                if (urlQuery.containsKey("server_truststore_pass"))
+                    serverTruststorePassword = urlQuery.get("server_truststore_pass");
+                if (urlQuery.containsKey("server_cipher_suites"))
+                    serverCipherSuites = urlQuery.get("server_cipher_suites");
+
                 //split size
                 if (urlQuery.containsKey("split_size"))
                     splitSize = Integer.parseInt(urlQuery.get("split_size"));
@@ -804,8 +888,15 @@ public class CqlNativeStorage extends AbstractCassandraStorage
                     "[&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_output_location=<bulk_output_location>][&bulk_buff_size=<bulk_buff_size>]"
+
+                    "[&storage_port=<storage_port>][&ssl_storage_port=<ssl_storage_port>]"
+
+                    "[&server_keystore=<server_keystore>][&server_keystore_pass=<server_keystore_pass>]"
+
+                    "[&server_truststore=<server_truststore>][&server_truststore_pass=<server_truststore_pass>]"
+
+                    "[&server_cipher_suites=<server_cipher_suites>][&internode_encrypt=<internode_encrypt>]"
+
+                    "[&bulk_stream_throttle=<bulk_stream_throttle>][&bulk_max_failed_hosts=<bulk_max_failed_hosts>]]':
" +  e.getMessage());
+         }
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
index c7277fa..66583ec 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
@@ -38,4 +38,3 @@ public class CqlStorage extends CqlNativeStorage
         super(pageSize);
     }
 }
-

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/src/java/org/apache/cassandra/tools/BulkLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java
index 88a4404..f4b30cb 100644
--- a/src/java/org/apache/cassandra/tools/BulkLoader.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoader.java
@@ -254,7 +254,7 @@ public class BulkLoader
         }
     }
 
-    static class ExternalClient extends SSTableLoader.Client
+    public static class ExternalClient extends SSTableLoader.Client
     {
         private final Map<String, CFMetaData> knownCfs = new HashMap<>();
         private final Set<InetAddress> hosts;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/test/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml
index ec988e2..7be72dd 100644
--- a/test/conf/cassandra.yaml
+++ b/test/conf/cassandra.yaml
@@ -10,6 +10,7 @@ commitlog_segment_size_in_mb: 5
 partitioner: org.apache.cassandra.dht.ByteOrderedPartitioner
 listen_address: 127.0.0.1
 storage_port: 7010
+ssl_storage_port: 7011
 rpc_port: 9170
 start_native_transport: true
 native_transport_port: 9042

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/test/pig/org/apache/cassandra/pig/CqlTableTest.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/CqlTableTest.java b/test/pig/org/apache/cassandra/pig/CqlTableTest.java
index 4ca043d..2e1758e 100644
--- a/test/pig/org/apache/cassandra/pig/CqlTableTest.java
+++ b/test/pig/org/apache/cassandra/pig/CqlTableTest.java
@@ -53,6 +53,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);",
@@ -160,10 +164,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
@@ -310,4 +317,33 @@ public class CqlTableTest extends PigTestBase
             Assert.fail("Can't fetch any data");
         }
     }
+
+    @Test
+    public void testCqlStorageSingleKeyTableBulkLoad()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, 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);
+     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b40735/test/pig/org/apache/cassandra/pig/PigTestBase.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/PigTestBase.java b/test/pig/org/apache/cassandra/pig/PigTestBase.java
index 4b3e422..e6964f8 100644
--- a/test/pig/org/apache/cassandra/pig/PigTestBase.java
+++ b/test/pig/org/apache/cassandra/pig/PigTestBase.java
@@ -65,7 +65,8 @@ 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.ByteOrderedPartitioner"
+
+                                               "&storage_port=7010&ssl_storage_port=7011&internode_encrypt=NONE";
     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";


Mime
View raw message