框架集成-SpringData-整体介绍

Spring Data是一个用于简化数据库、非关系型数据库、索引库访问，并支持云服务的开源框架。其主要目标是使得对数据的访问变得方便快捷，并支持 map-reduce框架和云计算数据服务。Spring Data可以极大的简化JPA(Elasticsearch…)的写法，可以在几乎不用写实现的情况下，实现对数据的访问和操作。除了CRUD 外，还包括如分页、排序等一些常用的功能。

Spring Data 的官网：https://spring.io/projects/spring-data
image.png

  • Spring Data 常用的功能模块如下：
    1. Spring Data JDBC
    2. Spring Data JPA
    3. Spring Data LDAP
    4. Spring Data MongoDB
    5. Spring Data Redis
    6. Spring Data R2DBC
    7. Spring Data REST
    8. Spring Data for Apache Cassandra
    9. Spring Data for Apache Geode
    10. Spring Data for Apache Solr
    11. Spring Data for Pivotal GemFire
    12. Spring Data Couchbase
    13. Spring Data Elasticsearch
    14. Spring Data Envers
    15. Spring Data Neo4j
    16. Spring Data JDBC Extensions
    17. Spring for Apache Hadoop

    Spring Data Elasticsearch 介绍

    Spring Data Elasticsearch基于Spring Data API简化 Elasticsearch 操作，将原始操作Elasticsearch 的客户端API进行封装。Spring Data为Elasticsearch 项目提供集成搜索引擎。Spring Data Elasticsearch POJO的关键功能区域为中心的模型与Elastichsearch交互文档和轻松地编写一个存储索引库数据访问层。

Spring Data Elasticsearch 官网：https://spring.io/projects/spring-data-elasticsearch

创建elasticsearch-springdata项目

修改pom文件，增加依赖关系

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>tian996cloud2020</artifactId>
  7.         <groupId>com.tian996.cloud</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  12.     <artifactId>elasticsearch-springdata</artifactId>
  14.     <properties>
  15.         <maven.compiler.source>8</maven.compiler.source>
  16.         <maven.compiler.target>8</maven.compiler.target>
  17.     </properties>
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.springframework.boot</groupId>
  22.             <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.projectlombok</groupId>
  26.             <artifactId>lombok</artifactId>
  27.             <optional>true</optional>
  28.         </dependency>
  30.         <dependency>
  31.             <groupId>org.springframework.boot</groupId>
  32.             <artifactId>spring-boot-devtools</artifactId>
  33.             <scope>runtime</scope>
  34.             <optional>true</optional>
  35.         </dependency>
  36.         <dependency>
  37.             <groupId>org.springframework.boot</groupId>
  38.             <artifactId>spring-boot-starter-test</artifactId>
  39.             <scope>test</scope>
  40.         </dependency>
  41.         <dependency>
  42.             <groupId>org.springframework.boot</groupId>
  43.             <artifactId>spring-boot-test</artifactId>
  44.         </dependency>
  45.         <dependency>
  46.             <groupId>junit</groupId>
  47.             <artifactId>junit</artifactId>
  48.         </dependency>
  49.         <dependency>
  50.             <groupId>org.springframework</groupId>
  51.             <artifactId>spring-test</artifactId>
  52.         </dependency>
  54.         <!-- hutool工具类 -->
  55.         <dependency>
  56.             <groupId>cn.hutool</groupId>
  57.             <artifactId>hutool-all</artifactId>
  58.             <version>${hutool.version}</version>
  59.         </dependency>
  60.     </dependencies>
  62. </project>

增加配置文件

  • 在 resources 目录中增加application.properties文件

    1. # es 服务地址
    2. elasticsearch.host=127.0.0.1
    3. # es 服务端口
    4. elasticsearch.port=9200
    5. # 配置日志级别,开启 debug 日志
    6. logging.level.com.atguigu.es=debug

    Spring Boot 主程序

    主启动类

    ```java @SpringBootApplication public class ElasticsearchApplicationMain {

    public static void main(String[] args) {

    1.   SpringApplication.run(ElasticsearchApplicationMain.class,args);

    }

}

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

配置类

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

需要自定义配置类，继承AbstractElasticsearchConfiguration，并实现elasticsearchClient()抽象方法，创建RestHighLevelClient对象。

  1. @ConfigurationProperties(prefix = "elasticsearch")
  2. @Configuration
  3. @Data
  4. public class ElasticsearchConfig extends AbstractElasticsearchConfiguration {
  7.     private String host;
  9.     private Integer port;
  11.     @Override
  12.     public RestHighLevelClient elasticsearchClient() {
  13.         RestClientBuilder builder = RestClient.builder(new HttpHost(host, port));
  14.         RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder);
  15.         return restHighLevelClient;
  16.     }
  17. }
  • AbstractElasticsearchConfiguration源码： ```java package org.springframework.data.elasticsearch.config;

import org.elasticsearch.client.RestHighLevelClient; import org.springframework.context.annotation.Bean; import org.springframework.data.elasticsearch.core.ElasticsearchOperations; import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;

public abstract class AbstractElasticsearchConfiguration extends ElasticsearchConfigurationSupport { public AbstractElasticsearchConfiguration() { }

  1. public abstract RestHighLevelClient elasticsearchClient();
  3. @Bean(
  4.     name = {"elasticsearchOperations", "elasticsearchTemplate"}
  5. )
  6. public ElasticsearchOperations elasticsearchOperations() {
  7.     return new ElasticsearchRestTemplate(this.elasticsearchClient(), this.elasticsearchConverter(), this.resultsMapper());
  8. }

}

  1. <a name="TPmzz"></a>
  2. #### DAO数据访问对象
  3. ```java
  4. import com.tian996.elasticsearch.entity.Product;
  5. import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
  6. import org.springframework.stereotype.Repository;
  8. @Repository
  9. public interface ProductDao extends ElasticsearchRepository<Product,Long> {
  11. }

索引操作

  1. import com.tian996.elasticsearch.entity.Product;
  2. import org.junit.Test;
  3. import org.junit.runner.RunWith;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.boot.test.context.SpringBootTest;
  6. import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
  7. import org.springframework.test.context.junit4.SpringRunner;
  9. @RunWith(SpringRunner.class)
  10. @SpringBootTest
  11. public class SpringDataESIndexTest {
  13.     @Autowired
  14.     private ElasticsearchRestTemplate elasticsearchRestTemplate;
  16.     @Test
  17.     public void createIndex(){
  18.         //创建索引，系统初始化会自动创建索引
  19.         boolean index = elasticsearchRestTemplate.createIndex(Product.class);
  20.         System.out.println("创建索引 = " + index);
  21.     }
  23.     @Test
  24.     public void deleteIndex(){
  25.         //创建索引，系统初始化会自动创建索引
  26.         boolean flg = elasticsearchRestTemplate.deleteIndex(Product.class);
  27.         System.out.println("删除索引 = " + flg);
  28.     }
  30. }
  • 用Postman 检测有没有创建和删除。
    1. #GET http://localhost:9200/_cat/indices?v

    文档操作

    ```java import com.tian996.elasticsearch.dao.ProductDao; import com.tian996.elasticsearch.entity.Product; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.data.domain.Page; import org.springframework.data.domain.PageRequest; import org.springframework.data.domain.Sort; import org.springframework.test.context.junit4.SpringRunner;

import java.util.ArrayList; import java.util.List;

@RunWith(SpringRunner.class) @SpringBootTest public class SpringDataESProductDaoTest {

  1. @Autowired
  2. private ProductDao productDao;
  4. /**
  5.  * 新增
  6.  * POSTMAN, GET http://localhost:9200/product/_doc/1
  7.  */
  8. @Test
  9. public void save(){
  10.     Product images = Product.builder().id(1L).title("华为手机").category("手机").price(4999.0).images("http://www.atguigu/hw.jpg").build();
  11.     productDao.save(images);

// Product images2 = Product.builder().id(2L).title(“小米手机”).category(“手机”).price(2999.0).images(“http://www.atguigu/hw.jpg").build(); // productDao.save(images2); }

  1. /**
  2.  * 修改
  3.  * POSTMAN, GET http://localhost:9200/product/_doc/1
  4.  */
  5. @Test
  6. public void update(){
  7.     Product product = Product.builder().id(2L).title("小米 2 手机").category("手机").price(9999.0).images("http://www.atguigu/xm.jpg").build();
  8.     productDao.save(product);
  9. }
  11. /**
  12.  * 根据 id 查询
  13.  *
  14.  */
  15. @Test
  16. public void findById(){
  17.     Product product = productDao.findById(1L).get();
  18.     System.out.println(product.toString());
  19. }
  21. @Test
  22. public void findAll(){
  23.     Iterable<Product> products = productDao.findAll();
  24.     for (Product product : products) {
  25.         System.out.println(product);
  26.     }
  27. }
  29. /**
  30.  * 删除
  31.  */
  32. @Test
  33. public void delete(){
  34.     Product product = new Product();
  35.     product.setId(2L);
  36.     productDao.delete(product);
  37. }
  39. /**
  40.  * 批量新增
  41.  */
  42. @Test
  43. public void saveAll(){
  44.     List<Product> productList = new ArrayList<>();
  45.     for (int i = 10; i < 20; i++) {
  46.         Product product = Product.builder().id(Long.valueOf(i)).title("["+i+"]小米手机").category("手机").price(2999.0).images("http://www.atguigu/hw.jpg").build();
  47.         productList.add(product);
  48.     }
  49.     productDao.saveAll(productList);
  50. }
  52. /**
  53.  * 分页查询
  54.  */
  55. @Test
  56. public void findByPageable(){
  57.     //设置排序(排序方式，正序还是倒序，排序的 id)
  58.     Sort sort = Sort.by(Sort.Direction.DESC,"id");
  59.     //当前页，第一页从 0 开始， 1 表示第二页
  60.     int currentPage=0;
  61.     //每页显示多少条
  62.     int pageSize = 5;
  63.     //设置查询分页
  64.     PageRequest pageRequest = PageRequest.of(currentPage, pageSize,sort);
  65.     //分页查询
  66.     Page<Product> productPage = productDao.findAll(pageRequest);
  67.     for (Product Product : productPage.getContent()) {
  68.         System.out.println(Product);
  69.     }
  70. }

}

  1. <a name="jQRhn"></a>
  2. # Spark Streaming 框架集成
  3. <a name="pVTGb"></a>
  4. ## Spark Streaming 框架介绍
  5. Spark Streaming 是 Spark core API 的扩展，支持实时数据流的处理，并且具有可扩展，
  6. 高吞吐量，容错的特点。数据可以从许多来源获取，如 Kafka，Flume，Kinesis 或 TCP sockets，
  7. 并且可以使用复杂的算法进行处理，这些算法使用诸如 map，reduce，join 和 window 等高
  8. 级函数表示。 最后，处理后的数据可以推送到文件系统，数据库等。 实际上，您可以将
  9. Spark 的机器学习和图形处理算法应用于数据流。<br />![](https://cdn.nlark.com/yuque/0/2021/png/1288078/1630634509791-377f9cc9-d611-4d8b-88af-4cbcb8338f43.png#clientId=uc63d090b-fc64-4&from=paste&id=uc8bf3509&margin=%5Bobject%20Object%5D&originHeight=262&originWidth=919&originalType=url&ratio=1&status=done&style=none&taskId=u116697b8-10ea-4a86-a417-7ae5111a729)
  11. **需要有scala编程语言**
  12. <a name="kg2E6"></a>
  13. ## 创建elasticsearch-sparkstreaming项目
  14. <a name="ILmF5"></a>
  15. ### 修改pom文件，增加依赖关系
  16. ```xml
  17. <dependencies>
  18.         <dependency>
  19.             <groupId>org.apache.spark</groupId>
  20.             <artifactId>spark-core_2.12</artifactId>
  21.             <version>3.0.0</version>
  22.         </dependency>
  23.         <dependency>
  24.             <groupId>org.apache.spark</groupId>
  25.             <artifactId>spark-streaming_2.12</artifactId>
  26.             <version>3.0.0</version>
  27.         </dependency>
  28.         <dependency>
  29.             <groupId>org.elasticsearch</groupId>
  30.             <artifactId>elasticsearch</artifactId>
  31.             <version>7.9.3</version>
  32.         </dependency>
  33.         <!-- elasticsearch 的客户端 -->
  34.         <dependency>
  35.             <groupId>org.elasticsearch.client</groupId>
  36.             <artifactId>elasticsearch-rest-high-level-client</artifactId>
  37.             <version>7.9.3</version>
  38.         </dependency>
  39.         <!-- elasticsearch 依赖 2.x 的 log4j -->
  40.         <dependency>
  41.             <groupId>org.apache.logging.log4j</groupId>
  42.             <artifactId>log4j-api</artifactId>
  43.             <version>2.8.2</version>
  44.         </dependency>
  45.         <dependency>
  46.             <groupId>org.apache.logging.log4j</groupId>
  47.             <artifactId>log4j-core</artifactId>
  48.             <version>2.8.2</version>
  49.         </dependency>
  50.         <!-- <dependency>-->
  51.         <!-- <groupId>com.fasterxml.jackson.core</groupId>-->
  52.         <!-- <artifactId>jackson-databind</artifactId>-->
  53.         <!-- <version>2.11.1</version>-->
  54.         <!-- </dependency>-->
  55.         <!-- &lt;!&ndash; junit 单元测试 &ndash;&gt;-->
  56.         <!-- <dependency>-->
  57.         <!-- <groupId>junit</groupId>-->
  58.         <!-- <artifactId>junit</artifactId>-->
  59.         <!-- <version>4.12</version>-->
  60.         <!-- </dependency>--> 
  61. </dependencies>

功能实现(不会scala语法直接拷贝的教程代码)

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

Flink 框架集成

Flink 框架介绍

image.png
Apache Spark 是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。
Apache Spark 掀开了内存计算的先河，以内存作为赌注，赢得了内存计算的飞速发展。 但是在其火热的同时，开发人员发现，在 Spark 中，计算框架普遍存在的缺点和不足依然没 有完全解决，而这些问题随着 5G 时代的来临以及决策者对实时数据分析结果的迫切需要而 凸显的更加明显：

  • 数据精准一次性处理（Exactly-Once）
  • 乱序数据，迟到数据
  • 低延迟，高吞吐，准确性
  • 容错性

Apache Flink 是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。在 Spark 火热的同时，也默默地发展自己，并尝试着解决其他计算框架的问题。 慢慢地，随着这些问题的解决，Flink 慢慢被绝大数程序员所熟知并进行大力推广，阿里公 司在 2015 年改进 Flink，并创建了内部分支 Blink，目前服务于阿里集团内部搜索、推荐、 广告和蚂蚁等大量核心实时业务。

创建elasticsearch-flink项目

修改pom文件，增加依赖关系

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.apache.flink</groupId>
  4.         <artifactId>flink-scala_2.12</artifactId>
  5.         <version>1.12.0</version>
  6.     </dependency>
  7.     <dependency>
  8.         <groupId>org.apache.flink</groupId>
  9.         <artifactId>flink-streaming-scala_2.12</artifactId>
  10.         <version>1.12.0</version>
  11.     </dependency>
  12.     <dependency>
  13.         <groupId>org.apache.flink</groupId>
  14.         <artifactId>flink-clients_2.12</artifactId>
  15.         <version>1.12.0</version>
  16.     </dependency>
  17.     <dependency>
  18.         <groupId>org.apache.flink</groupId>
  19.         <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
  20.         <version>1.12.0</version>
  21.     </dependency>
  22.     <!-- jackson -->
  23.     <dependency>
  24.         <groupId>com.fasterxml.jackson.core</groupId>
  25.         <artifactId>jackson-core</artifactId>
  26.         <version>2.11.1</version>
  27.     </dependency>
  28. </dependencies>

功能实现(不会scala语法直接拷贝的教程代码)

  1. package com.atguigu.es;
  2. import org.apache.flink.api.common.functions.RuntimeContext;
  3. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
  6. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
  7. import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
  8. import org.apache.http.HttpHost;
  9. import org.elasticsearch.action.index.IndexRequest;
  10. import org.elasticsearch.client.Requests;
  11. import java.util.ArrayList;
  12. import java.util.HashMap;
  13. import java.util.List;
  14. import java.util.Map;
  16. public class FlinkElasticsearchSinkTest {
  18.     public static void main(String[] args) throws Exception {
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
  22.         List<HttpHost> httpHosts = new ArrayList<>();
  23.          httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
  24.          //httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"));
  25.         // use a ElasticsearchSink.Builder to create an ElasticsearchSink
  26.         ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
  27.             httpHosts,
  28.             new ElasticsearchSinkFunction<String>() {
  29.                  public IndexRequest createIndexRequest(String element) {
  30.                     Map<String, String> json = new HashMap<>();
  31.                     json.put("data", element);
  32.                  return Requests.indexRequest()
  33.                      .index("my-index")
  34.                     //.type("my-type")
  35.                      .source(json);
  36.                  }
  38.                  @Override
  39.                 public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
  40.                      indexer.add(createIndexRequest(element));
  41.                  }
  42.              }
  43.          );
  45.         // configuration for the bulk requests; this instructs the sink to emit after every element, otherwise they would be buffered
  46.          esSinkBuilder.setBulkFlushMaxActions(1);
  48.         // provide a RestClientFactory for custom configuration on the internally created REST client
  49.         // esSinkBuilder.setRestClientFactory(
  50.             // restClientBuilder -> {
  51.                 // restClientBuilder.setDefaultHeaders(...)
  52.                 // restClientBuilder.setMaxRetryTimeoutMillis(...)
  53.                 // restClientBuilder.setPathPrefix(...)
  54.                 // restClientBuilder.setHttpClientConfigCallback(...)
  55.             // }
  56.         // );
  58.          source.addSink(esSinkBuilder.build());
  59.          env.execute("flink-es");
  61.     }
  63. }