1.首先用sqoop将MysqL数据定时导入到hdfs中,然后用spark streaming实时读取hdfs的数据,并把数据写入elasticsearch中。代码如下
------bigdata.project.spark----------
package bigdata.project.spark
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.elasticsearch.spark.sql._
object sparkstreamingcopynew {
def main(args: Array[String]): Unit = {
val sparkconf = new SparkConf().setMaster("local[2]").setAppName("sparkstreamingcopynew")
sparkconf.set("es.nodes", "localhost")
sparkconf.set("es.port", "9200")
sparkconf.set("es.index.auto.create", "true")
sparkconf.set("spark.driver.allowMultipleContexts","true")
sparkconf.set("empty", "true")
val sc= new SparkContext(sparkconf)
val ssc = new StreamingContext(sc,Seconds(10))
import org.apache.spark.streaming.Time
val lines = ssc.textFileStream("hdfs://hadoop:9000/ershoufang")
lines.foreachRDD((rdd: RDD[String],time: Time)=> {
val sqlContext = sqlContextSingleton.getInstance(rdd.sparkContext)
import sqlContext.implicits._
val wordsDataFrame = rdd.map(x => (x.split(",")(0), x.split(",")(1), x.split(",")(2),
x.split(",")(3), x.split(",")(4), x.split(",")(5), x.split(",")(6),
x.split(",")(8), x.split(",")(9), x.split(",")(10), x.split(",")(11),x.split(",")(12)))
.map(w => RecordEs(w._1.toInt, w._2, w._3,w._4,w._5,w._6,w._7,w._8,w._9,w._10,w._11,w._12)).toDF()
val dataDS=wordsDataFrame.as[RecordEs]
//val datardd= wordsDataFrame.rdd
Essparksql.savetoEs(dataDS,"zufang/docs")
wordsDataFrame.registerTempTable("RecordEs")
val wordCountsDataFrame =
sqlContext.sql("select id,title,city,huose_type,area,location,direction,price, url,origin,publish_date,true_date from RecordEs")
println(s"========= $time =========")
wordCountsDataFrame.show()
})
ssc.start()
ssc.awaitTermination()
}
}
package bigdata.project.spark
import org.apache.spark.SparkContext
import org.apache.spark.sql.sqlContext
object sqlContextSingleton {
@transient private var instance: sqlContext = _
def getInstance(sparkContext: SparkContext): sqlContext = {
if (instance == null) {
instance = new sqlContext(sparkContext)
}
instance
}
}
package bigdata.project.spark
package bigdata.project.spark
case class RecordEs(id: Int,title: String,city: String,huose_type: String,area:String,location:String,direction:String
,price:String, url:String,origin:String,publish_date:String,true_date:String)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。