Spark Streaming 过滤流数据答案

【问题标题】：Spark Streaming Filtering the Streaming dataSpark Streaming 过滤流数据
【发布时间】：2016-12-13 04:18:41
【问题描述】：

我正在尝试过滤流数据，并根据 id 列的值将数据保存到不同的表中

我有两张桌子

testTable_odd (id,data1,data2)
testTable_even (id,data1)

如果 id 值是奇数，那么我想将记录保存到 testTable_odd 表，如果值是偶数，那么我想将记录保存到 testTable_even。

这里的棘手部分是我的两个表有不同的列。尝试了多种方法，考虑了返回类型为 Either[obj1,obj2] 的 Scala 函数，但我无法成功，任何指针都将不胜感激。

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SaveMode
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.kafka.KafkaUtils
import com.datastax.spark.connector._

import kafka.serializer.StringDecoder
import org.apache.spark.rdd.RDD
import com.datastax.spark.connector.SomeColumns
import java.util.Formatter.DateTime

object StreamProcessor extends Serializable {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("StreamProcessor")
      .set("spark.cassandra.connection.host", "127.0.0.1")

    val sc = new SparkContext(sparkConf)

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

    val sqlContext = new SQLContext(sc)

    val kafkaParams = Map("metadata.broker.list" -> "localhost:9092")

    val topics = args.toSet

    val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc, kafkaParams, topics)


        stream
  .map { 
    case (_, msg) => 
      val result = msgParseMaster(msg)
      (result.id, result.data)
   }.foreachRDD(rdd => if (!rdd.isEmpty)     rdd.saveToCassandra("testKS","testTable",SomeColumns("id","data")))

      }
    }

    ssc.start()
    ssc.awaitTermination()

  }

  import org.json4s._
  import org.json4s.native.JsonMethods._
  case class wordCount(id: Long, data1: String, data2: String) extends serializable
  implicit val formats = DefaultFormats
  def msgParseMaster(msg: String): wordCount = {
    val m = parse(msg).extract[wordCount]
    return m

  }

}

【问题讨论】：

标签： apache-spark spark-streaming spark-cassandra-connector

【解决方案1】：

我认为您只想使用过滤器功能两次。你可以做类似的事情

val evenstream = stream.map { case (_, msg) => 
  val result = msgParseMaster(msg)
  (result.id, result.data)
}.filter{ k =>
  k._1 % 2 == 0
}

evenstream.foreachRDD{rdd=>
  //Do something with even stream
}

val oddstream = stream.map { case (_, msg) => 
  val result = msgParseMaster(msg)
  (result.id, result.data)
}.filter{ k =>
  k._1 % 2 == 1
}

oddstream.foreachRDD{rdd=>
  //Do something with odd stream
}

当我在一个项目here 上做了类似的事情时，如果你在第 191 行附近向下看，我使用了两次过滤器功能。我正在根据元组在 0 和 1 之间的值对它们进行分类和保存，所以请随意检查一下。

【讨论】：

谢谢。我通过一次过滤以不同的方式解决了它。
嘿，您介意发布您的解决方案，以便将其标记为正确答案吗？我也想看看：P
今天肯定会这样做。只是列出过程。我已经执行了以下步骤。 1) 从原始 JSON 字符串中提取详细信息并使用案例类 2) 创建超级 JSON（其中包含两个过滤条件所需的详细信息） 3) 将该 JSON 转换为 DataFrame 4) 对该 JSON 执行 select 和 where 子句 5)保存到 cassandra

【解决方案2】：

我已经执行了以下步骤。 1) 从原始 JSON 字符串和案例类中提取详细信息 2) 创建了超级 JSON（其中包含两个过滤条件所需的详细信息） 3) 将该 JSON 转换为 DataFrame 4) 对该 JSON 执行 select 和 where 子句 5) 保存到 Cassandra

【讨论】：