标签：hdfs String val streaming elasticsearch split ._ import spark

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)

标签：hdfs,String,val,streaming,elasticsearch,split,._,import,spark
来源： https://blog.csdn.net/qq_40493738/article/details/91464783

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。