如何将每个输入流的数据集合并为一个

Question

    val ssc = new StreamingContext(sc, Seconds(2))
val sqlContext = new SQLContext(sc)
val lines = ssc.textFileStream("input")

lines.foreachRDD { rdd =>
  val count = rdd.count()
  if (count > 0) {
    val dataSet = sqlContext.read.json(rdd)
    val accountIds = dataSet.select("accountId").distinct.collect.flatMap(_.toSeq)
    val accountIdArry = accountId.map(accountId => dataSet.where($"accountId" <=> accountId))
    accountIdArry.foreach { arrEle =>
      print(arrEle.count)
      arrEle.show
      arrEle.write.format("json").save("output")
    }
  }
}

org.apache.spark.SparkException: Task not serializable

    import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import play.api.libs.json._
import org.apache.spark.sql._
import org.apache.spark.streaming.dstream._


object SparkExample {
    def main(inputDir: String) {
        val ssc = new StreamingContext(sc, Seconds(2))
        val sqlContext = new SQLContext(sc)


        val lines: DStream[String] = ssc.textFileStream(inputDir)

        val jsonLines = lines.map[JsValue](l => Json.parse(l))

        val accountIdLines = jsonLines.map[(String, JsValue)](json => {
            val accountId = (json \ "accountId").as[String]
            (accountId, json)
        })

        val accountIdCounts = accountIdLines
            .map[(String, Long)]({ case (accountId, json) => {
            (accountId, 1)
        } })
        .reduceByKey((a, b) => a + b)


        // this DStream[(String, Long)] will have current accumulated count for accountId's
        val updatedAccountCounts = accountIdCounts
        .updateStateByKey(updatedCountOfAccounts _)
    }

    def updatedCountOfAccounts(a: Seq[Long], b: Option[Long]): Option[Long] = {
        b.map(i => i + a.sum).orElse(Some(a.sum))
    }
}

Answer 1

有两件事你需要记住。

首先，由于您使用的是StreamingContext带有2秒微批处理，因此DStreams将包含RDD，其中只有在这2秒内生成的数据，而不是所有数据。如果您需要对当时所有可用的数据执行操作，那么流就不适合您的问题。

第二，您不需要使用sql上下文来处理JSON。只需使用任何json库并将rdd分组在AccountID上即可。

import play.api.libs.json._

val ssc = new StreamingContext(sc, Seconds(2))
val sqlContext = new SQLContext(sc)
val dstreams = ssc.textFileStream("input")


dstreams.foreachRDD { rdd =>
  val jsonRdd = rdd.map(l => Json.parse(l))
  val grouped = jsonRdd.groupBy(json => (json \ "accountId").as[String])
}

import play.api.libs.json._

val ssc = new StreamingContext(sc, Seconds(2))
val sqlContext = new SQLContext(sc)


val lines: DStream[String] = ssc.textFileStream("inputPath")

val jsonLines = lines.map[JsValue](l => Json.parse(l))

val accountIdLines = jsonLines.map[(String, JsValue)](json => {
  val accountId = (json \ "accountId").as[String]
  (accountId, json)
})

val accounIdCounts = accountIdLines
  .map[(String, Long)]({ case (accountId, json) => {
    (accountId, 1)
  } })
  .reduceByKey((a, b) => a + b)


// this DStream[(String, Long)] will have current accumulated count for accountId's
val updatedAccountCounts = accountIdCounts
  .updateStateByKey(updateCountOfAccounts _)

def updatedCountOfAccounts(a: Seq[Long], b: Option[Long]): Option[Long] = {
  b.map(i => i + a.sum).orElse(Some(a.sum))
}