Getting java.lang.IllegalArgumentException: The byte[] must not be null

[mccstan]

When querying a non existing key in redis, using DataFrame API, I got a non emty DataFrame that throw this exception : java.lang.IllegalArgumentException: The byte[] must not be null

Thanks for your HELP

[fe2s]

Hi @mccstan,

Can you please provide more details?

I'm trying to reproduce it with the following:

test("read dataframe from non-existing keys") {
    val loadedDf = spark.read
      .format("org.apache.spark.sql.redis")
      .schema(
        StructType(Array(
          StructField("id", IntegerType)
        ))
      )
      .option("keys.pattern", "some-random-string:*")
      .load()

    println(s"count ${loadedDf.count()}")
    println(loadedDf.show())
}

and getting the following output without exceptions:

count 0
+---+
| id|
+---+
+---+

[mccstan]

Hi @fe2s thank you for your answer, when qurying with pattern, no issue is reported !
The issue happen when querying on specific key (this is needed for some use cases)

[fe2s]

Hi @mccstan,

I'm still not able to reproduce the exception. Could you please share your code?
I'm trying the following:

 val loadedDf = spark.read
      .format("org.apache.spark.sql.redis")
      .schema(
        StructType(Array(
          StructField("id", IntegerType)
        ))
      )
      .option("keys.pattern", "some-random-string")
      .load()


    println(s"count ${loadedDf.count()}")
    println(loadedDf.show())
    println(loadedDf.head())

and getting

count 1
+----+
|  id|
+----+
|null|
+----+

()
[null]

[mccstan]

Hi @fe2s please find here an example that can reproduce the issue !
Thanks !

package redis.sparkRedis

import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}
import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.types.StructType

object SparkRedisIssue  extends App{

  val redisServerDnsAddress = "localhost"
  val redisPortNumber = 7000

  val spark = SparkSession
    .builder()
    .master("local[*]")
    .appName("Testing-local")
    .config("spark.redis.host", redisServerDnsAddress)
    .config("spark.redis.port", redisPortNumber)
    .getOrCreate()

  import spark.implicits._

  case class Address (street : Int, road : String, postalCode : Int)
  case class User(id : String, firstname : String, lastname : String, addresses: List[Address])

  val NAMESPACE: String = "USERS:CACHE"

  val schema = ScalaReflection.schemaFor[User].dataType.asInstanceOf[StructType]


  val usersDataset = spark.createDataset(Seq(
    User("user01", "firstname01", "lastname01", List(Address(54, "Redis road", 97000)))
  ))

  usersDataset.write
    .format("org.apache.spark.sql.redis")
    .option("table", s"$NAMESPACE")
    .option("key.column", "id")
    .option("model", "binary")
    .option("ttl", 60)
    .mode(SaveMode.Overwrite)
    .save()

  val retrievedUsers: Dataset[User] = spark.read
    .format("org.apache.spark.sql.redis")
    .schema(schema)
    .option("model", "binary")
    .option("keys.pattern", s"$NAMESPACE:undefined01")
    .option("key.column", "id")
    .load()
    .as[User]

  retrievedUsers.show(false)
}

Then i am getting this error :

java.lang.IllegalArgumentException: The byte[] must not be null
	at org.apache.commons.lang3.SerializationUtils.deserialize(SerializationUtils.java:263) ~[commons-lang3-3.5.jar:3.5]
	at org.apache.spark.sql.redis.BinaryRedisPersistence.decodeRow(BinaryRedisPersistence.scala:36) ~[spark-redis-2.4.0.jar:na]
	at org.apache.spark.sql.redis.BinaryRedisPersistence.decodeRow(BinaryRedisPersistence.scala:14) ~[spark-redis-2.4.0.jar:na]
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1$$anonfun$apply$9.apply(RedisSourceRelation.scala:295) ~[spark-redis-2.4.0.jar:na]
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1$$anonfun$apply$9.apply(RedisSourceRelation.scala:293) ~[spark-redis-2.4.0.jar:na]
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) ~[scala-library-2.11.8.jar:na]
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) ~[scala-library-2.11.8.jar:na]
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) ~[scala-library-2.11.8.jar:na]
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) ~[scala-library-2.11.8.jar:na]
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234) ~[scala-library-2.11.8.jar:na]
	at scala.collection.AbstractTraversable.map(Traversable.scala:104) ~[scala-library-2.11.8.jar:na]
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1.apply(RedisSourceRelation.scala:293) ~[spark-redis-2.4.0.jar:na]
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1.apply(RedisSourceRelation.scala:280) ~[spark-redis-2.4.0.jar:na]
	at com.redislabs.provider.redis.util.ConnectionUtils$.withConnection(ConnectionUtils.scala:20) ~[spark-redis-2.4.0.jar:na]
	at org.apache.spark.sql.redis.RedisSourceRelation.org$apache$spark$sql$redis$RedisSourceRelation$$scanRows(RedisSourceRelation.scala:280) ~[spark-redis-2.4.0.jar:na]
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$buildScan$3$$anonfun$apply$7$$anonfun$apply$8.apply(RedisSourceRelation.scala:182) ~[spark-redis-2.4.0.jar:na]
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$buildScan$3$$anonfun$apply$7$$anonfun$apply$8.apply(RedisSourceRelation.scala:181) ~[spark-redis-2.4.0.jar:na]
	at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434) ~[scala-library-2.11.8.jar:na]
	at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440) ~[scala-library-2.11.8.jar:na]
	at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:438) ~[scala-library-2.11.8.jar:na]
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) ~[scala-library-2.11.8.jar:na]
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) ~[na:na]
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) ~[spark-sql_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636) ~[spark-sql_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255) ~[spark-sql_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247) ~[spark-sql_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.scheduler.Task.run(Task.scala:121) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414) ~[spark-core_2.11-2.4.3.jar:2.4.3]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_201]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_201]
	at java.lang.Thread.run(Thread.java:748) [na:1.8.0_201]
2019-09-09 23:17:26.627 [WARN ] o.a.s.s.TaskSetManager - Lost task 0.0 in stage 3.0 (TID 5, localhost, executor driver): java.lang.IllegalArgumentException: The byte[] must not be null
	at org.apache.commons.lang3.SerializationUtils.deserialize(SerializationUtils.java:263)
	at org.apache.spark.sql.redis.BinaryRedisPersistence.decodeRow(BinaryRedisPersistence.scala:36)
	at org.apache.spark.sql.redis.BinaryRedisPersistence.decodeRow(BinaryRedisPersistence.scala:14)
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1$$anonfun$apply$9.apply(RedisSourceRelation.scala:295)
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1$$anonfun$apply$9.apply(RedisSourceRelation.scala:293)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.AbstractTraversable.map(Traversable.scala:104)
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1.apply(RedisSourceRelation.scala:293)
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$org$apache$spark$sql$redis$RedisSourceRelation$$scanRows$1.apply(RedisSourceRelation.scala:280)
	at com.redislabs.provider.redis.util.ConnectionUtils$.withConnection(ConnectionUtils.scala:20)
	at org.apache.spark.sql.redis.RedisSourceRelation.org$apache$spark$sql$redis$RedisSourceRelation$$scanRows(RedisSourceRelation.scala:280)
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$buildScan$3$$anonfun$apply$7$$anonfun$apply$8.apply(RedisSourceRelation.scala:182)
	at org.apache.spark.sql.redis.RedisSourceRelation$$anonfun$buildScan$3$$anonfun$apply$7$$anonfun$apply$8.apply(RedisSourceRelation.scala:181)
	at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434)
	at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
	at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:438)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:121)
	at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

[mccstan]

Hi !
I was able to avoid the issue by adding * at the end of the key

option("keys.pattern", s"$NAMESPACE:undefined01*")

[fe2s]

Okay, I see what is the difference with my code. You use binary persistence model.

Another option to avoid the exception is to not use binary model and use hash one (default setting).

I will fix it in the next release.

[mccstan]

Yes it’s the binary model !
I can’t use the hash model because the model I use is a little bit complex and is not flat !

Thanks for your help, i will wait for the release or can help !