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");