Project Upgrade for Spark Streaming. by Ayush21-AI · Pull Request #1 · rockthejvm/spark-streaming

Ayush21-AI · 2023-10-09T20:56:01Z

@daniel-ciocirlan Please review this PR.

Ayush21-AI · 2023-10-09T20:58:18Z

  implicit val actorSystem = ActorSystem("ReceiverSystem", ConfigFactory.load("akkaconfig/remoteActors").getConfig("remoteSystem"))
-  implicit val actorMaterializer = ActorMaterializer()
+  implicit val actorMaterializer = Materializer(actorSystem)



Symbol ActorMaterializer is deprecated. In recent versions of Akka, Materializer.apply(system) is the recommended way to create a system-wide materializer.

Ayush21-AI · 2023-10-09T20:59:06Z

    val sentiments = sentences.map { sentence: CoreMap =>
-      val tree = sentence.get(classOf[SentimentCoreAnnotations.AnnotatedTree])
+      val tree = sentence.get(classOf[SentimentCoreAnnotations.SentimentAnnotatedTree])
      // convert the score to a double for each sentence


AnnotatedTree is no longer available, so used SentimentAnnotatedTree.

Ayush21-AI · 2023-10-09T20:59:46Z

    val kafkaProducer = getProducer()
-    val bindingFuture = Http().bindAndHandle(getRoute(kafkaProducer), "localhost", 9988)
+    val bindingFuture = Http().newServerAt("localhost", 9988).bind(getRoute(kafkaProducer))



Symbol bindAndHandle is deprecated. Used Http().newServerAt(...)...bindFlow() to create server bindings.

Ayush21-AI · 2023-10-09T21:00:10Z

    val kafkaProducer = getProducer()
-    val bindingFuture = Http().bindAndHandle(getRoute(kafkaProducer), "localhost", 9988)
+    val bindingFuture = Http().newServerAt("localhost", 9988).bind(getRoute(kafkaProducer))



Symbol bindAndHandle is deprecated. Used Http().newServerAt(...)...bindFlow() to create server bindings.

Ayush21-AI · 2023-10-09T21:16:54Z

@daniel-ciocirlan In many files, StreamingContext and DStream are both deprecated as of Apache Spark 3.x. They were replaced by structured streaming, which offers improved performance, scalability, and fault tolerance.

val ssc = new StreamingContext(spark.sparkContext, Seconds(1))

It's suggested to Migrate to Structured Streaming.

Correct me if I'm wrong, To migrate our application to use structured streaming, we need to replace StreamingContext object with a SparkSession object, and use the readStream() method to create a DataFrame representing our input data stream.

Do we need to do it ?

Need your guidance.

daniel-ciocirlan

Thanks for this, just some minor comments on lib upgrades

daniel-ciocirlan · 2023-10-11T09:51:50Z

+val sparkVersion = "3.5.0"
+val postgresVersion = "42.6.0"
+val cassandraConnectorVersion = "3.4.1" // preview version at the moment of writing (October, 2023)
+val akkaVersion = "2.8.0"


please use a 2.6 Akka at most, as they changed their license model and we don't want to use non-OSS libraries

daniel-ciocirlan · 2023-10-11T09:52:30Z

+val postgresVersion = "42.6.0"
+val cassandraConnectorVersion = "3.4.1" // preview version at the moment of writing (October, 2023)
+val akkaVersion = "2.8.0"
+val akkaHttpVersion = "10.5.0"


same for this one, I don't remember which was the last Akka HTTP version that was still OSS

daniel-ciocirlan · 2023-10-11T09:52:44Z

  implicit val actorSystem = ActorSystem("ReceiverSystem", ConfigFactory.load("akkaconfig/remoteActors").getConfig("remoteSystem"))
-  implicit val actorMaterializer = ActorMaterializer()
+  implicit val actorMaterializer = Materializer(actorSystem)



daniel-ciocirlan · 2023-10-11T09:53:08Z

    val sentiments = sentences.map { sentence: CoreMap =>
-      val tree = sentence.get(classOf[SentimentCoreAnnotations.AnnotatedTree])
+      val tree = sentence.get(classOf[SentimentCoreAnnotations.SentimentAnnotatedTree])
      // convert the score to a double for each sentence


daniel-ciocirlan · 2023-10-11T09:53:31Z

-          )
-        )
+        complete {
+          Using(Source.fromFile("src/main/html/whackamole.html")) { source =>


nice of you to auto-close the file 👍

daniel-ciocirlan · 2023-10-11T09:53:45Z

    val kafkaProducer = getProducer()
-    val bindingFuture = Http().bindAndHandle(getRoute(kafkaProducer), "localhost", 9988)
+    val bindingFuture = Http().newServerAt("localhost", 9988).bind(getRoute(kafkaProducer))



daniel-ciocirlan · 2023-10-11T10:03:10Z

@Ayush21-AI about DStreams - you can keep it as is for now. Please reply with a +1 if the Twitter project works with this new upgrade.

There is a way for us to read regular rows, but that means a more significant overhaul - we need to register a custom data source that implements the DataSource V2 API. That upgrade will involve some more interesting work in the code, if you're up to it. We can make that a separate PR.

Ayush21-AI · 2023-10-12T21:02:15Z

+val cassandraConnectorVersion = "3.4.1" // preview version at the moment of writing (October, 2023)
+val akkaVersion = "2.6.19"
+val akkaHttpVersion = "10.2.10"
 val twitter4jVersion = "4.0.7"


Updated the akka versions as per the akka licencing.

Ayush21-AI · 2023-10-12T21:11:01Z

@daniel-ciocirlan updated the akka versions. While running the TwitterProject, i'm getting this below error :
23/10/13 02:34:26 ERROR Utils: Exception encountered
java.io.NotSerializableException: Promises and Futures cannot be serialized
at scala.concurrent.impl.Promise$DefaultPromise.writeObject(Promise.scala:373)

This error indicates that we are trying to serialize a Promise or Future object. This is not possible because these objects are not serializable.
I think the issue arises from using Promise[TwitterStream] and Future[TwitterStream] inside the TwitterReceiver class, which extends the Receiver class.

kindly suggest what to do?

daniel-ciocirlan · 2023-10-13T15:16:24Z

I don't understand - can you revert to the original code and check whtether this worked?
We aren't changing anything in the code itself so that would be a bit weird.

Ayush21-AI · 2023-10-16T20:14:28Z

+  connection and reads data from the socket.
+     */
+  var mayBeSocket: Option[Socket] = None



Updated as per

Ayush21-AI · 2023-10-16T20:14:53Z

+  if a Twitter stream has been created.
+   */
+  var mayBeStream: Option[TwitterStream] = None



updated as per your suggestion.

Ayush21-AI commented Oct 9, 2023

View reviewed changes

daniel-ciocirlan reviewed Oct 11, 2023

View reviewed changes

Ayush21-AI force-pushed the Project-upgrade-for-Spark-Streaming branch from 1b068c0 to e0cf1f8 Compare October 12, 2023 20:59

Ayush21-AI commented Oct 12, 2023

View reviewed changes

Ayush21-AI force-pushed the Project-upgrade-for-Spark-Streaming branch from e0cf1f8 to 0496777 Compare October 16, 2023 20:08

Project Upgrade for Spark Streaming.

ca3563b

Ayush21-AI force-pushed the Project-upgrade-for-Spark-Streaming branch from 0496777 to ca3563b Compare October 16, 2023 20:13

Ayush21-AI commented Oct 16, 2023

View reviewed changes

Conversation

Ayush21-AI commented Oct 9, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ayush21-AI commented Oct 9, 2023

Uh oh!

daniel-ciocirlan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

daniel-ciocirlan commented Oct 11, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ayush21-AI commented Oct 12, 2023

Uh oh!

daniel-ciocirlan commented Oct 13, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants