一孕蝉、寫在前面
ElasticSearch 是一個(gè)快速索引檢索的庫。在實(shí)踐中掩驱，我們用Hbase 存儲海量業(yè)務(wù)數(shù)據(jù)，再通過ES存儲索引，以這種相互結(jié)合的方式欧穴，將數(shù)據(jù)暴露給Web服務(wù)端做海量數(shù)據(jù)的查詢民逼。
實(shí)際項(xiàng)目中遇到的問題是：
- Hadoop 平臺采用的JDK版本為 1.7， ES的JDK版本為1.8
- 需要頻繁大批量的初始化數(shù)據(jù),每次大約200G涮帘，要求在幾個(gè)小時(shí)內(nèi)導(dǎo)入
- 不能接受數(shù)據(jù)丟失
二拼苍、解決思路
ElasticSearch本身提供有ElasticSearch-hadoop 插件，當(dāng)由于JDK版本不同调缨，該插件不可用疮鲫。因此選擇Jest
一種Rest方式訪問ES。
另外同蜻，為保證數(shù)據(jù)導(dǎo)入速度棚点、成功率，對導(dǎo)入程序做以下改進(jìn)
- Spark對大文件拆分（200G拆分為200個(gè)文件）
- 監(jiān)控每個(gè)文件導(dǎo)入的日志
- 使用Bulk模式導(dǎo)入,每個(gè)分區(qū)做一次提交
- 壓測網(wǎng)絡(luò)傳輸和ES集群能接受的一次Bulk 數(shù)據(jù)量的峰值
- 壓測Spark運(yùn)行的核數(shù)湾蔓，節(jié)點(diǎn)數(shù)（連接起的過多會導(dǎo)致ES CPU占用率超過90%瘫析，進(jìn)而降低導(dǎo)入速率）
三、代碼實(shí)現(xiàn)
- 連接工廠
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
}
}
- 大文件拆分
大文件拆分默责，可用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)
}
- 存儲到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)
}
