[jira] [Commented] (KAFKA-5328) consider switching json parser from scala to jackson

[Onur Karaman (JIRA)]

    [ 
https://issues.apache.org/jira/browse/KAFKA-5328?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16027086#comment-16027086
 ] 

Onur Karaman commented on KAFKA-5328:
-------------------------------------

I copied the topic assignment znodes and partition state znodes from a cluster 
into two text files and just timed how long it took to parse the topic and 
partition znodes using both the scala json parser and jackson.

This experiment had 3066 topics and 95895 partitions (the actual cluster I got 
the znodes from was actually a third the size but I just duplicated lines to 
reflect some of our other clusters).

Here's the code:
{code}
/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package kafka.tools

import java.nio.file.{Files, Paths}

import com.fasterxml.jackson.databind.ObjectMapper
import kafka.api.LeaderAndIsr
import kafka.common.TopicAndPartition
import kafka.controller.LeaderIsrAndControllerEpoch
import kafka.utils.Json

import scala.collection.JavaConverters._
import scala.collection.Seq

object JsonPerformance {
  def main(args: Array[String]): Unit = {
    val assignments = 
Files.readAllLines(Paths.get("/Users/okaraman/code/read-zk-state/raw_assignments_big.txt"))
    val states = 
Files.readAllLines(Paths.get("/Users/okaraman/code/read-zk-state/raw_states_big.txt"))

    time(assignments.asScala.foreach(parseAssignmentWithScala), "scala json 
parser assignments")
    time(assignments.asScala.foreach(parseAssignmentWithJackson), "jackson json 
parser assignments")
    time(states.asScala.foreach(parseStateWithScala), "scala json parser 
states")
    time(states.asScala.foreach(parseStateWithJackson), "jackson json parser 
states")
  }

  def time(f: => Unit, messagePrefix: String): Unit = {
    val start = System.currentTimeMillis()
    f
    val end = System.currentTimeMillis()
    println(messagePrefix + s" took: ${end - start} ms")
  }

  def parseAssignmentWithScala(assignment: String): Map[TopicAndPartition, 
Seq[Int]] = {
    val json = Json.parseFull(assignment).get
    val jsonPartitions = json.asInstanceOf[Map[String, 
Any]].get("partitions").get
    val replicaMap = jsonPartitions.asInstanceOf[Map[String, Seq[Int]]]
    replicaMap.map { case (partition, replicas) => TopicAndPartition("t", 
partition.toInt) -> replicas }
  }

  def parseAssignmentWithJackson(assignment: String): Map[TopicAndPartition, 
Seq[Int]] = {
    val mapper = new ObjectMapper()
    val json = mapper.readTree(assignment)
    json.get("partitions").asScala.toList.flatMap(x => (0 until x.size()).map(i 
=> TopicAndPartition("t", i) -> x.get(i).asScala.toSeq.map(_.asInt()))).toMap
  }

  def parseStateWithScala(state: String): LeaderIsrAndControllerEpoch = {
    val json = Json.parseFull(state).get
    val leaderIsrAndEpochInfo = json.asInstanceOf[Map[String, Any]]
    val leader = leaderIsrAndEpochInfo.get("leader").get.asInstanceOf[Int]
    val epoch = leaderIsrAndEpochInfo.get("leader_epoch").get.asInstanceOf[Int]
    val isr = leaderIsrAndEpochInfo.get("isr").get.asInstanceOf[List[Int]]
    val controllerEpoch = 
leaderIsrAndEpochInfo.get("controller_epoch").get.asInstanceOf[Int]
    LeaderIsrAndControllerEpoch(LeaderAndIsr(leader, epoch, isr, 1), 
controllerEpoch)
  }

  def parseStateWithJackson(state: String): LeaderIsrAndControllerEpoch = {
    val mapper = new ObjectMapper()
    val json = mapper.readTree(state)
    val leader = json.get("leader").asInt()
    val epoch = json.get("leader_epoch").asInt()
    val isr = json.get("isr").asScala.toList.map(_.asInt())
    val controllerEpoch = json.get("controller_epoch").asInt()
    LeaderIsrAndControllerEpoch(LeaderAndIsr(leader, epoch, isr, 1), 
controllerEpoch)
  }
}
{code}

Here's the output:
{code}
scala json parser assignments took: 8730 ms
jackson json parser assignments took: 354 ms
scala json parser states took: 28395 ms
jackson json parser states took: 360 ms
{code}

So controller initialization time spent on json parsing would be reduced from 
37.1 seconds down to 0.7 seconds.

> consider switching json parser from scala to jackson
> ----------------------------------------------------
>
>                 Key: KAFKA-5328
>                 URL: https://issues.apache.org/jira/browse/KAFKA-5328
>             Project: Kafka
>          Issue Type: Sub-task
>            Reporter: Onur Karaman
>            Assignee: Onur Karaman
>
> The scala json parser is significantly slower than jackson.
> This can have a nontrivial impact on controller initialization since the 
> controller loads and parses almost all zookeeper state.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)