如何从SocketTCP获取数据以保存到Spark Scala中的数据帧？

Question

我尝试从Socket TCP获取数据以附加到数据帧我收到数据并将它们执行到Seq（），但当我使用foreach将它们附加到数据帧时出现问题这是我的代码：

object CustomReceiver {
  def main(args: Array[String]): Unit = {
    StreamingExamples.setStreamingLogLevels()

    // Create the context with a 1 second batch size
    val spark: SparkSession = SparkSession.builder()
      .master("local[2]")
      .appName("CustomReceiver")
      .getOrCreate()

    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(5))
    import spark.implicits._

   /*formatdata line data from Socket:  number1, 20210621090303, RadiusMessage, Stop, 84602496347, v241.66.85.130 */

    val linesData1 = ssc.receiverStream(new CustomReceiver("localhost", 11000))
    linesData1.flatMap(_.split(" ").map(_.trim))
    linesData1.foreachRDD { rdd =>
      rdd.foreach{ line => {

        val  arrraLine = line.split(",").toList
       // oke arrayLine data : List(number1, 20210621090303, RadiusMessage, Stop, 84602496347, 241.66.85.130)
        val testRDD = Seq(arrraLine).map(x =>(x(0), x(1), x(2), x(3), x(4)))
        // oke TestRDD : testRDD :List((number1,20210621090303,RadiusMessage,Stop,84602496347))

          val testDF = testRDD.toDF("cot1","cot2","cot3","cot4","cot5")
         // has an Problem
          testDF.show()
      }
      }
    }
       ssc.start()
        ssc.awaitTermination()
      }
    }

这是我跑步时的问题

Java语言组织中的lang.NullPointerException。阿帕奇。火花sql。SQLImplicits。localSeqToDatasetHolder（SQLImplicits.scala:231），位于Cl.CustomReceiver$$anonfun$main$4（CustomerReceiver.scala:52），Cl.CustomReceiver$$anonfun$main$4$adapted（CustomerReceiver.scala:45）位于scala。收集迭代器。scala上的foreach（迭代器。scala:943）。收集迭代器。foreach$（Iterator.scala:943）位于org。阿帕奇。火花util。完成计算器。foreach（CompletionIterator.scala:25）位于org。阿帕奇。火花rdd。RDD$anonfun$foreach$2（RDD.scala:1012）位于org。阿帕奇。火花rdd。RDD$anonfun$foreach$2$adapted（RDD.scala:1012）位于org。阿帕奇。火花SparkContext$org上的anonfun$runJob$5（SparkContext.scala:2236）。阿帕奇。火花调度程序。结果任务。在组织上运行任务（ResultTask.scala:90）。阿帕奇。火花调度程序。任务在组织上运行（Task.scala:131）。阿帕奇。火花执行人。执行者$TaskRunner$anonfun$run$3（Executor.scala:497）位于org。阿帕奇。火花util。Utils美元。tryWithSafeFinally（Utils.scala:1439）位于org。阿帕奇。火花执行人。执行者$TaskRunner。在java上运行（Executor.scala:500）。util。同时发生的线程池执行器。java上的runWorker（ThreadPoolExecutor.java:1149）。util。同时发生的ThreadPoolExecutor$工作者。在java上运行（ThreadPoolExecutor.java:624）。lang.Thread。运行（Thread.java:748）

Answer 1

你想做的事不太值得推荐。如果要使用数据帧，请使用Spark结构化流。您正在尝试在foreach RDD操作中创建DF。如果您使用旧的Spark流媒体版本，请使用RDD。

您可以在“foreachRDD”中创建一个DF，为每个小批次创建一个RDD，但这不是一个好主意。如果您进行测试，您将看到许多火花阶段来创建新的DF，并且对于每个小批次...使用结构化流，您可以直接创建数据帧。