Merge branch 'master' into branch-2.4

Author: gkorland

.travis.yml

@@ -1,13 +1,18 @@
 sudo: required
 language: scala
-scala:
-    - 2.11.2
+jdk:
+    - openjdk8
 install:
-    - wget http://download.redis.io/releases/redis-5.0.1.tar.gz
-    - tar -xzvf redis-5.0.1.tar.gz
-    - make -C redis-5.0.1 -j4
-    - export PATH=$PWD/redis-5.0.1/src:$PATH
-script: make test
+    - wget http://download.redis.io/releases/redis-6.0.3.tar.gz
+    - tar -xzvf redis-6.0.3.tar.gz
+    - make -C redis-6.0.3 -j4 BUILD_TLS=yes
+    - export PATH=$PWD/redis-6.0.3/src:$PATH
+script:
+    - make test # test with scala 2.11
+    - sleep 5s # let redis exit gracefully (we use kill, not kill -9 in makefile)
+    - ps aux | grep redis
+    - ./dev/change-scala-version.sh 2.12 # switch to scala 2.12
+    - make test # test with scala 2.12
 cache:
     directories:
         - $HOME/.m2

Makefile

@@ -9,6 +9,23 @@ appendonly no
 requirepass passwd
 endef
 
+# STANDALONE REDIS NODE WITH SSL
+define REDIS_STANDALONE_NODE_CONF_SSL
+daemonize yes
+port 0
+pidfile /tmp/redis_standalone_node__ssl_for_spark-redis.pid
+logfile /tmp/redis_standalone_node_ssl_for_spark-redis.log
+save ""
+appendonly no
+requirepass passwd
+tls-auth-clients no
+tls-port 6380
+tls-cert-file ./src/test/resources/tls/redis.crt
+tls-key-file ./src/test/resources/tls/redis.key
+tls-ca-cert-file ./src/test/resources/tls/ca.crt
+tls-dh-params-file ./src/test/resources/tls/redis.dh
+endef
+
 # CLUSTER REDIS NODES
 define REDIS_CLUSTER_NODE1_CONF
 daemonize yes
@@ -44,12 +61,14 @@ cluster-config-file /tmp/redis_cluster_node3_for_spark-redis.conf
 endef
 
 export REDIS_STANDALONE_NODE_CONF
+export REDIS_STANDALONE_NODE_CONF_SSL
 export REDIS_CLUSTER_NODE1_CONF
 export REDIS_CLUSTER_NODE2_CONF
 export REDIS_CLUSTER_NODE3_CONF
 
 start-standalone:
 	echo "$$REDIS_STANDALONE_NODE_CONF" | redis-server -
+	echo "$$REDIS_STANDALONE_NODE_CONF_SSL" | redis-server -
 
 
 start-cluster:
@@ -72,7 +91,8 @@ start:
 
 stop-standalone:
 	kill `cat /tmp/redis_standalone_node_for_spark-redis.pid`
-
+	kill `cat /tmp/redis_standalone_node__ssl_for_spark-redis.pid`
+	
 stop-cluster:
 	kill `cat /tmp/redis_cluster_node1_for_spark-redis.pid` || true
 	kill `cat /tmp/redis_cluster_node2_for_spark-redis.pid` || true
@@ -93,7 +113,7 @@ test:
 	make start
 	# with --batch-mode maven doesn't print 'Progress: 125/150kB', the progress lines take up 90% of the log and causes
 	# Travis build to fail with 'The job exceeded the maximum log length, and has been terminated'
-	mvn clean test -B
+	mvn clean test -B  -DargLine="-Djavax.net.ssl.trustStorePassword=password -Djavax.net.ssl.trustStore=./src/test/resources/tls/clientkeystore -Djavax.net.ssl.trustStoreType=jceks"
 	make stop
 
 benchmark:

README.md

@@ -1,8 +1,9 @@
 [![license](https://img.shields.io/github/license/RedisLabs/spark-redis.svg)](https://github.com/RedisLabs/spark-redis)
 [![GitHub issues](https://img.shields.io/github/release/RedisLabs/spark-redis.svg)](https://github.com/RedisLabs/spark-redis/releases/latest)
 [![Build Status](https://travis-ci.org/RedisLabs/spark-redis.svg)](https://travis-ci.org/RedisLabs/spark-redis)
-[![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.redislabs/spark-redis/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.redislabs/spark-redis)
-[![Javadocs](https://www.javadoc.io/badge/com.redislabs/spark-redis.svg)](https://www.javadoc.io/doc/com.redislabs/spark-redis)
+[![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.redislabs/spark-redis_2.11/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.redislabs/spark-redis_2.11)
+[![Javadocs](https://www.javadoc.io/badge/com.redislabs/spark-redis_2.11.svg)](https://www.javadoc.io/doc/com.redislabs/spark-redis_2.11)
+[![Gitter](https://badges.gitter.im/RedisLabs/spark-redis.svg)](https://gitter.im/RedisLabs/spark-redis?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge)
 <!--[![Codecov](https://codecov.io/gh/RedisLabs/spark-redis/branch/master/graph/badge.svg)](https://codecov.io/gh/RedisLabs/spark-redis)-->
 
 # Spark-Redis
@@ -19,11 +20,11 @@ Spark-Redis also supports Spark Streaming (DStreams) and Structured Streaming.
 The library has several branches, each corresponds to a different supported Spark version. For example, 'branch-2.3' works with any Spark 2.3.x version.
 The master branch contains the recent development for the next release.
 
-| Spark-Redis | Spark         | Redis            | Supported Scala Versions | 
-| ----------- | ------------- | ---------------- | ------------------------ |
-| 2.4         | 2.4.x         | >=2.9.0          | 2.11                     | 
-| 2.3         | 2.3.x         | >=2.9.0          | 2.11                     | 
-| 1.4         | 1.4.x         |                  | 2.10                     | 
+| Spark-Redis                                                     | Spark         | Redis            | Supported Scala Versions | 
+| ----------------------------------------------------------------| ------------- | ---------------- | ------------------------ |
+| [2.4](https://github.com/RedisLabs/spark-redis/tree/branch-2.4) | 2.4.x         | >=2.9.0          | 2.11, 2.12               | 
+| [2.3](https://github.com/RedisLabs/spark-redis/tree/branch-2.3) | 2.3.x         | >=2.9.0          | 2.11                     | 
+| [1.4](https://github.com/RedisLabs/spark-redis/tree/branch-1.4) | 1.4.x         |                  | 2.10                     | 
 
 
 ## Known limitations
@@ -35,6 +36,8 @@ This library is a work in progress so the API may change before the official rel
 
 ## Documentation
 
+Please make sure you use documentation from the correct branch ([2.4](https://github.com/RedisLabs/spark-redis/tree/branch-2.4#documentation), [2.3](https://github.com/RedisLabs/spark-redis/tree/branch-2.3#documentation), etc). 
+
   - [Getting Started](doc/getting-started.md)
   - [RDD](doc/rdd.md)
   - [Dataframe](doc/dataframe.md)

dev/change-scala-version.sh

@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+
+set -e
+
+VALID_VERSIONS=( 2.11 2.12 )
+
+SCALA_211_MINOR_VERSION="12"
+SCALA_212_MINOR_VERSION="9"
+
+usage() {
+  echo "Usage: $(basename $0) [-h|--help] <version>
+where :
+  -h| --help Display this help text
+  valid version values : ${VALID_VERSIONS[*]}
+" 1>&2
+  exit 1
+}
+
+if [[ ($# -ne 1) || ( $1 == "--help") ||  $1 == "-h" ]]; then
+  usage
+fi
+
+TO_MAJOR_VERSION=$1
+
+check_scala_version() {
+  for i in ${VALID_VERSIONS[*]}; do [ $i = "$1" ] && return 0; done
+  echo "Invalid Scala version: $1. Valid versions: ${VALID_VERSIONS[*]}" 1>&2
+  exit 1
+}
+
+check_scala_version "$TO_MAJOR_VERSION"
+
+if [ $TO_MAJOR_VERSION = "2.12" ]; then
+  FROM_MAJOR_VERSION="2.11"
+  FROM_MINOR_VERSION=$SCALA_211_MINOR_VERSION
+  TO_MINOR_VERSION=$SCALA_212_MINOR_VERSION
+else
+  FROM_MAJOR_VERSION="2.12"
+  FROM_MINOR_VERSION=$SCALA_212_MINOR_VERSION
+  TO_MINOR_VERSION=$SCALA_211_MINOR_VERSION
+fi
+
+sed_i() {
+  sed -e "$1" "$2" > "$2.tmp" && mv "$2.tmp" "$2"
+}
+
+export -f sed_i
+
+# change <artifactId>
+BASEDIR=$(dirname $0)/..
+find "$BASEDIR" -name 'pom.xml' -not -path '*target*' -print \
+  -exec bash -c "sed_i 's/\(artifactId.*\)_'$FROM_MAJOR_VERSION'/\1_'$TO_MAJOR_VERSION'/g' {}" \;
+
+# change <scala.major.version>
+find "$BASEDIR" -name 'pom.xml' -not -path '*target*' -print \
+  -exec bash -c "sed_i 's/\(<scala.major.version>\)'$FROM_MAJOR_VERSION'/\1'$TO_MAJOR_VERSION'/g' {}" \;
+
+# change <scala.complete.version>
+find "$BASEDIR" -name 'pom.xml' -not -path '*target*' -print \
+  -exec bash -c "sed_i 's/\(<scala.complete.version>.*\.\)'$FROM_MINOR_VERSION'/\1'$TO_MINOR_VERSION'/g' {}" \;
+

doc/configuration.md

@@ -10,6 +10,7 @@ topology from the initial node, so there is no need to provide the rest of the c
 * `spark.redis.timeout` - connection timeout in ms, 2000 ms by default
 * `spark.redis.max.pipeline.size` - the maximum number of commands per pipeline (used to batch commands). The default value is 100.
 * `spark.redis.scan.count` - count option of SCAN command (used to iterate over keys). The default value is 100.
+* `spark.redis.ssl` - set to true to use tls
 
 
 

doc/dataframe.md

@@ -341,6 +341,11 @@ root
 | max.pipeline.size      | maximum number of commands per pipeline (used to batch commands)                          | `Int`                 | 100     |
 | scan.count             | count option of SCAN command (used to iterate over keys)                                  | `Int`                 | 100     |
 | iterator.grouping.size | the number of items to be grouped when iterating over underlying RDD partition            | `Int`                 | 1000    |
+| host                   | overrides `spark.redis.host` configured in SparkSession (if set, any other connection setting from SparkSession is ignored )                        | `String`              | `localhost`    |
+| port                   | overrides `spark.redis.port` configured in SparkSession (if set, any other connection setting from SparkSession is ignored )                        | `Int`                 | `6379`         |
+| auth                   | overrides `spark.redis.auth` configured in SparkSession (if set, any other connection setting from SparkSession is ignored )                        | `String`              | -              |
+| dbNum                  | overrides `spark.redis.db` configured in SparkSession (if set, any other connection setting from SparkSession is ignored )                       | `Int`                 | `0`            |
+| timeout                | overrides `spark.redis.timeout` configured in SparkSession (if set, any other connection setting from SparkSession is ignored )                     | `Int`                 | `2000`         |
 
 
 ## Known limitations

doc/dev.md

@@ -25,3 +25,12 @@ To build Spark-Redis skipping tests, run:
 ```
 mvn clean package -DskipTests
 ```
+
+To change scala version use `./dev/change-scala-version.sh` script. It will change scala version in `pom.xml`. For example:
+```
+./dev/change-scala-version.sh 2.12
+```
+
+```
+./dev/change-scala-version.sh 2.11
+```

doc/getting-started.md

@@ -6,12 +6,30 @@
   <dependencies>
     <dependency>
       <groupId>com.redislabs</groupId>
-      <artifactId>spark-redis</artifactId>
-      <version>2.4.0</version>
+      <artifactId>spark-redis_2.11</artifactId>
+      <version>2.4.2</version>
     </dependency>
   </dependencies>
 ```
 
+Or
+
+```xml
+  <dependencies>
+    <dependency>
+      <groupId>com.redislabs</groupId>
+      <artifactId>spark-redis_2.12</artifactId>
+      <version>2.4.2</version>
+    </dependency>
+  </dependencies>
+```
+
+### SBT
+
+```scala
+libraryDependencies += "com.redislabs" %% "spark-redis" % "2.4.2"
+```
+
 ### Build form source
 You can download the library's source and build it:
 ```

doc/rdd.md

@@ -1,3 +1,4 @@
+
 # RDD
 
   - [The keys RDD](#the-keys-rdd)
@@ -114,12 +115,35 @@ For String values, your RDD should consist of the key-value pairs that are to be
 sc.toRedisKV(stringRDD)
 ```
 
+In order to set an expiry on the key, we can pass in the `ttl` (in seconds) as an additional argument:
+
+```scala
+sc.toRedisKV(stringRDD, ttl)
+```
+
+By default, Strings won't have any expiry set.
+
 #### Hashes
 To store a Redis Hash, the RDD should consist of its field-value pairs. If the RDD is called `hashRDD`, the following should be used for storing it in the key name specified by `hashName`:
 
 ```scala
 sc.toRedisHASH(hashRDD, hashName)
 ```
+In order to set an expiry on the key, we can pass in the `ttl` (in seconds) as an additional argument:
+
+```scala
+sc.toRedisHASH(hashRDD, hashName, ttl)
+```
+
+By default, Hashes won't have any expiry set.
+
+Use the following to store an RDD into multiple hashs:
+
+```scala
+sc.toRedisHASHes(hashRDD, ttl)
+```
+
+The `hashRDD` is a rdd of tuples (`hashname`, `map[field name, field value]`)
 
 #### Lists
 Use the following to store an RDD in a Redis List:
@@ -137,6 +161,31 @@ sc.toRedisFixedLIST(listRDD, listName, listSize)
 The `listRDD` is an RDD that contains all of the list's string elements in order, and `listName` is the list's key name.
 `listSize` is an integer which specifies the size of the Redis list; it is optional, and will default to an unlimited size.
 
+Use the following to store an RDD in multiple Redis Lists:
+
+```scala
+sc.toRedisLISTs(rdd)
+```
+
+The `rdd` is an RDD of tuples (`list name`, `list values`)
+
+Use the following to store an RDD of binary values in multiple Redis Lists:
+
+```scala
+sc.toRedisByteLISTs(byteListRDD)
+```
+
+The `byteListRDD` is an RDD of tuples (`list name`, `list values`) represented as byte arrays. 
+
+Expiry can be set on Lists by passing in an additional argument called `ttl` (in seconds) to the above methods except `toRedisFixedLIST`:
+```scala
+sc.toRedisLIST(listRDD, listName, ttl)
+sc.toRedisLISTs(rdd, ttl)
+sc.toRedisByteLISTs(byteListRDD, ttl)
+```
+
+By default, Lists won't have any expiry set.
+
 #### Sets
 For storing data in a Redis Set, use `toRedisSET` as follows:
 
@@ -146,13 +195,29 @@ sc.toRedisSET(setRDD, setName)
 
 Where `setRDD` is an RDD with the set's string elements and `setName` is the name of the key for that set.
 
+In order to set an expiry on the key, we can pass in the `ttl` (in seconds) as an additional argument:
+
+```scala
+sc.toRedisSET(setRDD, setName, ttl)
+```
+
+By default, Sets won't have any expiry set.
+
 #### Sorted Sets
 ```scala
 sc.toRedisZSET(zsetRDD, zsetName)
 ```
 
 The above example demonstrates storing data in Redis in a Sorted Set. The `zsetRDD` in the example should contain pairs of members and their scores, whereas `zsetName` is the name for that key.
 
+In order to set an expiry on the key, we can pass in the `ttl` (in seconds) as an additional argument:
+
+```scala
+sc.toRedisZSET(zsetRDD, zsetName, ttl)
+```
+
+By default, Sorted Sets won't have any expiry set.
+
 ### Read and write configuration options
 
 Some [configuration options](configuration.md) can be overridden for a particular RDD:

doc/streaming.md

@@ -119,6 +119,7 @@ import com.redislabs.provider.redis.streaming._
 val ssc = new StreamingContext(sc, Seconds(1))
 val redisStream = ssc.createRedisStream(Array("foo", "bar"), storageLevel = StorageLevel.MEMORY_AND_DISK_2)
 redisStream.print()
+ssc.start()
 ssc.awaitTermination()
 ```
 
@@ -132,5 +133,6 @@ import com.redislabs.provider.redis.streaming._
 val ssc = new StreamingContext(sc, Seconds(1))
 val redisStream = ssc.createRedisStreamWithoutListname(Array("foo", "bar"), storageLevel = StorageLevel.MEMORY_AND_DISK_2)
 redisStream.print()
+ssc.start()
 ssc.awaitTermination()
 ```