У меня есть базовый код Spark - Kafka, я пытаюсь запустить следующий код:
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.storage.StorageLevel
import java.util.regex.Pattern
import java.util.regex.Matcher
import org.apache.spark.streaming.kafka._
import kafka.serializer.StringDecoder
import Utilities._
object WordCount {
def main(args: Array[String]): Unit = {
val ssc = new StreamingContext("local[*]", "KafkaExample", Seconds(1))
setupLogging()
// Construct a regular expression (regex) to extract fields from raw Apache log lines
val pattern = apacheLogPattern()
// hostname:port for Kafka brokers, not Zookeeper
val kafkaParams = Map("metadata.broker.list" -> "localhost:9092")
// List of topics you want to listen for from Kafka
val topics = List("testLogs").toSet
// Create our Kafka stream, which will contain (topic,message) pairs. We tack a
// map(_._2) at the end in order to only get the messages, which contain individual
// lines of data.
val lines = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topics).map(_._2)
// Extract the request field from each log line
val requests = lines.map(x => {val matcher:Matcher = pattern.matcher(x); if (matcher.matches()) matcher.group(5)})
// Extract the URL from the request
val urls = requests.map(x => {val arr = x.toString().split(" "); if (arr.size == 3) arr(1) else "[error]"})
// Reduce by URL over a 5-minute window sliding every second
val urlCounts = urls.map(x => (x, 1)).reduceByKeyAndWindow(_ + _, _ - _, Seconds(300), Seconds(1))
// Sort and print the results
val sortedResults = urlCounts.transform(rdd => rdd.sortBy(x => x._2, false))
sortedResults.print()
// Kick it off
ssc.checkpoint("/home/")
ssc.start()
ssc.awaitTermination()
}
}
Я использую IntelliJ IDE и создаю проект scala с помощью sbt.Подробная информация о файле build.sbt:
name := "Sample"
version := "1.0"
organization := "com.sundogsoftware"
scalaVersion := "2.11.8"
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-core" % "2.2.0" % "provided",
"org.apache.spark" %% "spark-streaming" % "1.4.1",
"org.apache.spark" %% "spark-streaming-kafka" % "1.4.1",
"org.apache.hadoop" % "hadoop-hdfs" % "2.6.0"
)
Однако, когда я пытаюсь построить код, он создает следующую ошибку:
Ошибка: scalac: при обнаружении отсутствует или недопустимая зависимостьзагрузка файла класса 'StreamingContext.class'.Не удалось получить доступ к типу Logging в пакете org.apache.spark, потому что он (или его зависимости) отсутствуют.Проверьте определение сборки на наличие отсутствующих или конфликтующих зависимостей.(Перезапустите с -Ylog-classpath, чтобы увидеть проблемный путь к классу.) Полная перестройка может помочь, если 'StreamingContext.class' был скомпилирован с несовместимой версией org.apache.spark.
Ошибка: scalac: отсутствуетили при загрузке файла класса 'DStream.class' обнаружена неверная зависимость.Не удалось получить доступ к типу Logging в пакете org.apache.spark, потому что он (или его зависимости) отсутствуют.Проверьте определение сборки на наличие отсутствующих или конфликтующих зависимостей.(Повторно запустите с -Ylog-classpath, чтобы увидеть проблемный путь к классу.) Полная перестройка может помочь, если DStream.class был скомпилирован с несовместимой версией org.apache.spark.