Spark 流式传输到 ElasticSearch答案

【问题标题】：Spark Streaming to ElasticSearchSpark 流式传输到 ElasticSearch
【发布时间】：2016-09-21 00:49:39
【问题描述】：

我正在尝试使用 Spark Java API 复制 Holden Karau 的这个示例 Streamlining Search Indexing using Elastic Search。我已经成功地将它作为一个普通的 Java 应用程序工作，并对代码进行了一些更改。我没有使用saveAsHadoopDataset 方法发送推文，而是使用：

   JavaEsSpark.saveToEs(rdd,"/test/collection");

并运行我的代码：

 java -cp ./target/hbase-spark-playground-1.0-SNAPSHOT.jar spark.examples.SparkToElasticSearchStreaming local[2] collection-name

我目前的问题是如何在 Yarn Cluster 上执行它。我正在做的代码 sn-p 可以在这里找到：

https://gist.github.com/IvanFernandez/b3a3e25397f8b402256b

并以这种方式运行我的课程：

spark.examples.SparkToElasticSearchStreaming --master yarn-cluster --executor-memory 400m --num-executors 1 ./target/hbase-spark-playground-1.0-SNAPSHOT.jar yarn-cluster collection-name

我认为主要问题是我在 foreach 转换中没有任何 elasticSearch 配置，因此无法访问我的 elasticSearch 主服务器。有什么想法吗？

【问题讨论】：

标签： elasticsearch apache-spark

【解决方案1】：

es集群或其他配置信息应该在SparkConf中设置，这已经在你的代码sn-p as args[2]中设置为es.nodes。在您的 yarn 命令中，缺少 es host 的第三个参数，我也相信您的命令正在使用 spark-submit 提交应用程序。

【讨论】：

【解决方案2】：

您能否尝试在 SparkConf 中设置 spark.es.nodes 和 es.port 属性，如下所示：

sparkConf.set("spark.es.nodes", args[2]);
sparkConf.set("es.port", args[3]);   // HTTP Port of elastic search

并使用以下命令在 yarn 上运行应用程序：

spark-submit --class spark.examples.SparkToElasticSearchStreaming --master yarn-cluster --executor-memory 400m --num-executors 1 ./target/hbase-spark-playground-1.0-SNAPSHOT.jar yarn-cluster collection-name localhost 9200

【讨论】：