我要评论一、添加依赖
thisbuild / version := "0.1.0-snapshot"
thisbuild / scalaversion := "2.12.12"
librarydependencies ++= seq( "org.apache.spark" % "spark-core_2.12" % "3.0.0")
librarydependencies ++= seq( "org.apache.spark" % "spark-sql_2.12" % "3.0.0")
librarydependencies ++= seq( "org.apache.spark" % "spark-streaming_2.12" % "3.0.0")
librarydependencies ++= seq( "org.apache.spark" % "spark-streaming-kafka-0-10_2.12" % "3.0.0")
librarydependencies ++= seq( "com.fasterxml.jackson.core" % "jackson-core" % "2.10.1")
librarydependencies ++= seq( "mysql" % "mysql-connector-java" % "5.1.30")
lazy val root = (project in file("."))
.settings(
name := "scala-proj"
)
二、demo程序
package example3
import org.apache.kafka.clients.consumer.{consumerconfig, consumerrecord}
import org.apache.spark.sparkconf
import org.apache.spark.rdd.rdd
import org.apache.spark.storage.storagelevel
import org.apache.spark.streaming.dstream.{dstream, inputdstream, receiverinputdstream}
import org.apache.spark.streaming.kafka010.{consumerstrategies, kafkautils, locationstrategies}
import org.apache.spark.streaming.receiver.receiver
import org.apache.spark.streaming.{duration, seconds, streamingcontext}
import scala.collection.mutable
import scala.util.random
object hellostreaming04 {
def main(args: array[string]): unit = {
val sparkconf = new sparkconf().setmaster("local[*]").setappname("kafkaconsumer")
val ssc = new streamingcontext(sparkconf, seconds(5))
val kafkapara: map[string, object] = map[string, object](
consumerconfig.bootstrap_servers_config -> "xx.xx.xx.xx:9092",
consumerconfig.group_id_config -> "mygroup",
"key.deserializer" -> "org.apache.kafka.common.serialization.stringdeserializer",
"value.deserializer" -> "org.apache.kafka.common.serialization.stringdeserializer"
)
val kafkadatads: inputdstream[consumerrecord[string, string]] = kafkautils.createdirectstream[string, string](ssc,
locationstrategies.preferconsistent,
consumerstrategies.subscribe[string, string](set("mytopic"), kafkapara)
)
kafkadatads.map(_.value()).print()
ssc.start();
ssc.awaittermination();
}
}
发表评论