SparkStreaming 和 Kafka：值 reduceByKey 不是 org.apache.spark.streaming.dstream.DStream[Any] 的成员答案

【问题标题】：SparkStreaming & Kafka: value reduceByKey is not a member of org.apache.spark.streaming.dstream.DStream[Any]SparkStreaming 和 Kafka：值 reduceByKey 不是 org.apache.spark.streaming.dstream.DStream[Any] 的成员
【发布时间】：2018-01-14 17:13:32
【问题描述】：

我尝试使用 Kafka Consumer 和 SparkStreaming 对 DStream 进行 ETL，但出现以下错误。你能帮我解决这个问题吗？谢谢。

KafkaCardCount.scala:56:28: value reduceByKey is not a member of org.apache.spark.streaming.dstream.DStream[Any]
[error]       val wordCounts = etl.reduceByKey(_ + _)
[error]                            ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed
[error] Total time: 7 s, completed Jan 14, 2018 2:52:23 PM

我有这个示例代码。我发现很多文章建议添加import import org.apache.spark.streaming.StreamingContext._，但它似乎对我不起作用。

package example

import org.apache.spark.streaming.StreamingContext._
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Durations, StreamingContext}

val ssc = new StreamingContext(sparkConf, Durations.seconds(5))

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

val etl = stream.map(r => {
    val split = r.value.split("\t")
    val id = split(1)
    val numStr = split(4)
    if (numStr.matches("\\d+")) {
        val num = numStr.toInt
        val tpl = (id, num)
        tpl
    } else {
        ()
    }
})

// Create the counts per game
val wordCounts = etl.reduceByKey(_ + _)

wordCounts.print()

我有这个 build.sbt。

lazy val root = (project in file(".")).
  settings(
    inThisBuild(List(
      organization := "example",
      scalaVersion := "2.11.8",
      version      := "0.1.0-SNAPSHOT"
    )),
    name := "KafkaCardCount",
    libraryDependencies ++= Seq (
      "org.apache.spark" %% "spark-core" % "2.1.0",
      "org.apache.spark" % "spark-streaming_2.11" % "2.1.0",
      "org.apache.spark" %% "spark-streaming-kafka-0-10-assembly" % "2.1.0"
    )
  )

assemblyMergeStrategy in assembly := {
  case PathList("META-INF", xs @ _*) => MergeStrategy.discard
  case x => MergeStrategy.first
}

【问题讨论】：

标签： scala apache-spark apache-kafka spark-streaming

【解决方案1】：

你的问题在这里：

else {
    ()
}

(String, Int) 和 Unit 的通用超类型是 Any。

您需要做的是使用与您的成功 (if) 子句类似的类型发出处理失败的信号。例如：

else ("-1", -1)
 .filter { case (id, res) => id != "-1" && res != -1 }
 .reduceByKey(_ + _)

【讨论】：

谢谢！修好了。