1. Spring Data

目的：简化操作，实际就是将对索引、文档等的操作映射成对象进行操作

1.1 版本对照

目前最新 springboot 对应 Elasticsearch7.6.2，Spring boot2.3.x 一般可以兼容 Elasticsearch7.x

image.png

1.2 集成

1.2.1 依赖

  1. <!-- 继承Spring-boot-starter-parent -->
  2. <parent>
  3.     <groupId>org.springframework.boot</groupId>
  4.     <artifactId>spring-boot-starter-parent</artifactId>
  5.     <version>2.3.6.RELEASE</version>
  6.     <relativePath/>
  7. </parent>
  9. <properties>
  10.     <maven.compiler.source>8</maven.compiler.source>
  11.     <maven.compiler.target>8</maven.compiler.target>
  12. </properties>
  14. <dependencies>
  16.     <!-- lombok -->
  17.     <dependency>
  18.         <groupId>org.projectlombok</groupId>
  19.         <artifactId>lombok</artifactId>
  20.     </dependency>
  22.     <!-- spring-data-es的startrer -->
  23.     <dependency>
  24.         <groupId>org.springframework.boot</groupId>
  25.         <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
  26.     </dependency>
  28.     <!-- devtools-->
  29.     <dependency>
  30.         <groupId>org.springframework.boot</groupId>
  31.         <artifactId>spring-boot-devtools</artifactId>
  32.         <scope>runtime</scope>
  33.         <optional>true</optional>
  34.     </dependency>
  36.     <!--spring test的starter-->
  37.     <dependency>
  38.         <groupId>org.springframework.boot</groupId>
  39.         <artifactId>spring-boot-starter-test</artifactId>
  40.         <scope>test</scope>
  41.     </dependency>
  43.     <!--spring boot test-->
  44.     <dependency>
  45.         <groupId>org.springframework.boot</groupId>
  46.         <artifactId>spring-boot-test</artifactId>
  47.     </dependency>
  49.     <!--junit-->
  50.     <dependency>
  51.         <groupId>junit</groupId>
  52.         <artifactId>junit</artifactId>
  53.     </dependency>
  55.     <!-- spring test -->
  56.     <dependency>
  57.         <groupId>org.springframework</groupId>
  58.         <artifactId>spring-test</artifactId>
  59.     </dependency>
  60. </dependencies>

1.2.2 整合

1.2.2.1 配置文件

  1. # 集群环境及访问方式
  2. elasticsearch:
  3.   nodes: hadoop101:9200,hadoop102:9200,hadoop103:9200
  4.   schema: http

1.2.2.2 启动程序

  1. /**
  2.  * @author JShawn 2021/8/10 13:05
  3.  */
  4. @SpringBootApplication
  5. public class SpringDataElasticSearchMainApplication {
  6.     public static void main(String[] args) {
  7.         SpringApplication.run(SpringDataElasticSearchMainApplication.class, args);
  8.     }
  9. }

1.2.2.3 实体类

实体类需要配置mapping映射关系 @Id —- 指定 **_id** @Document —- 指定**索引的名称、分片、副本**等信息 @Field —- 指定**字段的数据类型、分词器类型、是否索引、是否单独存储**等信息

  1. /**
  2.  * @Id
  3.  *      必须有 id,这里的 id 是全局唯一的标识，等同于 es 中的"_id"
  4.  * @FiledType：
  5.  *      type : 字段数据类型
  6.  *      analyzer : 分词器类型
  7.  *      index : 是否索引(默认:true)
  8.  *      Keyword : 短语,不进行分词
  9.  *      store：是否单独存储
  10.  */
  11. @Data
  12. @NoArgsConstructor
  13. @AllArgsConstructor
  14. @ToString
  15. @Document(indexName = "product", shards = 3, replicas = 1)
  16. public class Product {
  18.     /**
  19.      * 商品唯一标识
  20.      */
  21.     @Id
  22.     private Long id;
  24.     /**
  25.      * 商品名称
  26.      */
  27.     @Field(type = FieldType.Text, analyzer = "ik_max_word")
  28.     private String title;
  30.     /**
  31.      * 分类名称
  32.      */
  33.     @Field(type = FieldType.Keyword)
  34.     private String category;
  36.     /**
  37.      * 商品价格
  38.      */
  39.     @Field(type = FieldType.Double)
  40.     private Double price;
  42.     /**
  43.      * 图片地址
  44.      */
  45.     @Field(type = FieldType.Keyword, index = false)
  46.     private String images;
  47. }

1.2.2.4 配置类

ElasticsearchRestTemplate 是 spring-data-elasticsearch 项目中的一个类，和其他 spring 项目中的 template类似。

在新版的 spring-data-elasticsearch 中，ElasticsearchRestTemplate 代替了原来的 ElasticsearchTemplate。原因是 ElasticsearchTemplate 基于 TransportClient，TransportClient 即将在 8.x 以后的版本中移除。所以，推荐使用 ElasticsearchRestTemplate。 ElasticsearchRestTemplate 基 于 RestHighLevelClient 客户端的。需要自定义配置类，继承 **AbstractElasticsearchConfiguration** ，并实现 **elasticsearchClient()** 抽象方法，创建 RestHighLevelClient 对象。

  1. /**
  2.  * @author JShawn 2021/8/10 13:07
  3.  */
  4. @ConfigurationProperties(prefix = "elasticsearch")
  5. @Configuration
  6. @Data
  7. public class ElasticsearchConfig extends AbstractElasticsearchConfiguration {
  9.     /**
  10.      * 集群地址
  11.      */
  12.     private String nodes;
  14.     /**
  15.      * 访问方式
  16.      */
  17.     private String schema;
  19.     @Override
  20.     public RestHighLevelClient elasticsearchClient() {
  21.         // 1. 获取ES客户端
  22.         // 1. ES集群地址配置
  23.         String[] clusters = nodes.split(",");
  24.         HttpHost[] hosts = new HttpHost[clusters.length];
  25.         for (int i = 0; i < clusters.length; i++) {
  26.             String[] address = clusters[i].split(":");
  27.             HttpHost httpHost = new HttpHost(address[0], Integer.parseInt(address[1]), schema);
  28.             hosts[i] = httpHost;
  29.         }
  30.         // 2. 创建客户端
  31.         RestHighLevelClient restHighLevelClient = new RestHighLevelClient(RestClient.builder(hosts));
  32.         return restHighLevelClient;
  33.     }
  34. }

1.2.2.5 数据访问层

跟其他SpringData整合一样，继承一个Repository接口

  1. /**
  2.  * @author JShawn 2021/8/10 13:19
  3.  */
  4. @Repository
  5. public interface ProductDao extends ElasticsearchRepository<Product,Long> {
  6. }

1.2.2.6 索引CRUD

  1. /**
  2.  * 索引CRUD测试
  3.  *
  4.  * @author JShawn 2021/8/10 17:15
  5.  */
  6. @RunWith(SpringRunner.class)
  7. @SpringBootTest
  8. public class SpringDataESIndexTest {
  10.     /**
  11.      * 注入 ElasticsearchRestTemplate
  12.      */
  13.     @Autowired
  14.     private ElasticsearchRestTemplate elasticsearchRestTemplate;
  16.     /**
  17.      * 1. 创建索引并增加映射配置（映射已在实体类中进行配置）
  18.      * 在初始化ElasticsearchRestTemplate时，也要初始化ProductDao
  19.      * 因此在读取Product类时，如果发现此索引没被创建，会自动创建
  20.      */
  21.     @Test
  22.     public void createIndex(){
  23.         //创建索引，系统初始化会自动创建索引
  24.         System.out.println("创建索引");
  25.     }
  27.     /**
  28.      * 2. 删除索引
  29.      */
  30.     @Test
  31.     public void deleteIndex(){
  32.         //创建索引，系统初始化会自动创建索引
  33.         boolean flg = elasticsearchRestTemplate.deleteIndex(Product.class);
  34.         System.out.println("删除索引 = " + flg);
  35.     }
  36. }

1.2.2.7 文档CRUD

  1. /**
  2.  * 文档CRUD测试
  3.  *
  4.  * @author JShawn 2021/8/10 17:23
  5.  */
  6. @RunWith(SpringRunner.class)
  7. @SpringBootTest
  8. public class SpringDataESProductDaoTest {
  10.     @Autowired
  11.     private ProductDao productDao;
  13.     /**
  14.      * 1. 新增文档
  15.      * save()
  16.      */
  17.     @Test
  18.     public void save(){
  19.         Product product = new Product();
  20.         product.setId(2L);
  21.         product.setTitle("华为手机");
  22.         product.setCategory("手机");
  23.         product.setPrice(2999.0);
  24.         product.setImages("http://www.jshawn/hw.jpg");
  25.         productDao.save(product);
  26.     }
  28.     /**
  29.      * 2. 修改文档
  30.      * 也是save()，存在id对应的记录就修改，没有就新增
  31.      */
  32.     @Test
  33.     public void update(){
  34.         Product product = new Product();
  35.         product.setId(1L);
  36.         product.setTitle("小米 2 手机");
  37.         product.setCategory("手机");
  38.         product.setPrice(9999.0);
  39.         product.setImages("http://www.jshawn/xm.jpg");
  40.         productDao.save(product);
  41.     }
  43.     /**
  44.      * 3. 根据 id 查询文档
  45.      */
  46.     @Test
  47.     public void findById(){
  48.         Product product = productDao.findById(1L).get();
  49.         System.out.println(product);
  50.     }
  52.     /**
  53.      * 4. 查询所有文档
  54.      */
  55.     @Test
  56.     public void findAll(){
  57.         Iterable<Product> products = productDao.findAll();
  58.         for (Product product : products) {
  59.             System.out.println(product);
  60.         }
  61.     }
  63.     /**
  64.      * 5. 删除
  65.      */
  66.     @Test
  67.     public void delete(){
  68.         Product product = new Product();
  69.         product.setId(2L);
  70.         productDao.delete(product);
  71.     }
  73.     /**
  74.      * 6. 批量新增
  75.      */
  76.     @Test
  77.     public void saveAll(){
  78.         List<Product> productList = new ArrayList<>();
  79.         for (int i = 0; i < 10; i++) {
  80.             Product product = new Product();
  81.             product.setId(Long.valueOf(i));
  82.             product.setTitle("["+i+"]小米手机");
  83.             product.setCategory("手机");
  84.             product.setPrice(1999.0+i);
  85.             product.setImages("http://www.atguigu/xm.jpg");
  86.             productList.add(product);
  87.         }
  88.         productDao.saveAll(productList);
  89.     }
  91.     /**
  92.      * 7. 分页查询
  93.      */
  94.     @Test
  95.     public void findByPageable(){
  96.         // 设置排序(排序方式，正序还是倒序，排序的 id)
  97.         Sort sort = Sort.by(Sort.Direction.DESC,"id");
  98.         // 当前页，特殊：******第一页从 0 开始，1 表示第二页******
  99.         int currentPage=0;
  100.         // 每页显示多少条
  101.         int pageSize = 5;
  102.         // 设置查询分页
  103.         PageRequest pageRequest = PageRequest.of(currentPage, pageSize,sort);
  104.         // 分页查询
  105.         Page<Product> productPage = productDao.findAll(pageRequest);
  106.         for (Product Product : productPage.getContent()) {
  107.             System.out.println(Product);
  108.         }
  109.     }
  110. }

1.2.2.8 高级搜索

更多操作看Bos中的SpringData整合ES或Java API

  1. /**
  2.  * 文档搜索测试------更多操作看Bos中的SpringData整合ES或Java API
  3.  *
  4.  * @author JShawn 2021/8/12 0:47
  5.  */
  6. @RunWith(SpringRunner.class)
  7. @SpringBootTest
  8. public class SpringDataESSearchTest {
  10.     @Autowired
  11.     private ProductDao productDao;
  13.     /**
  14.      * 1. term 查询
  15.      * search(termQueryBuilder) 调用搜索方法，参数查询构建器对象
  16.      */
  17.     @Test
  18.     public void termQuery(){
  19.         TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("title", "小米");
  20.         Iterable<Product> products = productDao.search(termQueryBuilder);
  21.         for (Product product : products) {
  22.             System.out.println(product);
  23.         }
  24.     }
  26.     /**
  27.      * 2. term 查询加分页
  28.      */
  29.     @Test
  30.     public void termQueryByPage(){
  31.         int currentPage= 0 ;
  32.         int pageSize = 5;
  33.         //设置查询分页
  34.         PageRequest pageRequest = PageRequest.of(currentPage, pageSize);
  35.         TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("title", "小米");
  36.         Iterable<Product> products = productDao.search(termQueryBuilder,pageRequest);
  37.         for (Product product : products) {
  38.             System.out.println(product);
  39.         }
  40.     }
  41. }

2. Spark Streaming

前提：需要Scala环境

2.1 依赖

  1. <properties>
  2.     <maven.compiler.source>8</maven.compiler.source>
  3.     <maven.compiler.target>8</maven.compiler.target>
  4. </properties>
  6. <dependencies>
  7.       <!-- Spark -->
  8.     <dependency>
  9.         <groupId>org.apache.spark</groupId>
  10.         <artifactId>spark-core_2.12</artifactId>
  11.         <version>3.0.0</version>
  12.     </dependency>
  14.       <!-- Spark Streaming -->
  15.     <dependency>
  16.         <groupId>org.apache.spark</groupId>
  17.         <artifactId>spark-streaming_2.12</artifactId>
  18.         <version>3.0.0</version>
  19.     </dependency>
  21.       <!-- ElasticSearch -->
  22.     <dependency>
  23.         <groupId>org.elasticsearch</groupId>
  24.         <artifactId>elasticsearch</artifactId>
  25.         <version>7.8.0</version>
  26.     </dependency>
  28.     <!-- elasticsearch 的高级客户端 -->
  29.     <dependency>
  30.         <groupId>org.elasticsearch.client</groupId>
  31.         <artifactId>elasticsearch-rest-high-level-client</artifactId>
  32.         <version>7.8.0</version>
  33.     </dependency>
  35.     <!-- elasticsearch 依赖 2.x 的 log4j -->
  36.     <dependency>
  37.         <groupId>org.apache.logging.log4j</groupId>
  38.         <artifactId>log4j-api</artifactId>
  39.         <version>2.8.2</version>
  40.     </dependency>
  41.     <dependency>
  42.         <groupId>org.apache.logging.log4j</groupId>
  43.         <artifactId>log4j-core</artifactId>
  44.         <version>2.8.2</version>
  45.     </dependency>
  47.     <!-- <dependency>-->
  48.     <!-- <groupId>com.fasterxml.jackson.core</groupId>-->
  49.     <!-- <artifactId>jackson-databind</artifactId>-->
  50.     <!-- <version>2.11.1</version>-->
  51.     <!-- </dependency>-->
  52.     <!-- &lt;!&ndash; junit 单元测试 &ndash;&gt;-->
  53.     <!-- <dependency>-->
  54.     <!-- <groupId>junit</groupId>-->
  55.     <!-- <artifactId>junit</artifactId>-->
  56.     <!-- <version>4.12</version>-->
  57.     <!-- </dependency>-->
  58. </dependencies>

2.2 集成

  1. object SparkStreamingESTest {
  2.     def main(args: Array[String]): Unit = {
  3.         val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ESTest")
  4.         val ssc = new StreamingContext(sparkConf, Seconds(3))
  5.         val ds: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
  6.         ds.foreachRDD(
  7.             rdd => {
  8.                 println("*************** " + new Date())
  9.                 rdd.foreach(
  10.           data => {
  11.             val client = new RestHighLevelClient(RestClient.builder(new HttpHost("localhost", 9200, "http")));
  12.             // 新增文档 - 请求对象
  13.             val request = new IndexRequest();
  14.             // 设置索引及唯一性标识
  15.             val ss = data.split(" ")
  16.             println("ss = " + ss.mkString(","))
  17.             request.index("sparkstreaming").id(ss(0));
  18.             val productJson =
  19.                                             s"""
  20.                             | { "data":"${ss(1)}" }
  21.                             |""".stripMargin;
  22.             // 添加文档数据，数据格式为 JSON 格式
  23.             request.source(productJson,XContentType.JSON);
  24.             // 客户端发送请求，获取响应对象
  25.             val response = client.index(request, RequestOptions.DEFAULT);
  26.             System.out.println("_index:" + response.getIndex());
  27.             System.out.println("_id:" + response.getId());
  28.             System.out.println("_result:" + response.getResult());
  29.             client.close()
  30.           }
  31.         )
  32.       }
  33.     )
  34.     ssc.start()
  35.     ssc.awaitTermination()
  36.   }
  37. }

3. Flink

可以参考Flink 五、Flink的流处理API 4.3 ElasticSearch

3.1 依赖

  1. <!-- Flink集成ES所需包 -->
  2. <dependency>
  3.     <groupId>org.apache.flink</groupId>
  4.     <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
  5.     <version>1.12.0</version>
  6. </dependency>
  7. <!-- Flink -->
  8. <dependency>
  9.     <groupId>org.apache.flink</groupId>
  10.     <artifactId>flink-scala_2.12</artifactId>
  11.     <version>1.12.0</version>
  12. </dependency>
  13. <dependency>
  14.     <groupId>org.apache.flink</groupId>
  15.     <artifactId>flink-streaming-scala_2.12</artifactId>
  16.     <version>1.12.0</version>
  17. </dependency>
  18. <dependency>
  19.     <groupId>org.apache.flink</groupId>
  20.     <artifactId>flink-clients_2.12</artifactId>
  21.     <version>1.12.0</version>
  22. </dependency>

3.2 集成

  1. /**
  2.  * Flink 集成 ES
  3.  *
  4.  * @author JShawn 2021/8/17 22:02
  5.  */
  6. public class FlinkElasticSearchTest {
  7.     public static void main(String[] args) throws Exception {
  9.         //1. 构建Flink环境
  10.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  12.         // 2. source
  13.         DataStream<String> source = env.socketTextStream("localhost", 7777);
  15.         // 3.sink
  16.         // 3.1 配置ES集群
  17.         ArrayList<HttpHost> httpHosts = new ArrayList<>();
  18.         httpHosts.add(new HttpHost("hadoop101", 9200));
  19.         httpHosts.add(new HttpHost("hadoop102", 9200));
  20.         httpHosts.add(new HttpHost("hadoop103", 9200));
  22.         // 3.2 构建ES的Sink
  23.         ElasticsearchSink.Builder esSinkBuilder = new ElasticsearchSink.Builder(httpHosts, new ElasticsearchSinkFunction<String>() {
  24.             @Override
  25.             public void process(String source, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
  26.                 // 3.3 封装数据
  27.                 HashMap<String, String> map = Maps.newHashMap();
  28.                 map.put("data", source);
  30.                 // 创建请求，作为向ES发起的写入命令
  31.                 IndexRequest indexRequest = Requests.indexRequest()
  32.                         .index("flink-es")
  33.                         .source(map);
  35.                 // 向index发送请求
  36.                 requestIndexer.add(indexRequest);
  37.             }
  38.         });
  40.         // 3.5 可选，由于默认是批处理形式，因此需要设置批处理刷新阈值为1，有一条写一条
  41.         esSinkBuilder.setBulkFlushMaxActions(1);
  43.         // 3.6 绑定source和sink
  44.         source.addSink(esSinkBuilder.build());
  46.         // 3.7 执行job
  47.         env.execute("flink-es");
  48.     }
  49. }