{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/14102","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/14102/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/14102/comments","events_url":"https://api.github.com/repos/apache/spark/issues/14102/events","html_url":"https://github.com/apache/spark/pull/14102","id":164440824,"number":14102,"title":"[SPARK-16434][SQL] Avoid per-record type dispatch in JSON when reading","user":{"login":"HyukjinKwon","id":6477701,"avatar_url":"https://avatars.githubusercontent.com/u/6477701?v=3","gravatar_id":"","url":"https://api.github.com/users/HyukjinKwon","html_url":"https://github.com/HyukjinKwon","followers_url":"https://api.github.com/users/HyukjinKwon/followers","following_url":"https://api.github.com/users/HyukjinKwon/following{/other_user}","gists_url":"https://api.github.com/users/HyukjinKwon/gists{/gist_id}","starred_url":"https://api.github.com/users/HyukjinKwon/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HyukjinKwon/subscriptions","organizations_url":"https://api.github.com/users/HyukjinKwon/orgs","repos_url":"https://api.github.com/users/HyukjinKwon/repos","events_url":"https://api.github.com/users/HyukjinKwon/events{/privacy}","received_events_url":"https://api.github.com/users/HyukjinKwon/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":19,"created_at":"2016-07-08T01:15:14Z","updated_at":"2016-08-11T02:47:43Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/14102","html_url":"https://github.com/apache/spark/pull/14102","diff_url":"https://github.com/apache/spark/pull/14102.diff","patch_url":"https://github.com/apache/spark/pull/14102.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nCurrently, `JacksonParser.parse` is doing type-based dispatch for each row to convert the tokens to appropriate values for Spark.\r\nIt might not have to be done like this because the schema is already kept.\r\n\r\nSo, appropriate converters can be created first according to the schema once, and then apply them to each row. \r\n\r\nThis PR corrects `JacksonParser` so that it creates all converters for the schema once and then applies them to each row rather than type dispatching for every row.\r\n\r\nBenchmark was proceeded with the codes below:\r\n\r\n#### Parser tests\r\n\r\n**Before**\r\n\r\n```scala\r\ntest(\"Benchmark for JSON converter\") {\r\n  val N = 500 << 8\r\n  val row =\r\n    \"\"\"{\"struct\":{\"field1\": true, \"field2\": 92233720368547758070},\r\n    \"structWithArrayFields\":{\"field1\":[4, 5, 6], \"field2\":[\"str1\", \"str2\"]},\r\n    \"arrayOfString\":[\"str1\", \"str2\"],\r\n    \"arrayOfInteger\":[1, 2147483647, -2147483648],\r\n    \"arrayOfLong\":[21474836470, 9223372036854775807, -9223372036854775808],\r\n    \"arrayOfBigInteger\":[922337203685477580700, -922337203685477580800],\r\n    \"arrayOfDouble\":[1.2, 1.7976931348623157E308, 4.9E-324, 2.2250738585072014E-308],\r\n    \"arrayOfBoolean\":[true, false, true],\r\n    \"arrayOfNull\":[null, null, null, null],\r\n    \"arrayOfStruct\":[{\"field1\": true, \"field2\": \"str1\"}, {\"field1\": false}, {\"field3\": null}],\r\n    \"arrayOfArray1\":[[1, 2, 3], [\"str1\", \"str2\"]],\r\n    \"arrayOfArray2\":[[1, 2, 3], [1.1, 2.1, 3.1]]\r\n   }\"\"\"\r\n  val data = List.fill(N)(row)\r\n  val dummyOption = new JSONOptions(Map.empty[String, String])\r\n  val schema =\r\n    InferSchema.infer(spark.sparkContext.parallelize(Seq(row)), \"\", dummyOption)\r\n  val factory = new JsonFactory()\r\n\r\n  val benchmark = new Benchmark(\"JSON converter\", N)\r\n  benchmark.addCase(\"convert JSON file\", 10) { _ =>\r\n    data.foreach { input =>\r\n      val parser = factory.createParser(input)\r\n      parser.nextToken()\r\n      JacksonParser.convertRootField(factory, parser, schema)\r\n    }\r\n  }\r\n  benchmark.run()\r\n}\r\n```\r\n\r\n```\r\nJSON converter:                          Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nconvert JSON file                             1697 / 1807          0.1       13256.9       1.0X\r\n```\r\n\r\n\r\n**After**\r\n\r\n```scala\r\ntest(\"Benchmark for JSON converter\") {\r\n  val N = 500 << 8\r\n  val row =\r\n    \"\"\"{\"struct\":{\"field1\": true, \"field2\": 92233720368547758070},\r\n    \"structWithArrayFields\":{\"field1\":[4, 5, 6], \"field2\":[\"str1\", \"str2\"]},\r\n    \"arrayOfString\":[\"str1\", \"str2\"],\r\n    \"arrayOfInteger\":[1, 2147483647, -2147483648],\r\n    \"arrayOfLong\":[21474836470, 9223372036854775807, -9223372036854775808],\r\n    \"arrayOfBigInteger\":[922337203685477580700, -922337203685477580800],\r\n    \"arrayOfDouble\":[1.2, 1.7976931348623157E308, 4.9E-324, 2.2250738585072014E-308],\r\n    \"arrayOfBoolean\":[true, false, true],\r\n    \"arrayOfNull\":[null, null, null, null],\r\n    \"arrayOfStruct\":[{\"field1\": true, \"field2\": \"str1\"}, {\"field1\": false}, {\"field3\": null}],\r\n    \"arrayOfArray1\":[[1, 2, 3], [\"str1\", \"str2\"]],\r\n    \"arrayOfArray2\":[[1, 2, 3], [1.1, 2.1, 3.1]]\r\n   }\"\"\"\r\n  val data = List.fill(N)(row)\r\n  val dummyOption = new JSONOptions(Map.empty[String, String], new SQLConf())\r\n  val schema =\r\n    InferSchema.infer(spark.sparkContext.parallelize(Seq(row)), dummyOption)\r\n\r\n  val benchmark = new Benchmark(\"JSON converter\", N)\r\n  benchmark.addCase(\"convert JSON file\", 10) { _ =>\r\n    val parser = new JacksonParser(schema, dummyOption)\r\n    data.foreach { input =>\r\n      parser.parse(input)\r\n    }\r\n  }\r\n  benchmark.run()\r\n}\r\n```\r\n\r\n```\r\nJSON converter:                          Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nconvert JSON file                             1401 / 1461          0.1       10947.4       1.0X\r\n```\r\n\r\nIt seems parsing time is improved by roughly ~20%\r\n\r\n\r\n#### End-to-End test\r\n\r\n```scala\r\ntest(\"Benchmark for JSON reader\") {\r\n  val N = 500 << 8\r\n  val row =\r\n    \"\"\"{\"struct\":{\"field1\": true, \"field2\": 92233720368547758070},\r\n    \"structWithArrayFields\":{\"field1\":[4, 5, 6], \"field2\":[\"str1\", \"str2\"]},\r\n    \"arrayOfString\":[\"str1\", \"str2\"],\r\n    \"arrayOfInteger\":[1, 2147483647, -2147483648],\r\n    \"arrayOfLong\":[21474836470, 9223372036854775807, -9223372036854775808],\r\n    \"arrayOfBigInteger\":[922337203685477580700, -922337203685477580800],\r\n    \"arrayOfDouble\":[1.2, 1.7976931348623157E308, 4.9E-324, 2.2250738585072014E-308],\r\n    \"arrayOfBoolean\":[true, false, true],\r\n    \"arrayOfNull\":[null, null, null, null],\r\n    \"arrayOfStruct\":[{\"field1\": true, \"field2\": \"str1\"}, {\"field1\": false}, {\"field3\": null}],\r\n    \"arrayOfArray1\":[[1, 2, 3], [\"str1\", \"str2\"]],\r\n    \"arrayOfArray2\":[[1, 2, 3], [1.1, 2.1, 3.1]]\r\n   }\"\"\"\r\n  val df = spark.sqlContext.read.json(spark.sparkContext.parallelize(List.fill(N)(row)))\r\n  withTempPath { path =>\r\n    df.write.format(\"json\").save(path.getCanonicalPath)\r\n\r\n    val benchmark = new Benchmark(\"JSON reader\", N)\r\n    benchmark.addCase(\"reading JSON file\", 10) { _ =>\r\n      spark.read.format(\"json\").load(path.getCanonicalPath).collect()\r\n    }\r\n    benchmark.run()\r\n  }\r\n}\r\n```\r\n\r\n**Before**\r\n\r\n```\r\nJSON reader:                             Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nreading JSON file                             6485 / 6924          0.0       50665.0       1.0X\r\n```\r\n\r\n**After**\r\n\r\n```\r\nJSON reader:                             Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nreading JSON file                             6350 / 6529          0.0       49609.3       1.0X\r\n```\r\n\r\n\r\n## How was this patch tested?\r\n\r\nExisting test cases should cover this.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/239060738","html_url":"https://github.com/apache/spark/pull/14102#issuecomment-239060738","issue_url":"https://api.github.com/repos/apache/spark/issues/14102","id":239060738,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars.githubusercontent.com/u/8108735?v=3","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2016-08-11T02:47:42Z","updated_at":"2016-08-11T02:47:42Z","body":"**[Test build #63575 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/63575/consoleFull)** for PR 14102 at commit [`bceda7b`](https://github.com/apache/spark/commit/bceda7ba4f06c0b6fd99f11ef2662f9f3a154af0)."}}
{"actor":{"display_login":"SparkQA"}}