在独立模式下使用spark实现elasticsearch的索引

Question

SparkConf conf = new SparkConf().setMaster("local")
            .setAppName("Indexer").set("spark.driver.maxResultSize", "2g");
    conf.set("es.index.auto.create", "true");
    conf.set("es.nodes", "localhost");
    conf.set("es.port", "9200");
    conf.set("es.write.operation", "index");
    JavaSparkContext sc = new JavaSparkContext(conf);
    Map<String, ?> numbers = ImmutableMap.of("one", 1, "two", 2);
    Map<String, ?> airports = ImmutableMap.of("OTP", "Otopeni", "SFO", "San Fran");

    JavaRDD<Map<String, ?>> javaRDD = jsc.parallelize(ImmutableList.of(doc1, doc2));
    JavaEsSpark.saveToEs(javaRDD, "spark/docs");      

我试着在本地写文件，这是工作良好的....可能是配置中的问题。

这些是pom.xml中的依赖项

<dependencies>
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>2.1.0</version>
    </dependency>
    <!-- <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> 
        <version>2.6.4</version> </dependency> -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.10</artifactId>
        <version>1.5.1</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch-spark_2.10</artifactId>
        <version>2.1.0</version>
    </dependency>

</dependencies>                               

Answer 1

stacktrace中的相关行是：

java.lang.StringIndexOutOfBoundsException：字符串索引超出范围：-1在java.lang.String.SubString(String.java:1967)在...

此错误不是来自您的代码。这是由于elasticsearch版本与您正在使用的elasticsearch-hadoop适配器不兼容造成的。版本为2.0.x和2.1.x的elasticsearch-hadoop适配器仅可用于ElasticSearch1.x。我在使用elasticsearch 2.1.1时遇到了同样的错误，因此不得不将我的elasticsearch版本降级到1.4.4。错误消失了。