{"push_id":2835541073,"size":9,"distinct_size":9,"ref":"refs/heads/master","head":"20b7c684cc4a8136b9a9c56390a4948de04e7c34","before":"103854028e99846aabeb6f27eb6fd255ecc96381","commits":[{"sha":"bbbf8146916aa70d9774543643776eed9d9d9373","author":{"name":"Bo Meng","email":"111616690713bef657d310664219beff1688ad21@hotmail.com"},"message":"[SPARK-22357][CORE] SparkContext.binaryFiles ignore minPartitions parameter\n\n## What changes were proposed in this pull request?\nFix the issue that minPartitions was not used in the method. This is a simple fix and I am not trying to make it complicated. The purpose is to still allow user to control the defaultParallelism through the value of minPartitions, while also via sc.defaultParallelism parameters.\n\n## How was this patch tested?\nI have not provided the additional test since the fix is very straightforward.\n\nCloses #21638 from bomeng/22357.\n\nLead-authored-by: Bo Meng <mengbo@hotmail.com>\nCo-authored-by: Bo Meng <bo.meng@jd.com>\nSigned-off-by: Sean Owen <sean.owen@databricks.com>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/bbbf8146916aa70d9774543643776eed9d9d9373"},{"sha":"32c8a3d7beac4b47a75f5ec3c69b13ebc57de0c7","author":{"name":"Marco Gaido","email":"368e8ff04b169d99d4d85a32ca87c1aa4ab0fa7e@gmail.com"},"message":"[MINOR] Avoid code duplication for nullable in Higher Order function\n\n## What changes were proposed in this pull request?\n\nMost of  `HigherOrderFunction`s have the same `nullable` definition, ie. they are nullable when one of their arguments is nullable. The PR refactors it in order to avoid code duplication.\n\n## How was this patch tested?\n\nNA\n\nCloses #22243 from mgaido91/MINOR_nullable_hof.\n\nAuthored-by: Marco Gaido <marcogaido91@gmail.com>\nSigned-off-by: hyukjinkwon <gurwls223@apache.org>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/32c8a3d7beac4b47a75f5ec3c69b13ebc57de0c7"},{"sha":"68ec207a320bd50ca61e820c9ff559f799c2ab0a","author":{"name":"Arun Mahadevan","email":"4379f93dea59fea8d2b1a4522aa71b3af9b07910@apache.org"},"message":"[SPARK-25260][SQL] Fix namespace handling in SchemaConverters.toAvroType\n\n## What changes were proposed in this pull request?\n\n`toAvroType` converts spark data type to avro schema. It always appends the record name to namespace so its impossible to have an Avro namespace independent of the record name.\n\nWhen invoked with a spark data type like,\n\n```java\nval sparkSchema = StructType(Seq(\n    StructField(\"name\", StringType, nullable = false),\n    StructField(\"address\", StructType(Seq(\n        StructField(\"city\", StringType, nullable = false),\n        StructField(\"state\", StringType, nullable = false))),\n    nullable = false)))\n\n// map it to an avro schema with record name \"employee\" and top level namespace \"foo.bar\",\nval avroSchema = SchemaConverters.toAvroType(sparkSchema,  false, \"employee\", \"foo.bar\")\n\n// result is\n// avroSchema.getName = employee\n// avroSchema.getNamespace = foo.bar.employee\n// avroSchema.getFullname = foo.bar.employee.employee\n```\nThe patch proposes to fix this so that the result is\n\n```\navroSchema.getName = employee\navroSchema.getNamespace = foo.bar\navroSchema.getFullname = foo.bar.employee\n```\n## How was this patch tested?\n\nNew and existing unit tests.\n\nPlease review http://spark.apache.org/contributing.html before opening a pull request.\n\nCloses #22251 from arunmahadevan/avro-fix.\n\nAuthored-by: Arun Mahadevan <arunm@apache.org>\nSigned-off-by: hyukjinkwon <gurwls223@apache.org>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/68ec207a320bd50ca61e820c9ff559f799c2ab0a"},{"sha":"38391c9aa8a88fcebb337934f30298a32d91596b","author":{"name":"Imran Rashid","email":"b54fbb0e17aa598fdc04bb3d79ba76b5f0737666@cloudera.com"},"message":"[SPARK-25253][PYSPARK] Refactor local connection & auth code\n\nThis eliminates some duplication in the code to connect to a server on localhost to talk directly to the jvm.  Also it gives consistent ipv6 and error handling.  Two other incidental changes, that shouldn't matter:\n1) python barrier tasks perform authentication immediately (rather than waiting for the BARRIER_FUNCTION indicator)\n2) for `rdd._load_from_socket`, the timeout is only increased after authentication.\n\nCloses #22247 from squito/py_connection_refactor.\n\nAuthored-by: Imran Rashid <irashid@cloudera.com>\nSigned-off-by: hyukjinkwon <gurwls223@apache.org>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/38391c9aa8a88fcebb337934f30298a32d91596b"},{"sha":"ff8dcc1d4c684e1b68e63d61b3f20284b9979cca","author":{"name":"DB Tsai","email":"0430fe8116355e2ecfa7f2b79341c5ad5a243c66@apple.com"},"message":"[SPARK-25235][SHELL] Merge the REPL code in Scala 2.11 and 2.12 branches\n\n## What changes were proposed in this pull request?\n\nUsing some reflection tricks to merge Scala 2.11 and 2.12 codebase.\n\n## How was this patch tested?\n\nExisting tests.\n\nCloses #22246 from dbtsai/repl.\n\nLead-authored-by: DB Tsai <d_tsai@apple.com>\nCo-authored-by: Liang-Chi Hsieh <viirya@gmail.com>\nSigned-off-by: DB Tsai <d_tsai@apple.com>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/ff8dcc1d4c684e1b68e63d61b3f20284b9979cca"},{"sha":"82c18c240a6913a917df3b55cc5e22649561c4dd","author":{"name":"Bryan Cutler","email":"060dd14c05a8ffa8f201883ef57d0676b7c9c04a@gmail.com"},"message":"[SPARK-23030][SQL][PYTHON] Use Arrow stream format for creating from and collecting Pandas DataFrames\n\n## What changes were proposed in this pull request?\n\nThis changes the calls of `toPandas()` and `createDataFrame()` to use the Arrow stream format, when Arrow is enabled.  Previously, Arrow data was written to byte arrays where each chunk is an output of the Arrow file format.  This was mainly due to constraints at the time, and caused some overhead by writing the schema/footer on each chunk of data and then having to read multiple Arrow file inputs and concat them together.\n\nUsing the Arrow stream format has improved these by increasing performance, lower memory overhead for the average case, and simplified the code.  Here are the details of this change:\n\n**toPandas()**\n\n_Before:_\nSpark internal rows are converted to Arrow file format, each group of records is a complete Arrow file which contains the schema and other metadata.  Next a collect is done and an Array of Arrow files is the result.  After that each Arrow file is sent to Python driver which then loads each file and concats them to a single Arrow DataFrame.\n\n_After:_\nSpark internal rows are converted to ArrowRecordBatches directly, which is the simplest Arrow component for IPC data transfers.  The driver JVM then immediately starts serving data to Python as an Arrow stream, sending the schema first. It then starts a Spark job with a custom handler that sends Arrow RecordBatches to Python. Partitions arriving in order are sent immediately, and out-of-order partitions are buffered until the ones that precede it come in. This improves performance, simplifies memory usage on executors, and improves the average memory usage on the JVM driver.  Since the order of partitions must be preserved, the worst case is that the first partition will be the last to arrive all data must be buffered in memory until then. This case is no worse that before when doing a full collect.\n\n**createDataFrame()**\n\n_Before:_\nA Pandas DataFrame is split into parts and each part is made into an Arrow file.  Then each file is prefixed by the buffer size and written to a temp file.  The temp file is read and each Arrow file is parallelized as a byte array.\n\n_After:_\nA Pandas DataFrame is split into parts, then an Arrow stream is written to a temp file where each part is an ArrowRecordBatch.  The temp file is read as a stream and the Arrow messages are examined.  If the message is an ArrowRecordBatch, the data is saved as a byte array.  After reading the file, each ArrowRecordBatch is parallelized as a byte array.  This has slightly more processing than before because we must look each Arrow message to extract the record batches, but performance ends up a litle better.  It is cleaner in the sense that IPC from Python to JVM is done over a single Arrow stream.\n\n## How was this patch tested?\n\nAdded new unit tests for the additions to ArrowConverters in Scala, existing tests for Python.\n\n## Performance Tests - toPandas\n\nTests run on a 4 node standalone cluster with 32 cores total, 14.04.1-Ubuntu and OpenJDK 8\nmeasured wall clock time to execute `toPandas()` and took the average best time of 5 runs/5 loops each.\n\nTest code\n```python\ndf = spark.range(1 << 25, numPartitions=32).toDF(\"id\").withColumn(\"x1\", rand()).withColumn(\"x2\", rand()).withColumn(\"x3\", rand()).withColumn(\"x4\", rand())\nfor i in range(5):\n\tstart = time.time()\n\t_ = df.toPandas()\n\telapsed = time.time() - start\n```\n\nCurrent Master | This PR\n---------------------|------------\n5.803557 | 5.16207\n5.409119 | 5.133671\n5.493509 | 5.147513\n5.433107 | 5.105243\n5.488757 | 5.018685\n\nAvg Master | Avg This PR\n------------------|--------------\n5.5256098 | 5.1134364\n\nSpeedup of **1.08060595**\n\n## Performance Tests - createDataFrame\n\nTests run on a 4 node standalone cluster with 32 cores total, 14.04.1-Ubuntu and OpenJDK 8\nmeasured wall clock time to execute `createDataFrame()` and get the first record. Took the average best time of 5 runs/5 loops each.\n\nTest code\n```python\ndef run():\n\tpdf = pd.DataFrame(np.random.rand(10000000, 10))\n\tspark.createDataFrame(pdf).first()\n\nfor i in range(6):\n\tstart = time.time()\n\trun()\n\telapsed = time.time() - start\n\tgc.collect()\n\tprint(\"Run %d: %f\" % (i, elapsed))\n```\n\nCurrent Master | This PR\n--------------------|----------\n6.234608 | 5.665641\n6.32144 | 5.3475\n6.527859 | 5.370803\n6.95089 | 5.479151\n6.235046 | 5.529167\n\nAvg Master | Avg This PR\n---------------|----------------\n6.4539686 | 5.4784524\n\nSpeedup of **1.178064192**\n\n## Memory Improvements\n\n**toPandas()**\n\nThe most significant improvement is reduction of the upper bound space complexity in the JVM driver.  Before, the entire dataset was collected in the JVM first before sending it to Python.  With this change, as soon as a partition is collected, the result handler immediately sends it to Python, so the upper bound is the size of the largest partition.  Also, using the Arrow stream format is more efficient because the schema is written once per stream, followed by record batches.  The schema is now only send from driver JVM to Python.  Before, multiple Arrow file formats were used that each contained the schema.  This duplicated schema was created in the executors, sent to the driver JVM, and then Python where all but the first one received are discarded.\n\nI verified the upper bound limit by running a test that would collect data that would exceed the amount of driver JVM memory available.  Using these settings on a standalone cluster:\n```\nspark.driver.memory 1g\nspark.executor.memory 5g\nspark.sql.execution.arrow.enabled true\nspark.sql.execution.arrow.fallback.enabled false\nspark.sql.execution.arrow.maxRecordsPerBatch 0\nspark.driver.maxResultSize 2g\n```\n\nTest code:\n```python\nfrom pyspark.sql.functions import rand\ndf = spark.range(1 << 25, numPartitions=32).toDF(\"id\").withColumn(\"x1\", rand()).withColumn(\"x2\", rand()).withColumn(\"x3\", rand())\ndf.toPandas()\n```\n\nThis makes total data size of 33554432×8×4 = 1073741824\n\nWith the current master, it fails with OOM but passes using this PR.\n\n**createDataFrame()**\n\nNo significant change in memory except that using the stream format instead of separate file formats avoids duplicated the schema, similar to toPandas above.  The process of reading the stream and parallelizing the batches does cause the record batch message metadata to be copied, but it's size is insignificant.\n\nCloses #21546 from BryanCutler/arrow-toPandas-stream-SPARK-23030.\n\nAuthored-by: Bryan Cutler <cutlerb@gmail.com>\nSigned-off-by: hyukjinkwon <gurwls223@apache.org>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/82c18c240a6913a917df3b55cc5e22649561c4dd"},{"sha":"1fd59c129a7aa16f9960b109128b166952992f32","author":{"name":"Sean Owen","email":"3f9d48d7c3f24f7009479c5fa630bd1131ffc724@databricks.com"},"message":" [WIP][SPARK-25044][SQL] (take 2) Address translation of LMF closure primitive args to Object in Scala 2.12\n\n## What changes were proposed in this pull request?\n\nAlternative take on https://github.com/apache/spark/pull/22063 that does not introduce udfInternal.\nResolve issue with inferring func types in 2.12 by instead using info captured when UDF is registered -- capturing which types are nullable (i.e. not primitive)\n\n## How was this patch tested?\n\nExisting tests.\n\nCloses #22259 from srowen/SPARK-25044.2.\n\nAuthored-by: Sean Owen <sean.owen@databricks.com>\nSigned-off-by: Wenchen Fan <wenchen@databricks.com>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/1fd59c129a7aa16f9960b109128b166952992f32"},{"sha":"3864480e14a4961720cc1be43635c7c7dec08c09","author":{"name":"sarutak","email":"7d45abe39fd242ee2f6575da28f13d66568418b3@oss.nttdata.co.jp"},"message":"[SPARK-25266][CORE] Fix memory leak in Barrier Execution Mode\n\n## What changes were proposed in this pull request?\n\nBarrierCoordinator uses Timer and TimerTask. `TimerTask#cancel()` is invoked in ContextBarrierState#cancelTimerTask but `Timer#purge()` is never invoked.\n\nOnce a TimerTask is scheduled, the reference to it is not released until `Timer#purge()` is invoked even though `TimerTask#cancel()` is invoked.\n\n## How was this patch tested?\n\nI checked the number of instances related to the TimerTask using jmap.\n\nCloses #22258 from sarutak/fix-barrierexec-oom.\n\nAuthored-by: sarutak <sarutak@oss.nttdata.co.jp>\nSigned-off-by: Xiangrui Meng <meng@databricks.com>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/3864480e14a4961720cc1be43635c7c7dec08c09"},{"sha":"20b7c684cc4a8136b9a9c56390a4948de04e7c34","author":{"name":"Xiangrui Meng","email":"9616cf314f60cbadfd3d5d51110130ad7923defb@databricks.com"},"message":"[SPARK-25248][.1][PYSPARK] update barrier Python API\n\n## What changes were proposed in this pull request?\n\nI made one pass over the Python APIs for barrier mode and updated them to match the Scala doc in #22240 . Major changes:\n\n* export the public classes\n* expand the docs\n* add doc for BarrierTaskInfo.addresss\n\ncc: jiangxb1987\n\nCloses #22261 from mengxr/SPARK-25248.1.\n\nAuthored-by: Xiangrui Meng <meng@databricks.com>\nSigned-off-by: Xiangrui Meng <meng@databricks.com>","distinct":true,"url":"https://api.github.com/repos/srowen/spark/commits/20b7c684cc4a8136b9a9c56390a4948de04e7c34"}]}
{"actor":{"display_login":"srowen"}}