{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/12750","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/12750/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/12750/comments","events_url":"https://api.github.com/repos/apache/spark/issues/12750/events","html_url":"https://github.com/apache/spark/pull/12750","id":151543244,"number":12750,"title":"[SPARK-14972] Improve performance of JSON schema inference's compatibleType method","user":{"login":"JoshRosen","id":50748,"avatar_url":"https://avatars.githubusercontent.com/u/50748?v=3","gravatar_id":"","url":"https://api.github.com/users/JoshRosen","html_url":"https://github.com/JoshRosen","followers_url":"https://api.github.com/users/JoshRosen/followers","following_url":"https://api.github.com/users/JoshRosen/following{/other_user}","gists_url":"https://api.github.com/users/JoshRosen/gists{/gist_id}","starred_url":"https://api.github.com/users/JoshRosen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JoshRosen/subscriptions","organizations_url":"https://api.github.com/users/JoshRosen/orgs","repos_url":"https://api.github.com/users/JoshRosen/repos","events_url":"https://api.github.com/users/JoshRosen/events{/privacy}","received_events_url":"https://api.github.com/users/JoshRosen/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":8,"created_at":"2016-04-28T03:48:11Z","updated_at":"2016-04-30T07:07:36Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/12750","html_url":"https://github.com/apache/spark/pull/12750","diff_url":"https://github.com/apache/spark/pull/12750.diff","patch_url":"https://github.com/apache/spark/pull/12750.patch"},"body":"This patch improves the performance of `InferSchema.compatibleType` and `inferField`. The net result of this patch is a 2x-4x speedup in local benchmarks running against cached data with a massive nested schema.\r\n\r\nThe key idea is to remove unnecessary sorting in `compatibleType`'s `StructType` merging code. This code takes two structs, merges the fields with matching names, and copies over the unique fields, producing a new schema which is the union of the two structs' schemas. Previously, this code performed a very inefficient `groupBy()` to match up fields with the same name, but this is unnecessary because `inferField` already sorts structs' fields by name: since both lists of fields are sorted, we can simply merge them in a single pass.\r\n\r\nThis patch also speeds up the existing field sorting in `inferField`: the old sorting code allocated unnecessary intermediate collections, while the new code uses mutable collects and performs in-place sorting.\r\n\r\nFinally, I replaced a `treeAggregate` call with `fold`: I doubt that `treeAggregate` will benefit us very much because the schemas would have to be enormous to realize large savings in network traffic. Since most schemas are probably fairly small in serialized form, they should typically fit within a direct task result and therefore can be incrementally merged at the driver as individual tasks finish. This change eliminates an entire (short) scheduler stage. "},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/215943176","html_url":"https://github.com/apache/spark/pull/12750#issuecomment-215943176","issue_url":"https://api.github.com/repos/apache/spark/issues/12750","id":215943176,"user":{"login":"JoshRosen","id":50748,"avatar_url":"https://avatars.githubusercontent.com/u/50748?v=3","gravatar_id":"","url":"https://api.github.com/users/JoshRosen","html_url":"https://github.com/JoshRosen","followers_url":"https://api.github.com/users/JoshRosen/followers","following_url":"https://api.github.com/users/JoshRosen/following{/other_user}","gists_url":"https://api.github.com/users/JoshRosen/gists{/gist_id}","starred_url":"https://api.github.com/users/JoshRosen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JoshRosen/subscriptions","organizations_url":"https://api.github.com/users/JoshRosen/orgs","repos_url":"https://api.github.com/users/JoshRosen/repos","events_url":"https://api.github.com/users/JoshRosen/events{/privacy}","received_events_url":"https://api.github.com/users/JoshRosen/received_events","type":"User","site_admin":false},"created_at":"2016-04-30T07:07:36Z","updated_at":"2016-04-30T07:07:36Z","body":"After a bit more time in a profiler, I was able to make this 2x faster than my previous best time.\r\n\r\nTo give a rough idea of the structure of my benchmark:\r\n\r\nI have a folder containing a bunch of gigantic JSON documents with huge, deeply-nested schemas. Imagine the schema being the union of a huge number of smaller schemas, so it's pretty big and sparse.\r\n\r\nMy benchmark harness reads in the raw JSON, caches it, unions it together a bunch of times to inflate the benchmark runtime (to avoid small constant factor noise), then coalesces it to a single partition to avoid task-launch overheads (and so we measure single-core performance):\r\n\r\n```scala\r\nval lines = spark.read.text(\"...\").rdd\r\nlines.cache.count()\r\nval coalescedLines =  (lines ++ lines ++ lines ++ lines ++ lines).coalesce(1)\r\n\r\nfor (i <- 1 to 10) {\r\n  val startTime = System.currentTimeMillis\r\n  spark.read.json(coalescedLines)\r\n  println(s\"Took ${System.currentTimeMillis - startTime}\")\r\n}\r\n```\r\n\r\nBefore (f5da592fc63b8d3bc09d49c196d6c5d98cd2a013), this took around 88 seconds.\r\n\r\nAfter (6bf5ee634ee7e509a88e64236f18e3c6e7a07aa2), this takes about **14.5 seconds**.\r\n\r\nCoupled with the changes in #12741, a change that led to a massive speedup in the `reduce` / `treeAggregate` step, this should be a huge speedup vs 1.6.x."}}