我尝试使用sparksql查询来自kafka的数据,使用zeppelin进行实时趋势分析,但没有成功。
下面是我在齐柏林飞艇中运行的简单代码片段
//Load Dependency
%dep
z.reset()
z.addRepo("Spark Packages Repo").url("http://repo1.maven.org/maven2/")
z.load("org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.1")
z.load("org.apache.spark:spark-core_2.11:2.0.1")
z.load("org.apache.spark:spark-sql_2.11:2.0.1")
z.load("org.apache.spark:spark-streaming_2.11:2.0.1"
//simple streaming
%spark
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import _root_.kafka.serializer.StringDecoder
import org.apache.spark.sql.SparkSession
val conf = new SparkConf()
.setAppName("clickstream")
.setMaster("local[*]")
.set("spark.streaming.stopGracefullyOnShutdown", "true")
.set("spark.driver.allowMultipleContexts","true")
val spark = SparkSession
.builder()
.appName("Spark SQL basic example")
.config(conf)
.getOrCreate()
val ssc = new StreamingContext(conf, Seconds(1))
val topicsSet = Set("timer")
val kafkaParams = Map[String, String]("metadata.broker.list" -> "192.168.25.1:9091,192.168.25.1:9092,192.168.25.1:9093")
val lines = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSet).map(_._2)
lines.window(Seconds(60)).foreachRDD{ rdd =>
val clickDF = spark.read.json(rdd) //doesn't have to be json
clickDF.createOrReplaceTempView("testjson1")
//olderway
//clickDF.registerTempTable("testjson2")
clickDF.show
}
lines.print()
ssc.start()
ssc.awaitTermination()
我可以打印每个kafka消息,但是当我运行简单的sql时 %sql select * from testjson1 // or testjson2
,我得到以下错误
java.util.NoSuchElementException: None.get
at scala.None$.get(Option.scala:347)
at scala.None$.get(Option.scala:345)
at org.apache.spark.storage.BlockInfoManager.releaseAllLocksForTask(BlockInfoManager.scala:343)
at org.apache.spark.storage.BlockManager.releaseAllLocksForTask(BlockManager.scala:646)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:281)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
在本文中,正在查询流式处理后的数据(以twitter为例)。所以我认为Kafka流媒体应该是可能的。所以我想,也许,我做错了什么,或者遗漏了什么?
欢迎提出任何意见、建议、建议
1条答案
按热度按时间enxuqcxy1#
错误消息不会说明缺少临时视图。错误消息表明,类型none不提供名为“get”的元素。
使用spark,在调用某个操作时,将执行基于rdd的计算。因此,在创建临时表之前,不会执行任何计算。所有的计算都是在表上执行查询时执行的。如果您的表不存在,您将收到另一条错误消息。
也许可以打印kafka消息,但是异常告诉您,none示例不知道“get”。因此,我认为源json数据包含没有数据的项,而这些项用none表示,因此在spark执行计算时会导致execption。
我建议您通过测试不包含空json元素的示例数据来验证您的解决方案是否正常工作。