SPARK-3926 [CORE] Result of JavaRDD.collectAsMap() is not Serializable

Make JavaPairRDD.collectAsMap result Serializable since Java Maps generally are

Author: Sean Owen <sowen@cloudera.com>

Closes #2805 from srowen/SPARK-3926 and squashes the following commits:

ecb78ee [Sean Owen] Fix conflict between java.io.Serializable and use of Scala's Serializable
f4717f9 [Sean Owen] Oops, fix compile problem
ae1b36f [Sean Owen] Expand to cover Maps returned from other Java API methods as well
51c26c2 [Sean Owen] Make JavaPairRDD.collectAsMap result Serializable since Java Maps generally are
[SPARK-2546] Clone JobConf for each task (branch-1.0 / 1.1 backport)

This patch attempts to fix SPARK-2546 in `branch-1.0` and `branch-1.1`.  The underlying problem is that thread-safety issues in Hadoop Configuration objects may cause Spark tasks to get stuck in infinite loops.  The approach taken here is to clone a new copy of the JobConf for each task rather than sharing a single copy between tasks.  Note that there are still Configuration thread-safety issues that may affect the driver, but these seem much less likely to occur in practice and will be more complex to fix (see discussion on the SPARK-2546 ticket).

This cloning is guarded by a new configuration option (`spark.hadoop.cloneConf`) and is disabled by default in order to avoid unexpected performance regressions for workloads that are unaffected by the Configuration thread-safety issues.

Author: Josh Rosen <joshrosen@apache.org>

Closes #2684 from JoshRosen/jobconf-fix-backport and squashes the following commits:

f14f259 [Josh Rosen] Add configuration option to control cloning of Hadoop JobConf.
b562451 [Josh Rosen] Remove unused jobConfCacheKey field.
dd25697 [Josh Rosen] [SPARK-2546] [1.0 / 1.1 backport] Clone JobConf for each task.
[SPARK-3948][Shuffle]Fix stream corruption bug in sort-based shuffle

Kernel 2.6.32 bug will lead to unexpected behavior of transferTo in copyStream, and this will corrupt the shuffle output file in sort-based shuffle, which will somehow introduce PARSING_ERROR(2), deserialization error or offset out of range. Here fix this by adding append flag, also add some position checking code. Details can be seen in [SPARK-3948](https://issues.apache.org/jira/browse/SPARK-3948).

Author: jerryshao <saisai.shao@intel.com>

Closes #2824 from jerryshao/SPARK-3948 and squashes the following commits:

be0533a [jerryshao] Address the comments
a82b184 [jerryshao] add configuration to control the NIO way of copying stream
e17ada2 [jerryshao] Fix kernel 2.6.32 bug led unexpected behavior of transferTo

(cherry picked from commit c7aeecd08fd329085760fa89025ec0d9c04f5e3f)
Signed-off-by: Josh Rosen <joshrosen@databricks.com>

Conflicts:
	core/src/main/scala/org/apache/spark/util/Utils.scala
[SPARK-4010][Web UI]Spark UI returns 500 in yarn-client mode

The problem caused by #1966
CC YanTangZhai andrewor14

Author: GuoQiang Li <witgo@qq.com>

Closes #2858 from witgo/SPARK-4010 and squashes the following commits:

9866fbf [GuoQiang Li] Spark UI returns 500 in yarn-client mode

(cherry picked from commit 51afde9d8b8a67958c4632a13af143d7c7fd1f04)
Signed-off-by: Andrew Or <andrewor14@gmail.com>
[SPARK-3877][YARN] Throw an exception when application is not successful so that the exit code wil be set to 1 (for branch-1.1)

This is a patch to fix SPARK-3877 in branch-1.1. See also #2732

Author: zsxwing <zsxwing@gmail.com>

Closes #2748 from zsxwing/SPARK-3877-branch-1.1 and squashes the following commits:

3701984 [zsxwing] Remove System.exit from Client.scala
8681881 [zsxwing] [SPARK-3877] Throw an exception when application is not successful so that the exit code wil be set to 1
[SPARK-3426] Fix sort-based shuffle error when spark.shuffle.compress and spark.shuffle.spill.compress settings are different

This PR fixes SPARK-3426, an issue where sort-based shuffle crashes if the
`spark.shuffle.spill.compress` and `spark.shuffle.compress` settings have
different values.

The problem is that sort-based shuffle's read and write paths use different
settings for determining whether to apply compression.  ExternalSorter writes
runs to files using `TempBlockId` ids, which causes
`spark.shuffle.spill.compress` to be used for enabling compression, but these
spilled files end up being shuffled over the network and read as shuffle files
using `ShuffleBlockId` by BlockStoreShuffleFetcher, which causes
`spark.shuffle.compress` to be used for enabling decompression.  As a result,
this leads to errors when these settings disagree.

Based on the discussions in #2247 and #2178, it sounds like we don't want to
remove the `spark.shuffle.spill.compress` setting.  Therefore, I've tried to
come up with a fix where `spark.shuffle.spill.compress` is used to compress
data that's read and written locally and `spark.shuffle.compress` is used to
compress any data that will be fetched / read as shuffle blocks.

To do this, I split `TempBlockId` into two new id types, `TempLocalBlockId` and
`TempShuffleBlockId`, which map to `spark.shuffle.spill.compress` and
`spark.shuffle.compress`, respectively.  ExternalAppendOnlyMap also used temp
blocks for spilling data.  It looks like ExternalSorter was designed to be
a generic sorter but its configuration already happens to be tied to sort-based
shuffle, so I think it's fine if we use `spark.shuffle.compress` to compress
its spills; we can move the compression configuration to the constructor in
a later commit if we find that ExternalSorter is being used in other contexts
where we want different configuration options to control compression.  To
summarize:

**Before:**

|       | ExternalAppendOnlyMap        | ExternalSorter               |
|-------|------------------------------|------------------------------|
| Read  | spark.shuffle.spill.compress | spark.shuffle.compress       |
| Write | spark.shuffle.spill.compress | spark.shuffle.spill.compress |

**After:**

|       | ExternalAppendOnlyMap        | ExternalSorter         |
|-------|------------------------------|------------------------|
| Read  | spark.shuffle.spill.compress | spark.shuffle.compress |
| Write | spark.shuffle.spill.compress | spark.shuffle.compress |

Thanks to andrewor14 for debugging this with me!

Author: Josh Rosen <joshrosen@databricks.com>

Closes #2890 from JoshRosen/SPARK-3426 and squashes the following commits:

1921cf6 [Josh Rosen] Minor edit for clarity.
c8dd8f2 [Josh Rosen] Add comment explaining use of createTempShuffleBlock().
2c687b9 [Josh Rosen] Fix SPARK-3426.
91e7e40 [Josh Rosen] Combine tests into single test of all combinations
76ca65e [Josh Rosen] Add regression test for SPARK-3426.

Conflicts:
	core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
[SPARK-2652] [PySpark] donot use KyroSerializer as default serializer

KyroSerializer can not serialize customized class without registered explicitly, use it as default serializer in PySpark will introduce some regression in MLlib.

cc mengxr

Author: Davies Liu <davies@databricks.com>

Closes #2916 from davies/revert and squashes the following commits:

43eb6d3 [Davies Liu] donot use KyroSerializer as default serializer

(cherry picked from commit 809c785bcc33e684a68ea14240a466def864199a)
Signed-off-by: Xiangrui Meng <meng@databricks.com>
[SPARK-4076] Parameter expansion in spark-config is wrong

In sbin/spark-config.sh, parameter expansion is used to extract source root as follows.

    this="${BASH_SOURCE-$0}"

I think, the parameter expansion should be ":" instead of "".
If we use "-" and BASH_SOURCE="", (empty character is set, not unset),
"" (empty character) is set to $this.

Author: Kousuke Saruta <sarutak@oss.nttdata.co.jp>

Closes #2930 from sarutak/SPARK-4076 and squashes the following commits:

32a0370 [Kousuke Saruta] Fixed wrong parameter expansion

(cherry picked from commit 30ea2868e7afbec20bfc83818249b6d2d7dc6aec)
Signed-off-by: Andrew Or <andrew@databricks.com>

Conflicts:
	sbin/spark-config.sh
[SPARK-4075] [Deploy] Jar url validation is not enough for Jar file

In deploy.ClientArguments.isValidJarUrl, the url is checked as follows.

    def isValidJarUrl(s: String): Boolean = s.matches("(.+):(.+)jar")

So, it allows like 'hdfs:file.jar' (no authority).

Author: Kousuke Saruta <sarutak@oss.nttdata.co.jp>

Closes #2925 from sarutak/uri-syntax-check-improvement and squashes the following commits:

cf06173 [Kousuke Saruta] Improved URI syntax checking

(cherry picked from commit 098f83c7ccd7dad9f9228596da69fe5f55711a52)
Signed-off-by: Andrew Or <andrew@databricks.com>
[SPARK-4006] In long running contexts, we encountered the situation of d...

...ouble registe...

...r without a remove in between. The cause for that is unknown, and assumed a temp network issue.

However, since the second register is with a BlockManagerId on a different port, blockManagerInfo.contains() returns false, while blockManagerIdByExecutor returns Some. This inconsistency is caught in a conditional statement that does System.exit(1), which is a huge robustness issue for us.

The fix - simply remove the old id from both maps during register when this happens. We are mimicking the behavior of expireDeadHosts(), by doing local cleanup of the maps before trying to add new ones.

Also - added some logging for register and unregister.

This is just like https://github.com/apache/spark/pull/2886 except it's on branch-1.1

Author: Tal Sliwowicz <tal.s@taboola.com>

Closes #2915 from tsliwowicz/branch-1.1-block-mgr-removal and squashes the following commits:

d122236 [Tal Sliwowicz] [SPARK-4006] In long running contexts, we encountered the situation of double registe...
[SPARK-4080] Only throw IOException from [write|read][Object|External]

If classes implementing Serializable or Externalizable interfaces throw
exceptions other than IOException or ClassNotFoundException from their
(de)serialization methods, then this results in an unhelpful
"IOException: unexpected exception type" rather than the actual exception that
produced the (de)serialization error.

This patch fixes this by adding a utility method that re-wraps any uncaught
exceptions in IOException (unless they are already instances of IOException).

Author: Josh Rosen <joshrosen@databricks.com>

Closes #2932 from JoshRosen/SPARK-4080 and squashes the following commits:

cd3a9be [Josh Rosen] [SPARK-4080] Only throw IOException from [write|read][Object|External].

(cherry picked from commit 6c98c29ae0033556fd4424f41d1de005c509e511)
Signed-off-by: Josh Rosen <joshrosen@databricks.com>

Conflicts:
	core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
	core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala
	streaming/src/main/scala/org/apache/spark/streaming/api/python/PythonDStream.scala
[SPARK-4056] Upgrade snappy-java to 1.1.1.5

This upgrades snappy-java to 1.1.1.5, which improves error messages when attempting to deserialize empty inputs using SnappyInputStream (see https://github.com/xerial/snappy-java/issues/89).

Author: Josh Rosen <rosenville@gmail.com>
Author: Josh Rosen <joshrosen@databricks.com>

Closes #2911 from JoshRosen/upgrade-snappy-java and squashes the following commits:

adec96c [Josh Rosen] Use snappy-java 1.1.1.5
cc953d6 [Josh Rosen] [SPARK-4056] Upgrade snappy-java to 1.1.1.4

(cherry picked from commit 898b22ab1fe90e8a3935b19566465046f2256fa6)
Signed-off-by: Josh Rosen <joshrosen@databricks.com>

Conflicts:
	pom.xml
Revert "[SPARK-4056] Upgrade snappy-java to 1.1.1.5"

This reverts commit b7541ae89c3db71979f11f2f0b2cb737cb5d1fb3.
Fix build breakage introduced by 6c10c2770c718287f9cc2af4109b701fa1057b70
[MLlib] SPARK-3987: add test case on objective value for NNLS

Also update step parameter to pass the proposed test

Author: coderxiang <shuoxiangpub@gmail.com>

Closes #2965 from coderxiang/nnls-test and squashes the following commits:

24b06f9 [coderxiang] add test case on objective value for NNLS; update step parameter to pass the test

(cherry picked from commit 7e3a1ada86e6adf1ddd4d8a321824daf5f3b2c75)
Signed-off-by: Xiangrui Meng <meng@databricks.com>
[SPARK-4110] Wrong comments about default settings in spark-daemon.sh

In spark-daemon.sh, thare are following comments.

    #   SPARK_CONF_DIR  Alternate conf dir. Default is ${SPARK_PREFIX}/conf.
    #   SPARK_LOG_DIR   Where log files are stored.  PWD by default.

But, I think the default value for SPARK_CONF_DIR is `${SPARK_HOME}/conf` and for SPARK_LOG_DIR is `${SPARK_HOME}/logs`.

Author: Kousuke Saruta <sarutak@oss.nttdata.co.jp>

Closes #2972 from sarutak/SPARK-4110 and squashes the following commits:

5a171a2 [Kousuke Saruta] Fixed wrong comments

(cherry picked from commit 44d8b45a38c8d934628373a3b21084432516ee00)
Signed-off-by: Andrew Or <andrew@databricks.com>
[SPARK-4107] Fix incorrect handling of read() and skip() return values (branch-1.1 backport)

`read()` may return fewer bytes than requested; when this occurred, the old code would silently return less data than requested, which might cause stream corruption errors.  `skip()` faces similar issues, too.

This patch fixes several cases where we mis-handle these methods' return values.

This is a backport of #2969 to `branch-1.1`.

Author: Josh Rosen <joshrosen@databricks.com>

Closes #2974 from JoshRosen/spark-4107-branch-1.1-backport and squashes the following commits:

d82c05b [Josh Rosen] [SPARK-4107] Fix incorrect handling of read() and skip() return values
[SPARK-4065] Add check for IPython on Windows

This issue employs logic similar to the bash launcher (pyspark) to check
if IPTYHON=1, and if so launch ipython with options in IPYTHON_OPTS.
This fix assumes that ipython is available in the system Path, and can
be invoked with a plain "ipython" command.

Author: Michael Griffiths <msjgriffiths@gmail.com>

Closes #2910 from msjgriffiths/pyspark-windows and squashes the following commits:

ef34678 [Michael Griffiths] Change build message to comply with [SPARK-3775]
361e3d8 [Michael Griffiths] [SPARK-4065] Add check for IPython on Windows
9ce72d1 [Michael Griffiths] [SPARK-4065] Add check for IPython on Windows

(cherry picked from commit 2f254dacf4b7ab9c59c7cef59fd364ca682162ae)
Signed-off-by: Andrew Or <andrew@databricks.com>