Some performance suggestions for Spark

[koeninger]

From reading

http://yahooeng.tumblr.com/post/135321837876/benchmarking-streaming-computation-engines-at

First, it appears you had fewer Kafka partitions than you had worker nodes. Because there's a 1:1 relationship between spark partitions and kafka partitions, you're not going to get full utilization of your workers. As you noted, shuffling can be expensive, you're better off doing the partitioning at the time you produce into kafka (ie add more kafka partitions)

Second, groupByKey is almost always a bad idea in Spark. You want e.g. reduceByKey to get aggregation work done before the shuffle (this is similar to a hadoop combiner).

Finally, it's not clear from the code whether the json and redis code paths are identical across the different benchmarks. For instance, there's a comment in the code that redis caching is not being done in the Spark case, but is in the Storm case. It doesn't seem like a fair comparison can be made without controlling for other variables.

If you want assistance with anything, let me know.

[granturing]

Hi Cody, I was just working on this in fact as I noticed the same thing about groupByKey and a few other things. I also made a small change to the output in my PR #6 which removed 1 stage from the job.

[amimimor]

I would argue that the #queryRedisTopLevel method and its downstream calls need a couple of improvements:

1. since it creates a new Jedis Pool per partition, but also this is called per batch interval, i.e. every second. Seems more reasonable to use a singleton connection pool, such as in http://stackoverflow.com/a/30593303
2. The HashMap built using the Redis results (why read from Redis per batch interval ?) could and should be initialized in the driver (if no refresh is needed) and then broadcasted (you could broadcast objects of up to ~1GB) or RDD'ed to be later joined

[compae]

In Stratio we have one example for the Redis writes in Spark Streaming:

https://github.com/Stratio/Sparkta/blob/master/plugins/output-redis/src/main/scala/com/stratio/sparkta/plugin/output/redis/RedisOutput.scala

With the DAO object you have one connection por executor:

https://github.com/Stratio/Sparkta/blob/master/plugins/output-redis/src/main/scala/com/stratio/sparkta/plugin/output/redis/dao/AbstractRedisDAO.scala

The SDK:
https://github.com/Stratio/Sparkta/blob/master/sdk/src/main/scala/com/stratio/sparkta/sdk/Output.scala

The function on the SDK:

protected def persistMetricOperation(stream: DStream[(DimensionValuesTime, Map[String, Option[Any]])]): Unit =
          stream.foreachRDD(rdd => {
            if (rdd.take(1).length > 0) {
              rdd.foreachPartition(
                ops => {
                  Try(upsert(ops)) match {
                    case Success(value) => value
                    case Failure(exception) => {
                      val error = s"Failure[Output]: ${ops.toString} | Message: ${exception.getLocalizedMessage}"
                      log.error(error, exception)
                    }
                  }
                }
              )
            } else log.info("Empty event received")
          })