如何使用Spark大规模并行构建索引

使用Spark构建索引非常简单,因为spark提供了更高级的抽象rdd分布式弹性数据集,相比以前的使用Hadoop的MapReduce来构建大规模索引,Spark具有更灵活的api操作,性能更高,语法更简洁等一系列优点。 先看下,整体的拓扑图:

然后,再来看下,使用scala写的spark程序:

Java代码

  1. package com.easy.build.index
  2. import java.util
  3. import org.apache.solr.client.solrj.beans.Field
  4. import org.apache.solr.client.solrj.impl.HttpSolrClient
  5. import org.apache.spark.rdd.RDD
  6. import org.apache.spark.{SparkConf, SparkContext}
  7. import scala.annotation.meta.field
  8. /**
  9. * Created by qindongliang on 2016/1/21.
  10. */
  11. //注册model,时间类型可以为字符串,只要后台索引配置为Long即可,注解映射形式如下
  12. case class Record(
  13. @(Field@field)("rowkey") rowkey:String,
  14. @(Field@field)("title") title:String,
  15. @(Field@field)("content") content:String,
  16. @(Field@field)("isdel") isdel:String,
  17. @(Field@field)("t1") t1:String,
  18. @(Field@field)("t2")t2:String,
  19. @(Field@field)("t3")t3:String,
  20. @(Field@field)("dtime") dtime:String
  21. )
  22. /***
  23. * Spark构建索引==>Solr
  24. */
  25. object SparkIndex {
  26. //solr客户端
  27. val client=new HttpSolrClient("http://192.168.1.188:8984/solr/monitor");
  28. //批提交的条数
  29. val batchCount=10000;
  30. def main2(args: Array[String]) {
  31. val d1=new Record("row1","title","content","1","01","57","58","3");
  32. val d2=new Record("row2","title","content","1","01","57","58","45");
  33. val d3=new Record("row3","title","content","1","01","57","58",null);
  34. client.addBean(d1);
  35. client.addBean(d2)
  36. client.addBean(d3)
  37. client.commit();
  38. println("提交成功!")
  39. }
  40. /***
  41. * 迭代分区数据(一个迭代器集合),然后进行处理
  42. * @param lines 处理每个分区的数据
  43. */
  44. def indexPartition(lines:scala.Iterator[String] ): Unit ={
  45. //初始化集合,分区迭代开始前,可以初始化一些内容,如数据库连接等
  46. val datas = new util.ArrayList[Record]()
  47. //迭代处理每条数据,符合条件会提交数据
  48. lines.foreach(line=>indexLineToModel(line,datas))
  49. //操作分区结束后,可以关闭一些资源,或者做一些操作,最后一次提交数据
  50. commitSolr(datas,true);
  51. }
  52. /***
  53. * 提交索引数据到solr中
  54. *
  55. * @param datas 索引数据
  56. * @param isEnd 是否为最后一次提交
  57. */
  58. def commitSolr(datas:util.ArrayList[Record],isEnd:Boolean): Unit ={
  59. //仅仅最后一次提交和集合长度等于批处理的数量时才提交
  60. if ((datas.size()>0&&isEnd)||datas.size()==batchCount) {
  61. client.addBeans(datas);
  62. client.commit(); //提交数据
  63. datas.clear();//清空集合,便于重用
  64. }
  65. }
  66. /***
  67. * 得到分区的数据具体每一行,并映射
  68. * 到Model,进行后续索引处理
  69. *
  70. * @param line 每行具体数据
  71. * @param datas 添加数据的集合,用于批量提交索引
  72. */
  73. def indexLineToModel(line:String,datas:util.ArrayList[Record]): Unit ={
  74. //数组数据清洗转换
  75. val fields=line.split("\1",-1).map(field =>etl_field(field))
  76. //将清洗完后的数组映射成Tuple类型
  77. val tuple=buildTuble(fields)
  78. //将Tuple转换成Bean类型
  79. val recoder=Record.tupled(tuple)
  80. //将实体类添加至集合,方便批处理提交
  81. datas.add(recoder);
  82. //提交索引到solr
  83. commitSolr(datas,false);
  84. }
  85. /***
  86. * 将数组映射成Tuple集合,方便与Bean绑定
  87. * @param array field集合数组
  88. * @return tuple集合
  89. */
  90. def buildTuble(array: Array[String]):(String, String, String, String, String, String, String, String)={
  91. array match {
  92. case Array(s1, s2, s3, s4, s5, s6, s7, s8) => (s1, s2, s3, s4, s5, s6, s7,s8)
  93. }
  94. }
  95. /***
  96. * 对field进行加工处理
  97. * 空值替换为null,这样索引里面就不会索引这个字段
  98. * ,正常值就还是原样返回
  99. *
  100. * @param field 用来走特定规则的数据
  101. * @return 映射完的数据
  102. */
  103. def etl_field(field:String):String={
  104. field match {
  105. case "" => null
  106. case _ => field
  107. }
  108. }
  109. /***
  110. * 根据条件清空某一类索引数据
  111. * @param query 删除的查询条件
  112. */
  113. def deleteSolrByQuery(query:String): Unit ={
  114. client.deleteByQuery(query);
  115. client.commit()
  116. println("删除成功!")
  117. }
  118. def main(args: Array[String]) {
  119. //根据条件删除一些数据
  120. deleteSolrByQuery("t1:03")
  121. //远程提交时,需要提交打包后的jar
  122. val jarPath = "target\\spark-build-index-1.0-SNAPSHOT.jar";
  123. //远程提交时,伪装成相关的hadoop用户,否则,可能没有权限访问hdfs系统
  124. System.setProperty("user.name", "webmaster");
  125. //初始化SparkConf
  126. val conf = new SparkConf().setMaster("spark://192.168.1.187:7077").setAppName("build index ");
  127. //上传运行时依赖的jar包
  128. val seq = Seq(jarPath) :+ "D:\\tmp\\lib\\noggit-0.6.jar" :+ "D:\\tmp\\lib\\httpclient-4.3.1.jar" :+ "D:\\tmp\\lib\\httpcore-4.3.jar" :+ "D:\\tmp\\lib\\solr-solrj-5.1.0.jar" :+ "D:\\tmp\\lib\\httpmime-4.3.1.jar"
  129. conf.setJars(seq)
  130. //初始化SparkContext上下文
  131. val sc = new SparkContext(conf);
  132. //此目录下所有的数据,将会被构建索引,格式一定是约定好的
  133. val rdd = sc.textFile("hdfs://192.168.1.187:9000/user/monitor/gs/");
  134. //通过rdd构建索引
  135. indexRDD(rdd);
  136. //关闭索引资源
  137. client.close();
  138. //关闭SparkContext上下文
  139. sc.stop();
  140. }
  141. /***
  142. * 处理rdd数据,构建索引
  143. * @param rdd
  144. */
  145. def indexRDD(rdd:RDD[String]): Unit ={
  146. //遍历分区,构建索引
  147. rdd.foreachPartition(line=>indexPartition(line));
  148. }
  149. }

ok,至此,我们的建索引程序就写完了,本例子中用的是远程提交模式,实际上它也可以支持spark on yarn (cluster 或者 client ) 模式,不过此时需要注意的是,不需要显式指定setMaster的值,而由提交任务时,通过--master来指定运行模式,另外,依赖的相关jar包,也需要通过--jars参数来提交到集群里面,否则的话,运行时会报异常,最后看下本例子里面的solr是单机模式的,所以使用spark建索引提速并没有达到最大值,真正能发挥最大威力的是,多台search集群正如我画的架构图里面,每台机器是一个shard,这就是solrcloud的模式,或者在elasticsearch里面的集群shard,这样以来,才能真正达到高效批量的索引构建

本文分享自微信公众号 - 我是攻城师(woshigcs)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2016-02-01

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

扫码关注云+社区

领取腾讯云代金券

年度创作总结 领取年终奖励