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

Question

如何在应用转换之前将来自每个 spark 输入流的数据集合并为一个。我正在使用 spark-2.0.0

    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")
    }
  }
}

我想通过考虑所有输入流将每个 accountId 计数大于 100000 的记录写入输出文件。为此，我想在执行转换之前将所有 DStream 合并为一个。

现在它将所有记录写入输出文件。有什么帮助吗？

更新

org.apache.spark.SparkException: Task not serializable

在 org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298) 在 org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288) 在 org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108) 在 org.apache.spark.SparkContext.clean(SparkContext.scala:2037) 在 org.apache.spark.streaming.dstream.PairDStreamFunctions$$anonfun$updateStateByKey$3.apply(PairDStreamFunctions.scala:433) 在 org.apache.spark.streaming.dstream.PairDStreamFunctions$$anonfun$updateStateByKey$3.apply(PairDStreamFunctions.scala:432) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) 在 org.apache.spark.SparkContext.withScope(SparkContext.scala:682) 在 org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264) 在 org.apache.spark.streaming.dstream.PairDStreamFunctions.updateStateByKey(PairDStreamFunctions.scala:432) 在 org.apache.spark.streaming.dstream.PairDStreamFunctions$$anonfun$updateStateByKey$1.apply(PairDStreamFunctions.scala:400) 在 org.apache.spark.streaming.dstream.PairDStreamFunctions$$anonfun$updateStateByKey$1.apply(PairDStreamFunctions.scala:400) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) 在 org.apache.spark.SparkContext.withScope(SparkContext.scala:682) 在 org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264) 在 org.apache.spark.streaming.dstream.PairDStreamFunctions.updateStateByKey(PairDStreamFunctions.scala:399) 在 SparkExample$.main(:60) ... 56 省略 引起：java.io.NotSerializableException：SparkExample$ 序列化栈： - 对象不可序列化（类：SparkExample$，值：SparkExample$@ab3b54） - 字段（类：SparkExample$$anonfun$5，名称：$outer，类型：类 SparkExample$） - 对象（类 SparkExample$$anonfun$5, ） 在 org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40) 在 org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46) 在 org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100) 在 org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:295) ... 74 更多

SparkExample.scala

    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 seconds 微批处理，因此您的 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])
}

如果您想使用updateStateByKey，那么请继续使用DStreams，

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))
}