[SPARK-23522][Python] always use sys.exit over builtin exit

The exit() builtin is only for interactive use. applications should use sys.exit().

dev/merge_spark_pr.py

@@ -510,7 +510,7 @@ def main():
     import doctest
     (failure_count, test_count) = doctest.testmod()
     if failure_count:
-        exit(-1)
+        sys.exit(-1)
     try:
         main()
     except:

dev/run-tests.py

@@ -621,7 +621,7 @@ def _test():
     import doctest
     failure_count = doctest.testmod()[0]
     if failure_count:
-        exit(-1)
+        sys.exit(-1)
 
 if __name__ == "__main__":
     _test()

examples/src/main/python/avro_inputformat.py

@@ -61,7 +61,7 @@
         Assumes you have Avro data stored in <data_file>. Reader schema can be optionally specified
         in [reader_schema_file].
         """, file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     path = sys.argv[1]
 

examples/src/main/python/kmeans.py

@@ -49,7 +49,7 @@ def closestPoint(p, centers):
 
     if len(sys.argv) != 4:
         print("Usage: kmeans <file> <k> <convergeDist>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     print("""WARN: This is a naive implementation of KMeans Clustering and is given
        as an example! Please refer to examples/src/main/python/ml/kmeans_example.py for an

examples/src/main/python/logistic_regression.py

@@ -48,7 +48,7 @@ def readPointBatch(iterator):
 
     if len(sys.argv) != 3:
         print("Usage: logistic_regression <file> <iterations>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     print("""WARN: This is a naive implementation of Logistic Regression and is
       given as an example!

examples/src/main/python/ml/dataframe_example.py

@@ -33,7 +33,7 @@
 if __name__ == "__main__":
     if len(sys.argv) > 2:
         print("Usage: dataframe_example.py <libsvm file>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     elif len(sys.argv) == 2:
         input = sys.argv[1]
     else:

examples/src/main/python/mllib/correlations.py

@@ -31,7 +31,7 @@
 if __name__ == "__main__":
     if len(sys.argv) not in [1, 2]:
         print("Usage: correlations (<file>)", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     sc = SparkContext(appName="PythonCorrelations")
     if len(sys.argv) == 2:
         filepath = sys.argv[1]

examples/src/main/python/mllib/kmeans.py

@@ -36,7 +36,7 @@ def parseVector(line):
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: kmeans <file> <k>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     sc = SparkContext(appName="KMeans")
     lines = sc.textFile(sys.argv[1])
     data = lines.map(parseVector)

examples/src/main/python/mllib/logistic_regression.py

@@ -42,7 +42,7 @@ def parsePoint(line):
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: logistic_regression <file> <iterations>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     sc = SparkContext(appName="PythonLR")
     points = sc.textFile(sys.argv[1]).map(parsePoint)
     iterations = int(sys.argv[2])

examples/src/main/python/mllib/random_rdd_generation.py

@@ -29,7 +29,7 @@
 if __name__ == "__main__":
     if len(sys.argv) not in [1, 2]:
         print("Usage: random_rdd_generation", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     sc = SparkContext(appName="PythonRandomRDDGeneration")
 

examples/src/main/python/mllib/sampled_rdds.py

@@ -29,7 +29,7 @@
 if __name__ == "__main__":
     if len(sys.argv) not in [1, 2]:
         print("Usage: sampled_rdds <libsvm data file>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     if len(sys.argv) == 2:
         datapath = sys.argv[1]
     else:
@@ -43,7 +43,7 @@
     numExamples = examples.count()
     if numExamples == 0:
         print("Error: Data file had no samples to load.", file=sys.stderr)
-        exit(1)
+        sys.exit(1)
     print('Loaded data with %d examples from file: %s' % (numExamples, datapath))
 
     # Example: RDD.sample() and RDD.takeSample()

examples/src/main/python/mllib/streaming_linear_regression_example.py

@@ -36,7 +36,7 @@
     if len(sys.argv) != 3:
         print("Usage: streaming_linear_regression_example.py <trainingDir> <testDir>",
               file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     sc = SparkContext(appName="PythonLogisticRegressionWithLBFGSExample")
     ssc = StreamingContext(sc, 1)

examples/src/main/python/pagerank.py

@@ -47,7 +47,7 @@ def parseNeighbors(urls):
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: pagerank <file> <iterations>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     print("WARN: This is a naive implementation of PageRank and is given as an example!\n" +
           "Please refer to PageRank implementation provided by graphx",

examples/src/main/python/parquet_inputformat.py

@@ -45,7 +45,7 @@
                 /path/to/examples/parquet_inputformat.py <data_file>
         Assumes you have Parquet data stored in <data_file>.
         """, file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     path = sys.argv[1]
 

examples/src/main/python/sort.py

@@ -25,7 +25,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 2:
         print("Usage: sort <file>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     spark = SparkSession\
         .builder\

examples/src/main/python/sql/streaming/structured_kafka_wordcount.py

@@ -49,7 +49,7 @@
         print("""
         Usage: structured_kafka_wordcount.py <bootstrap-servers> <subscribe-type> <topics>
         """, file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     bootstrapServers = sys.argv[1]
     subscribeType = sys.argv[2]

examples/src/main/python/sql/streaming/structured_network_wordcount.py

@@ -38,7 +38,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: structured_network_wordcount.py <hostname> <port>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     host = sys.argv[1]
     port = int(sys.argv[2])

examples/src/main/python/sql/streaming/structured_network_wordcount_windowed.py

@@ -53,7 +53,7 @@
         msg = ("Usage: structured_network_wordcount_windowed.py <hostname> <port> "
                "<window duration in seconds> [<slide duration in seconds>]")
         print(msg, file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     host = sys.argv[1]
     port = int(sys.argv[2])

examples/src/main/python/streaming/direct_kafka_wordcount.py

@@ -39,7 +39,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: direct_kafka_wordcount.py <broker_list> <topic>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     sc = SparkContext(appName="PythonStreamingDirectKafkaWordCount")
     ssc = StreamingContext(sc, 2)

examples/src/main/python/streaming/flume_wordcount.py

@@ -39,7 +39,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: flume_wordcount.py <hostname> <port>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     sc = SparkContext(appName="PythonStreamingFlumeWordCount")
     ssc = StreamingContext(sc, 1)

examples/src/main/python/streaming/hdfs_wordcount.py

@@ -35,7 +35,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 2:
         print("Usage: hdfs_wordcount.py <directory>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     sc = SparkContext(appName="PythonStreamingHDFSWordCount")
     ssc = StreamingContext(sc, 1)

examples/src/main/python/streaming/kafka_wordcount.py

@@ -39,7 +39,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: kafka_wordcount.py <zk> <topic>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     sc = SparkContext(appName="PythonStreamingKafkaWordCount")
     ssc = StreamingContext(sc, 1)

examples/src/main/python/streaming/network_wordcount.py

@@ -35,7 +35,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: network_wordcount.py <hostname> <port>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     sc = SparkContext(appName="PythonStreamingNetworkWordCount")
     ssc = StreamingContext(sc, 1)
 

examples/src/main/python/streaming/network_wordjoinsentiments.py

@@ -47,7 +47,7 @@ def print_happiest_words(rdd):
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: network_wordjoinsentiments.py <hostname> <port>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     sc = SparkContext(appName="PythonStreamingNetworkWordJoinSentiments")
     ssc = StreamingContext(sc, 5)

examples/src/main/python/streaming/recoverable_network_wordcount.py

@@ -101,7 +101,7 @@ def filterFunc(wordCount):
     if len(sys.argv) != 5:
         print("Usage: recoverable_network_wordcount.py <hostname> <port> "
               "<checkpoint-directory> <output-file>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     host, port, checkpoint, output = sys.argv[1:]
     ssc = StreamingContext.getOrCreate(checkpoint,
                                        lambda: createContext(host, int(port), output))

examples/src/main/python/streaming/sql_network_wordcount.py

@@ -48,7 +48,7 @@ def getSparkSessionInstance(sparkConf):
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: sql_network_wordcount.py <hostname> <port> ", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     host, port = sys.argv[1:]
     sc = SparkContext(appName="PythonSqlNetworkWordCount")
     ssc = StreamingContext(sc, 1)

examples/src/main/python/streaming/stateful_network_wordcount.py

@@ -39,7 +39,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 3:
         print("Usage: stateful_network_wordcount.py <hostname> <port>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
     sc = SparkContext(appName="PythonStreamingStatefulNetworkWordCount")
     ssc = StreamingContext(sc, 1)
     ssc.checkpoint("checkpoint")

examples/src/main/python/wordcount.py

@@ -26,7 +26,7 @@
 if __name__ == "__main__":
     if len(sys.argv) != 2:
         print("Usage: wordcount <file>", file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
     spark = SparkSession\
         .builder\

python/pyspark/accumulators.py

@@ -266,4 +266,4 @@ def _start_update_server():
     import doctest
     (failure_count, test_count) = doctest.testmod()
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/broadcast.py

@@ -162,4 +162,4 @@ def clear(self):
     import doctest
     (failure_count, test_count) = doctest.testmod()
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/conf.py

@@ -217,7 +217,7 @@ def _test():
     import doctest
     (failure_count, test_count) = doctest.testmod(optionflags=doctest.ELLIPSIS)
     if failure_count:
-        exit(-1)
+        sys.exit(-1)
 
 
 if __name__ == "__main__":

python/pyspark/context.py

@@ -1035,7 +1035,7 @@ def _test():
     (failure_count, test_count) = doctest.testmod(globs=globs, optionflags=doctest.ELLIPSIS)
     globs['sc'].stop()
     if failure_count:
-        exit(-1)
+        sys.exit(-1)
 
 
 if __name__ == "__main__":

python/pyspark/daemon.py

@@ -89,7 +89,7 @@ def shutdown(code):
         signal.signal(SIGTERM, SIG_DFL)
         # Send SIGHUP to notify workers of shutdown
         os.kill(0, SIGHUP)
-        exit(code)
+        sys.exit(code)
 
     def handle_sigterm(*args):
         shutdown(1)

python/pyspark/find_spark_home.py

@@ -68,7 +68,7 @@ def is_spark_home(path):
         return next(path for path in paths if is_spark_home(path))
     except StopIteration:
         print("Could not find valid SPARK_HOME while searching {0}".format(paths), file=sys.stderr)
-        exit(-1)
+        sys.exit(-1)
 
 if __name__ == "__main__":
     print(_find_spark_home())

python/pyspark/heapq3.py

@@ -883,7 +883,7 @@ def nlargest(n, iterable, key=None):
 
 
 if __name__ == "__main__":
-    import doctest
+    import doctest, sys
     (failure_count, test_count) = doctest.testmod()
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/ml/classification.py

@@ -16,6 +16,7 @@
 #
 
 import operator
+import sys
 from multiprocessing.pool import ThreadPool
 
 from pyspark import since, keyword_only
@@ -2043,4 +2044,4 @@ def _to_java(self):
         except OSError:
             pass
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/ml/clustering.py

@@ -1181,4 +1181,4 @@ def getKeepLastCheckpoint(self):
         except OSError:
             pass
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/ml/evaluation.py

@@ -15,6 +15,8 @@
 # limitations under the License.
 #
 
+import sys
+
 from abc import abstractmethod, ABCMeta
 
 from pyspark import since, keyword_only
@@ -446,4 +448,4 @@ def getDistanceMeasure(self):
         except OSError:
             pass
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/ml/feature.py

@@ -3717,4 +3717,4 @@ def setSize(self, value):
         except OSError:
             pass
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/ml/image.py

@@ -24,6 +24,8 @@
    :members:
 """
 
+import sys
+
 import numpy as np
 from pyspark import SparkContext
 from pyspark.sql.types import Row, _create_row, _parse_datatype_json_string
@@ -251,7 +253,7 @@ def _test():
         optionflags=doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE)
     spark.stop()
     if failure_count:
-        exit(-1)
+        sys.exit(-1)
 
 
 if __name__ == "__main__":

python/pyspark/ml/linalg/__init__.py

@@ -1158,7 +1158,7 @@ def _test():
     import doctest
     (failure_count, test_count) = doctest.testmod(optionflags=doctest.ELLIPSIS)
     if failure_count:
-        exit(-1)
+        sys.exit(-1)
 
 if __name__ == "__main__":
     _test()

python/pyspark/ml/recommendation.py

@@ -480,4 +480,4 @@ def recommendForItemSubset(self, dataset, numUsers):
         except OSError:
             pass
     if failure_count:
-        exit(-1)
+        sys.exit(-1)

python/pyspark/ml/regression.py

@@ -1812,4 +1812,4 @@ def __repr__(self):
         except OSError:
             pass
     if failure_count:
-        exit(-1)
+        sys.exit(-1)