[SPARK-42075][DSTREAM] Deprecate DStream API

### What changes were proposed in this pull request?
Deprecate the DStream interface by deprecating the StreamingContext api.

### Why are the changes needed?
DStream is not longer maintained. Encourage user to migrate to structured streaming.

### Does this PR introduce _any_ user-facing change?
Yes, user that use DStream will see deprecation warning.

### How was this patch tested?
No functional change

Closes apache#39647 from chaoqin-li1123/deprecate_dstream.

Authored-by: Chaoqin Li <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>

Author: chaoqin-li1123

python/pyspark/streaming/context.py

@@ -26,6 +26,8 @@
 from pyspark.streaming.listener import StreamingListener
 from pyspark.streaming.util import TransformFunction, TransformFunctionSerializer
 
+import warnings
+
 __all__ = ["StreamingContext"]
 
 T = TypeVar("T")
@@ -41,6 +43,12 @@ class StreamingContext:
     respectively. `context.awaitTermination()` allows the current thread
     to wait for the termination of the context by `stop()` or by an exception.
 
+    .. deprecated:: Spark 3.4.0
+       This is deprecated as of Spark 3.4.0.
+       There are no longer updates to DStream and it's a legacy project.
+       There is a newer and easier to use streaming engine in Spark called Structured Streaming.
+       You should use Spark Structured Streaming for your streaming applications.
+
     Parameters
     ----------
     sparkContext : :class:`SparkContext`
@@ -61,6 +69,10 @@ def __init__(
         batchDuration: Optional[int] = None,
         jssc: Optional[JavaObject] = None,
     ):
+        warnings.warn(
+            "DStream is deprecated as of Spark 3.4.0. Migrate to Structured Streaming.",
+            FutureWarning,
+        )
         self._sc = sparkContext
         self._jvm = self._sc._jvm
         self._jssc = jssc or self._initialize_context(self._sc, batchDuration)

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -60,7 +60,13 @@ import org.apache.spark.util.{CallSite, ShutdownHookManager, ThreadUtils, Utils}
  * using `context.start()` and `context.stop()`, respectively.
  * `context.awaitTermination()` allows the current thread to wait for the termination
  * of the context by `stop()` or by an exception.
+ * @deprecated This is deprecated as of Spark 3.4.0.
+ *             There are no longer updates to DStream and it's a legacy project.
+ *             There is a newer and easier to use streaming engine
+ *             in Spark called Structured Streaming.
+ *             You should use Spark Structured Streaming for your streaming applications.
  */
+@deprecated("DStream is deprecated. Migrate to Structured Streaming.", "Spark 3.4.0")
 class StreamingContext private[streaming] (
     _sc: SparkContext,
     _cp: Checkpoint,
@@ -740,13 +746,20 @@ class StreamingContext private[streaming] (
 /**
  * StreamingContext object contains a number of utility functions related to the
  * StreamingContext class.
+ *
+ * @deprecated This is deprecated as of Spark 3.4.0.
+ *             There are no longer updates to DStream and it's a legacy project.
+ *             There is a newer and easier to use streaming engine
+ *             in Spark called Structured Streaming.
+ *             You should use Spark Structured Streaming for your streaming applications.
  */
-
+@deprecated("DStream is deprecated. Migrate to Structured Streaming.", "Spark 3.4.0")
 object StreamingContext extends Logging {
 
   /**
    * Lock that guards activation of a StreamingContext as well as access to the singleton active
    * StreamingContext in getActiveOrCreate().
+   *
    */
   private val ACTIVATION_LOCK = new Object()
 

streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingContext.scala

@@ -50,7 +50,13 @@ import org.apache.spark.streaming.scheduler.StreamingListener
  * computation can be started and stopped using `context.start()` and `context.stop()`,
  * respectively. `context.awaitTermination()` allows the current thread to wait for the
  * termination of a context by `stop()` or by an exception.
+ * @deprecated This is deprecated as of Spark 3.4.0.
+ *             There are no longer updates to DStream and it's a legacy project.
+ *             There is a newer and easier to use streaming engine
+ *             in Spark called Structured Streaming.
+ *             You should use Spark Structured Streaming for your streaming applications.
  */
+@deprecated("DStream is deprecated. Migrate to Structured Streaming.", "Spark 3.4.0")
 class JavaStreamingContext(val ssc: StreamingContext) extends Closeable {
 
   /**