Spark 集成 ElasticSearch

对于spark的典型应用场景为批处理，一般由基本数据源（文件系统如：hdfs）或者高级数据源（flume、kafka）作为spark的数据接入端。输出一样可以是文件系统或数据库等等。本文介绍一个用java写的demo程序，功能是从本地接收数据，经过spark处理之后输出到Elasticsearch。

先上代码：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.3.1</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.11</artifactId>
    <version>2.3.1</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId>
    <version>2.3.1</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-hadoop</artifactId>
    <version>6.3.0</version>
</dependency>
</dependencies>

代码

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.SparkSession;

import com.unigroup.utils.BCConvert;

import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
import org.elasticsearch.spark.rdd.api.java.JavaEsspark;
public static void main(String[] args) throws Exception {

    //利用sparksql创建Context
    SparkSession spark = SparkSession.builder()
            .appName("JavaWordCount")
            .master("local[2]")//如果是本地环境可以是local[n]，n是线程数，必须大于1
      //    .master("spark://192.168.7.51:7077")//如果是集群模式，需要指定master地址
            .config("spark.es.nodes", "192.168.7.51")//指定es地址
            .config("spark.es.port", "9200")//指定es端口号
            .getorCreate();
    
    //指定本地文件路径，如果spark是集群模式，需要每个节点上对应路径下都要有此文件。或者使用hdfs。
    JavaRDD<String> lines = spark.read().textFile("/path/to/test.txt").javaRDD();
    
    //切分单词
    JavaRDD<String> words = lines.flatMap(s -> Arrays.asList(SPACE.split(s)).iterator());
    
    //转PairRDD
    JavaPairRDD<String, Integer> ones = words.mapToPair(s -> new Tuple2<>(s, 1));
    
    //统计单词数
    JavaPairRDD<String, Integer> counts = ones.reduceByKey((i1, i2) -> i1 + i2);

      //全角转半角
    JavaPairRDD<String, String> ones1 = words.mapToPair(s -> new Tuple2<>(s, BCConvert.bj2qj(s)));
    
    //将结果保存到文件系统
    // counts.saveAsTextFile("/Users/liubowen/sparkstream/out");

     //返回结果
    // List<Tuple2<String, Integer>> output = counts.collect();

      //循环结果
    // for (Tuple2<?,?> tuple : output) {
    //         System.out.println(tuple._1() + ": " + tuple._2());
    // }

      //将结果保存到es
    //JavaEsspark.saveJsonToEs(lines, "/spark/doc");
    
    // List list = ones.collect();
    // System.out.println(list);
    spark.stop();
}

读取文件系统数据
spark读取文件系统数据一般使用textFile()，注意这里如果是从本地文件中取数据，在集群模式下，必须保证每个节点对应路径下都要有数据文件。也可以使用hdfs或nfs。

textFile("/path/to/test.txt")
textFile("hdfs://hadoop:9000/test.txt")

输出结果到Elasticsearch
maven依赖，使用elasticsearch-hadoop包，注意对应Elasticsearch版本

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-hadoop</artifactId>
    <version>6.3.0</version>
</dependency>

scala

import org.elasticsearch.spark.rdd.Esspark;

java

import org.elasticsearch.spark.rdd.api.java.JavaEsspark;

需要指定Elasticsearch，可以写在配置文件里，也可以在程序里指定。

配置文件：spark-default.conf

1. spark.es.nodes　　  eshosts
2. spark.es.port　　   9200

写在程序里

SparkConf conf = new SparkConf().setMaster(master).setAppName("StreamingTest")
        // .set("spark.executor.memory", "1g")
        // .set("spark.dynamicAllocation.enabled", "false")
          .set("spark.es.nodes", "192.168.7.51")
          .set("spark.es.port", "9200");

写入

JavaEssparkStreaming.saveJsonToEs(out, "/spark/doc");