spark 写入 es

spark 2.4

es 7.10.2

Scala2.11  

<dependency>

            <groupId>org.elasticsearch</groupId>

            <artifactId>elasticsearch-hadoop</artifactId>

            <version>7.10.2</version>

        </dependency>

        <dependency>

            <groupId>org.elasticsearch</groupId>

            <artifactId>elasticsearch-spark-20_2.11</artifactId>

            <version>7.10.0</version>

        </dependency>

        <dependency>

            <groupId>io.searchbox</groupId>

            <artifactId>jest</artifactId>

            <version>6.3.1</version>

        </dependency>

        <dependency>

            <groupId>org.projectlombok</groupId>

            <artifactId>lombok</artifactId>

            <version>1.16.10</version>

        </dependency>

        <dependency>

            <groupId>commons-logging</groupId>

            <artifactId>commons-logging</artifactId>

            <version>1.1.1</version>

        </dependency>

        <dependency>

            <groupId>commons-codec</groupId>

            <artifactId>commons-codec</artifactId>

            <version>1.4</version>

        </dependency>

        <dependency>

            <groupId>commons-httpclient</groupId>

            <artifactId>commons-httpclient</artifactId>

            <version>3.0.1</version>

        </dependency>

================

package com

import com.constant.PropConstants

import com.javaUtil.PropertieUtil

import io.searchbox.client.JestClient

import io.searchbox.core.{Bulk, BulkResult, Index}

import org.apache.hadoop.security.UserGroupInformation

import org.apache.log4j.Logger

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.functions.lit

import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession}

import org.elasticsearch.spark.sql.EsSparkSQL

import java.sql.Timestamp

import java.text.SimpleDateFormat

import java.util

import java.util.{Date, Properties}

object center_materiel_hotspot2es {

  def main(args: Array[String]): Unit = {

    val log: Logger = Logger.getRootLogger

    //读取集群配置文件

    val prop: Properties = PropertieUtil.load("config.properties")

    //本地测试读文件

//    val prop: Properties = PropertieUtil.getProperties("/config.properties")

    //读hive 的Kerberos认证

    System.setProperty("java.security.krb5.conf", prop.getProperty(PropConstants.KRB5_CONF_PATH))

    System.setProperty("HADOOP_USER_NAME", prop.getProperty(PropConstants.HADOOP_USER_NAME))

    System.setProperty("user.name", prop.getProperty(PropConstants.USER_NAME))

    UserGroupInformation.loginUserFromKeytab(

      prop.getProperty(PropConstants.KEYTAB_NAME), prop.getProperty(PropConstants.KEYTAB_FILE_PATH)

    )

    val session: SparkSession = SparkSession.builder()

//      .master("local[9]")

//      .config("spark.testing.memory","4718592000")

//      .appName("spark to es")

//      .config("spark.yarn.am.waitTime", "1000")

      .config("spark.hadoop.hive.exec.dynamic.partition", "true")//开启动态分区

      .config("spark.hadoop.hive.exec.dynamic.partition.mode", "nonstrict")//开启动态分区

      .enableHiveSupport() //支持hive

      .getOrCreate()

    import session.implicits._

    //首先处理

    val dataFrame: DataFrame = session.sql(

      """

        |select id,create_by,create_at,update_by,update_at,position_top,position_left,materiel_group_child_id,

        |drawing_no,width,height,drawing_type from

        | dws.center_materiel_hotspot

        |""".stripMargin)

    val rdd = dataFrame.map(x => {

      val id: String = x.getAs(0)

      val create_by: String = x.getAs(1)

      val create_at: String = x.getAs(2).toString

      val update_by: String = x.getAs(3)

      val update_at: String = x.getAs(4).toString

      val position_top: String = if(x.getAs(5)!=null) x.getAs(5).toString else x.getAs(5)

      val position_left: String = if(x.getAs(6)!=null) x.getAs(6).toString else x.getAs(6)

      val materiel_group_child_id: String = x.getAs(7)

      val drawing_no: String = x.getAs(8)

      val width: String = if(x.getAs(9)!=null) x.getAs(9).toString else x.getAs(9)

      val height: String = if(x.getAs(10)!=null) x.getAs(10).toString else x.getAs(10)

      val drawing_type: String = x.getAs(11)

      val createTime: Long = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(create_at).getTime

      val updateTime: Long = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(update_at).getTime

      center_materiel_hotspot_local(id, create_by, createTime, update_by, updateTime, position_top,position_left,

        materiel_group_child_id,drawing_no,width,height,drawing_type)

    })

//    rdd.show(2)

    rdd.foreachPartition(x=>{

      val tuples: List[(String, center_materiel_hotspot_local)] = x.toList.map { data => (data.id, data) }

      //tuples 数据，参二：索引名称  ,参三：类型

      val dt: String = new SimpleDateFormat("yyyyMMdd").format(new Date())

      savaBulk(tuples,"center_materiel_hotspot_local","_doc")

    })

    session.stop()

  }

  case class center_materiel_hotspot_local(id:String, createBy:String, createTime:Long, updateBy:String,

                                           updateTime:Long,positionTop:String, positionLeft:String,

                                           materielGroupChildId:String,drawingNo:String, width:String,

                                           height:String, drawingType:String )

  def savaBulk(dataList:List[(String,AnyRef)],indexName:String,typeName:String):Unit={

    if (dataList!=null && dataList.nonEmpty) {

      val client: JestClient = jestClient

      val builder = new Bulk.Builder

      builder.defaultIndex(indexName).defaultType(typeName)

      for ((id, data) <- dataList) {

        val index: Index = new Index.Builder(data).id(id).build()

        builder.addAction(index)

      }

      val bulk: Bulk = builder.build()

      //获取执行的返回值

      val items: util.List[BulkResult#BulkResultItem] = client.execute(bulk).getItems

      println("以保存：" + items.size() + "条数据！")

      client.close()

    }

  }

  import io.searchbox.client.JestClient

  import io.searchbox.client.JestClientFactory

  import io.searchbox.client.config.HttpClientConfig

  def jestClient: JestClient = {

    val factory = new JestClientFactory

    factory.setHttpClientConfig(new HttpClientConfig.Builder(

      "http://es-cn-tl32b5q0t003a6n6g.public.elasticsearch.aliyuncs.com:9200")

      .multiThreaded(true)

      .defaultMaxTotalConnectionPerRoute(2)

      .maxTotalConnection(10)

      .build)

    factory.getObject

  }

}