一、寫(xiě)在前面

ElasticSearch 是一個(gè)快速索引檢索的庫(kù)。在實(shí)踐中，我們用Hbase 存儲(chǔ)海量業(yè)務(wù)數(shù)據(jù)，再通過(guò)ES存儲(chǔ)索引，以這種相互結(jié)合的方式，將數(shù)據(jù)暴露給Web服務(wù)端做海量數(shù)據(jù)的查詢(xún)。
實(shí)際項(xiàng)目中遇到的問(wèn)題是：

1. Hadoop 平臺(tái)采用的JDK版本為 1.7， ES的JDK版本為1.8
2. 需要頻繁大批量的初始化數(shù)據(jù),每次大約200G，要求在幾個(gè)小時(shí)內(nèi)導(dǎo)入
3. 不能接受數(shù)據(jù)丟失

二、解決思路

ElasticSearch本身提供有ElasticSearch-hadoop 插件，當(dāng)由于JDK版本不同，該插件不可用。因此選擇Jest
一種Rest方式訪問(wèn)ES。
另外，為保證數(shù)據(jù)導(dǎo)入速度、成功率，對(duì)導(dǎo)入程序做以下改進(jìn)

1. Spark對(duì)大文件拆分（200G拆分為200個(gè)文件）
2. 監(jiān)控每個(gè)文件導(dǎo)入的日志
3. 使用Bulk模式導(dǎo)入,每個(gè)分區(qū)做一次提交
4. 壓測(cè)網(wǎng)絡(luò)傳輸和ES集群能接受的一次Bulk 數(shù)據(jù)量的峰值
5. 壓測(cè)Spark運(yùn)行的核數(shù)，節(jié)點(diǎn)數(shù)（連接起的過(guò)多會(huì)導(dǎo)致ES CPU占用率超過(guò)90%，進(jìn)而降低導(dǎo)入速率）

三、代碼實(shí)現(xiàn)

1. 連接工廠

package org.hhl.esETL.es
import com.google.gson.GsonBuilder
import io.searchbox.client.JestClientFactory
import io.searchbox.client.config.HttpClientConfig
/**
  * Created by huanghl4 on 2017/11/15.
  */
object esConnFactory extends Serializable{
  @transient private var factory: JestClientFactory = null
  def getESFactory(): JestClientFactory = {
    //設(shè)置連接ES
    if (factory == null) {
      factory = new JestClientFactory()
      factory.setHttpClientConfig(new HttpClientConfig.Builder("http://10.120.193.9:9200")
        .addServer("http://10.120.193.10:9200").addServer("http://10.120.193.26:9200")
        .maxTotalConnection(20)//.defaultMaxTotalConnectionPerRoute(10)
        .gson(new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss") create())
        .readTimeout(100000)
        .build())
    }
    factory
  }
}

2. 大文件拆分
   大文件拆分，可用randomsplit 或者repatition 之后，存成parquet.

方法一，使用Spark 拆分為Parquet,再讀HDFS文件

val data = spark.sql("select id,json from hive.table")
    data.repartition(200).write.parquet("/data")
// 返回路徑下文件列表
def getPathFileNameList(sc: SparkContext, path: String): List[String] = {
    val hdfs = org.apache.hadoop.fs.FileSystem.get(sc.hadoopConfiguration)
    val hdfsPath = new org.apache.hadoop.fs.Path(path)
    val listBuff = new ListBuffer[String]

    if (hdfs.exists(hdfsPath)) {
      val it =  hdfs.listFiles(hdfsPath,false)
      while(it.hasNext){
        val f = it.next().getPath.getName
        if (f.startsWith("part")) listBuff.append(f)
      }
    }
    listBuff.toList
  }
// 讀取文件
val fileList = HdfsFileUntil.getPathFileNameList(spark.sparkContext, userGraphPath)
    for (fileName <- fileList){
      val filePath = userGraphPath + "/" + fileName
// 保存到ES
      saveToES(spark, filePath)
    }

方法二，randomSplit 拆分

// 拆分
def splitTable(df:DataFrame,prefix:String,num:Int) =  {
    val weight = new Array[Double](num)
    val average = 1 / num
    for (i <- 1 until  weight.size) weight(i) = average
    val splitDF = df.randomSplit(weight)
    for(i<- 0 until splitDF.size) splitDF(i).write.mode(SaveMode.Overwrite).saveAsTable(s"$prefix" +"_"+ i)
  }
// 讀取
for(I<- 0 to num-1)   {
val df = spark.read.table(s"$prefix" +"_"+ i)
saveToES(df)
}

3. 存儲(chǔ)到ES

private def saveToES(rdd: RDD[(String, String)], repartitions: Int): Unit = {
    rdd.repartition(repartitions).foreachPartition(x => {
      val client = EsUtil.getESFactory().getObject()
      val bulk = new Bulk.Builder().defaultIndex(USER_GRAPH_INDEX).defaultType(USER_GRAPH_TYPE)
      x.foreach(msg => {
        val index = new Index.Builder(msg._2).id(msg._1).build()
        bulk.addAction(index)
      })
      try {
        client.execute(bulk.build())
      } catch {
        case e: Exception => {
          Thread.sleep(10000)
          client.execute(bulk.build())
        }
      }
    })
    rdd.unpersist(true)
  }

參考：https://github.com/Smallhi/ElasticSearchETL