spark集群--elasticsearch

• spark集群搭建
• elasticsearch讀寫數據

====================================================

• spark集群搭建
  • spark官網：http://spark.apache.org
  • 各個版本：spark-2.1.0， java 1.8.0_121以及elasticsearch-5.2.0
  • 本集群利用的是spark的獨立集群管理器
  • 准備N台服務器（1台主節點，n-1台工作節點）
  • 分別將N台機器裝上java 1.8.0_121以及將spark-2.1.0-bin-hadoop2.7.tgz解壓到相同目錄下
  • 設置好從主節點機器到其他機器的SSH無密碼登錄（本事例采用的dsa）
  • 編輯主節點的 SPARK_HOME/config/slavers文件並填上所有工作節點的ip（1.此處可隨個人愛好填寫主機名。2.多工作節點用空格分割）
  • 在主節點上啟動集群 SPARK_HOME/sbin/start-all.sh
  • 訪問 http://主節點:8080 來查看集群狀態
  • 提交應用 SPARK_HOME/bin/spark-submit --master spark://XXX:7077 yourapp
  • 訪問 http://主節點:8080 來查看您的應用是否正常運行（1.Running Applications 有你的appname。2.列出了所使用的核心和內存均大於0）
  • 事例如下
  • 
  • 備注：不知道什么原因，集群的時候 需要在SPARK_HOME/conf/spark-env.sh 再次設置下JAVA_HOME，工作節點讀取不到JAVA_HOME的環境變量

 

• elasticsearch讀寫數據

   

  • 引用的jar包

  •       <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.10</artifactId>
                <version>2.1.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-hive_2.10</artifactId>
                <version>2.1.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.elasticsearch</groupId>
                <artifactId>elasticsearch-spark-20_2.11</artifactId>
                <version>5.1.2</version>
            </dependency>

     

  • 讀取數據

  •  
    SparkConf conf = new SparkConf().setAppName("e").setMaster("spark://主節點:7077");
    conf.set("es.nodes", "elasticsearchIP");
    conf.set("es.port", "9200");
    
    JavaSparkContext jsc = new JavaSparkContext(conf);
    JavaRDD<Map<String, Object>> esRDD =
                    JavaEsSpark.esRDD(jsc, "logstash-spark_test/spark_test", "?q=selpwd").values();

     

  • 寫數據

  •     JavaEsSpark.saveToEs(inJPRDD.values(), "logstash-spark_test/spark_test");