贾攀的流水账

Panmax's Blog

0%

Spark 操作 Elasticsearch 示例

Posted on Valine:

上周五调研了下如何用 Spark 读写 Elasticsearch（下文简称 es），中间被官方提供的 jar 包卡了很久，所以本来想周末记录一下，结果一发懒就没做，就蹭到周一晚上来写一下了，最近调研的东西很多，有很多要记得东西，一点一点来吧。

不废话，直接 Show you the code：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark._
 
 
object ElasticSparkHelloWorld {
  def main(args: Array[String]) {
 
 
    val conf = new SparkConf().setAppName(ElasticSparkHelloWorld.getClass.getName)
    conf.setMaster("local")
    conf.set("es.nodes", "localhost")
    conf.set("es.port", "9200")
    conf.set("es.index.auto.create", "true")
    conf.set("es.nodes.wan.only", "true")
    conf.set("es.query", "?q=*")
    conf.set("es.resource", "spark/docs")
 
    val sc = new SparkContext(conf)
    val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3, "four" -> 4)
    val airports = Map("OTP" -> "Otopeni", "SFO" -> "San Fran")
 
    sc.makeRDD(Seq(numbers, airports)).saveToEs("spark/docs")
 
    println(sc.esRDD().count())
 
  }
}

其实上边这些代码从网上一搜一大堆，重点是下边 sbt 部分的配置：

1
2
3
4
5
6
7
8
9
10
name := "spark-es-demo"
 
version := "1.0"
 
scalaVersion := "2.11.11"
 
//scalacOptions += "-Ylog-classpath"
 
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.2"
libraryDependencies += "org.elasticsearch" % "elasticsearch-spark-13_2.11" % "5.5.2"

需要注意 scala 版本，spark 版本还有 es 版本一定要对应，否则无法运行

比如

  • scalaVersion 版本是 2.11.11
  • spark 版本是 1.6.2
  • es 版本是 5.5.2

依赖需要写成下边这样：

1
2
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.2"  // 这里指定 spark-core 的版本
libraryDependencies += "org.elasticsearch" % "elasticsearch-spark-13_2.11" % "5.5.2"

解释一下 "elasticsearch-spark-13_2.11" % "5.5.2" 这部分

-13 是给 Spark1.3-1.6 提供的
-20 是给 Spark2.0 提供的

_2.11 是 scalaVersion 的前边两位

5.5.2 是 elasticsearch 的版本号

官方文档中提到

The Spark connector framework is the most sensitive to version incompatibilities.

Spark 连接器框架是对版本号非常敏感并且不兼容的。

另外一个坑是，elasticsearch-spark-13_2.11 这个 jar 包所依赖的包无法在 maven 官方源中找到，需要添加另一个源：conjars: http://conjars.org/repo

在 ~/.sbt 下新建 repositories 文件，我的 repositories 内容如下：

1
2
3
4
5
[repositories]
  local
  aliyun: http://maven.aliyun.com/nexus/content/groups/public/
  conjars: http://conjars.org/repo
  central: http://repo1.maven.org/maven2/

将阿里源放在上边，可以让官方依赖下载更快。

完整代码见：https://github.com/Panmax/spark-es-demo