Flink 数据序列化-CFANZ编程社区

MySQL	Elasticsearch	说明
Table	Index	索引（index），就是文档的集合，类似数据库的表（table）
Row	Document	文档（Document），就是一条条的数据，类似数据库中的行（ROW），文档都是JSON格式
Column	Field	字段（Field），就是JSON文档中的字段，类似数据库中的列（Column）
Schema	Mapping	Mapping（映射）是索引中文档的约束，例如字段类型约束。类似数据库的表结构（Schema）
SQL	DSL	DSL是elasticsearch提供的JSON风格的请求语句，用来操作elasticsearch,实现CRUD

架构

安装es

创建网络：es-net是自己取的名字，随便取

加载镜像：使用提供的es.tar，拖到虚拟机的tmp目录下

es.tarhttps://pan.baidu.com/s/13Z74D-liQaDL0_tM-Rl7Rg?pwd=47qm运行加载命令：

运行docker命令：

访问网页：虚拟机ip和9200端口。成功

安装kibana

加载镜像：使用提供的es.tar，拖到虚拟机的tmp目录下

kibanahttps://pan.baidu.com/s/1N3NiLRxLzX42jMxkK9ackQ?pwd=lh2y运行docker命令

访问网页：虚拟机ip和5601端口。成功

点击主页的Explore on my own后，打开Dev Tools

模拟一次请求：可以看到右边的数据和端口9200的数据一模一样

安装ik分词器

分词器

es在创建倒排索引时需要对文档分词;在搜索时，需要对用户输入内容分词。但默认的分词规则对中文处理并不友好。

我们可以看到：右边的数据分词并不友好，比如：世界本应该是一起的，它却分开了

安装ik分词器

查看es-plugins数据卷所在的位置

把ik文件夹放到该路径：Mountpoint就是要的位置

ik文件夹https://pan.baidu.com/s/1EIkGJDvVjcGx06hDUo34Eg?pwd=dads重启es

测试

ik_smart：从最多字开始判断是否切分，若切分，则不会再继续判断已被切分的词是否继续切分

ik_max_word：会判断每个词是否能再继续分

ik分词器的拓展和停用词典

修改一个ik分词器目录中的config目录中的lkAnalyzer.cfg.xml文件

在第一个箭头这行，添加ext.dic：这是要用来拓展词典的文件，要在lkAnalyzer.cfg.xml所在的同级目录下创建出来。

在第二个箭头这行，添加stopword.dic：这是要用来停用词典的文件，在lkAnalyzer.cfg.xml所在的同级目录已经创建好了，不需要再创建。

在ext.dic添加要拓展的词典

在stopword.dic添加要停用的词典

重启es：这次重启后，需要几十秒的时间才能访问网站，否则网站会报错

再次访问网站，可以看到这次测试“奥利给”并没有被分开，并且“的”字并没有出现在右边。

是因为“奥利给”被写进了ext.dic， “的”被写进了stopword.dic

操作索引库

mapping属性

mapping是对索引库中文档的约束，常见的mapping属性包括：

创建索引库

这里创建了一个名叫heima的索引库，mappings代表它是做映射的，properties代表里面是具体的字段，type代表该字段的数据类型，analyzer代表该字段的分词器，index代表该字段是否创建索引

运行该DSL语句后，右边会出现创建的索引库名字，代表创建成功

查询、删除、修改索引库

查看索引库语法

删除索引库语法

索引库和mapping一旦创建无法修改，但是可以添加新的字段

文档操作

新增查询删除文档

新增文档语法

新增文档，右边的数据result为created。成功

查询文档语法

查询文档，右边会出现文档的数据。成功

删除文档语法

删除文档，右边数据result为deleted，成功

修改文档

方法一：全量修改

会删除旧文档，添加新文档

当文档存在时：修改后，右边的数据result为updated。修改成功

当文档不存在时：修改变成了创建。

方法二：增量修改

修改指定的字段，注意只能写指定的字段，而不是把所有字段都写上

修改后，右边的result为updated。修改成功

RestClient的操作

什么是RestClient

下载提供的资料

hotel的demo和sqlhttps://pan.baidu.com/s/1uxl7PzshHu09PXsd9zDWbA?pwd=v8bf在本地新建一个数据库：heima，若使用其他数据库名，记得在demo里修改yml文件

hotel数据结构分析

写出数据库中该表的DSL语句，但是不要执行。我们要使用java来执行

索引库操作

初始化JavaRestClient

在hotel.pom文件中引入es的RestHighLevelclient依赖

因为SpringBoot默认的ES版本是7.6.2，所以我们需要覆盖默认的ES版本

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.10.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>cn.itcast.demo</groupId>
    <artifactId>hotel-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>hotel-demo</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
        <elasticsearch.version>7.12.1</elasticsearch.version>
    </properties>
    <dependencies>
        <!--elasticsearch-->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.4.2</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
		<!--FastJson-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.71</version>
        </dependency>
		<dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

初始化RestHighLevelClient：xxx填虚拟机ip

该代码我写成了

package cn.itcast.hotel;

import net.sf.jsqlparser.statement.create.index.CreateIndex;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.IOException;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

public class HotelIndexTest {
    private RestHighLevelClient client;

    @Test
    void testInit() {
        System.out.println(client);
    }


    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

创建索引库

这个MAPPING_TEMPLATE的位置是要填写DSL语句，但因为太长，我就把它写成了常量

package cn.itcast.hotel;

import net.sf.jsqlparser.statement.create.index.CreateIndex;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.IOException;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

public class HotelIndexTest {
    private RestHighLevelClient client;

    @Test
    void testInit() {
        System.out.println(client);
    }

    @Test
    void createHotelIndex() throws IOException {
        //1、创建Request对象
        CreateIndexRequest request = new CreateIndexRequest("hotel");
        //2、准备请求的参数，DSL语句
        request.source(MAPPING_TEMPLATE, XContentType.JSON);
        //3、发送请求
        client.indices().create(request, RequestOptions.DEFAULT);
    }


    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

package cn.itcast.hotel.constants;

public class HotelConstants {
    public static final String MAPPING_TEMPLATE = "{\n" +
            "  \"mappings\": {\n" +
            "    \"properties\": {\n" +
            "      \"id\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"name\": {\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"address\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"price\": {\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"score\": {\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"brand\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"city\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"starName\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"business\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"location\": {\n" +
            "        \"type\": \"geo_point\"\n" +
            "      },\n" +
            "      \"pic\": {\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"all\": {\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\"\n" +
            "      }\n" +
            "    }\n" +
            "  }\n" +
            "}";
}

运行测试代码，可以看到控制台运行成功。

删除索引库

package cn.itcast.hotel;

import net.sf.jsqlparser.statement.create.index.CreateIndex;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.IOException;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

public class HotelIndexTest {
    private RestHighLevelClient client;

    @Test
    void testInit() {
        System.out.println(client);
    }

    @Test
    void createHotelIndex() throws IOException {
        //1、创建Request对象
        CreateIndexRequest request = new CreateIndexRequest("hotel");
        //2、准备请求的参数，DSL语句
        request.source(MAPPING_TEMPLATE, XContentType.JSON);
        //3、发送请求
        client.indices().create(request, RequestOptions.DEFAULT);
    }

    @Test
    void testDeleteHotelIndex() throws IOException {
        //1、创建Request对象
        DeleteIndexRequest request = new DeleteIndexRequest("hotel");
        //2、发送请求
        client.indices().delete(request, RequestOptions.DEFAULT);
    }


    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

点击该测试，控制台运行成功。

判断索引库是否存在

package cn.itcast.hotel;

import net.sf.jsqlparser.statement.create.index.CreateIndex;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.io.IOException;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

public class HotelIndexTest {
    private RestHighLevelClient client;

    @Test
    void testInit() {
        System.out.println(client);
    }

    @Test
    void createHotelIndex() throws IOException {
        //1、创建Request对象
        CreateIndexRequest request = new CreateIndexRequest("hotel");
        //2、准备请求的参数，DSL语句
        request.source(MAPPING_TEMPLATE, XContentType.JSON);
        //3、发送请求
        client.indices().create(request, RequestOptions.DEFAULT);
    }

    @Test
    void testDeleteHotelIndex() throws IOException {
        //1、创建Request对象
        DeleteIndexRequest request = new DeleteIndexRequest("hotel");
        //2、发送请求
        client.indices().delete(request, RequestOptions.DEFAULT);
    }

    @Test
    void testExistsHotelIndex() throws IOException {
        //1、创建Request对象
        GetIndexRequest request = new GetIndexRequest("hotel");
        //2、发送请求
        boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
        //3、输出
        System.err.println(exists ? "索引库存在" : "索引库不存在");
    }

    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

运行该测试，控制台运行成功，并且打印了“索引库不存在”，因为刚刚删除了索引库

文档操作

初始化JavaRestClient

xxx写虚拟机ip

@SpringBootTest
public class HotelDocumentTest {
    @Autowired
    private IHotelService hotelService;
    private RestHighLevelClient client;


    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

添加文档

我这里的第二步用的是实体类转JSON的方式

package cn.itcast.hotel;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.List;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

@SpringBootTest
public class HotelDocumentTest {
    @Autowired
    private IHotelService hotelService;
    private RestHighLevelClient client;

    @Test
    void testAddDocument() throws IOException {
        //根据id查询酒店数据
        Hotel hotel = hotelService.getById(61083L);
        //转换成文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        //1、准备Request对象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        //2、准备json文档
        request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);
        //3、发送请求
        client.index(request,RequestOptions.DEFAULT);
    }

 
    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

实体类

package cn.itcast.hotel.pojo;

import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
public class HotelDoc {
    private Long id;
    private String name;
    private String address;
    private Integer price;
    private Integer score;
    private String brand;
    private String city;
    private String starName;
    private String business;
    private String location;
    private String pic;

    public HotelDoc(Hotel hotel) {
        this.id = hotel.getId();
        this.name = hotel.getName();
        this.address = hotel.getAddress();
        this.price = hotel.getPrice();
        this.score = hotel.getScore();
        this.brand = hotel.getBrand();
        this.city = hotel.getCity();
        this.starName = hotel.getStarName();
        this.business = hotel.getBusiness();
        this.location = hotel.getLatitude() + ", " + hotel.getLongitude();
        this.pic = hotel.getPic();
    }
}

运行该测试，控制台运行成功

查询文档

package cn.itcast.hotel;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.List;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

@SpringBootTest
public class HotelDocumentTest {
    @Autowired
    private IHotelService hotelService;
    private RestHighLevelClient client;

    @Test
    void testAddDocument() throws IOException {
        //根据id查询酒店数据
        Hotel hotel = hotelService.getById(61083L);
        //转换成文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        //1、准备Request对象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        //2、准备json文档
        request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);
        //3、发送请求
        client.index(request,RequestOptions.DEFAULT);
    }

    @Test
    void testGetDocumentById() throws IOException {
        //1、准备Request
        GetRequest request = new GetRequest("hotel", "61083");
        //2、发送请求，得到响应
        GetResponse response = client.get(request, RequestOptions.DEFAULT);
        //3、解析响应结果
        String json = response.getSourceAsString();
        //反序列化
        HotelDoc hotelDoc = JSON.parseObject(json,HotelDoc.class);
        System.out.println(hotelDoc);
    }

    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

运行该测试，控制台运行成功，并且把数据打印了出来

修改文档

这里我只演示局部更新

package cn.itcast.hotel;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.List;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

@SpringBootTest
public class HotelDocumentTest {
    @Autowired
    private IHotelService hotelService;
    private RestHighLevelClient client;

    @Test
    void testAddDocument() throws IOException {
        //根据id查询酒店数据
        Hotel hotel = hotelService.getById(61083L);
        //转换成文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        //1、准备Request对象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        //2、准备json文档
        request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);
        //3、发送请求
        client.index(request,RequestOptions.DEFAULT);
    }

    @Test
    void testGetDocumentById() throws IOException {
        //1、准备Request
        GetRequest request = new GetRequest("hotel", "61083");
        //2、发送请求，得到响应
        GetResponse response = client.get(request, RequestOptions.DEFAULT);
        //3、解析响应结果
        String json = response.getSourceAsString();
        //反序列化
        HotelDoc hotelDoc = JSON.parseObject(json,HotelDoc.class);
        System.out.println(hotelDoc);
    }

    @Test
    void testUpdateDocument() throws IOException {
        //1、准备Request
        UpdateRequest request = new UpdateRequest("hotel", "61083");
        //2、准备请求参数
        request.doc(
                "price", "952",
                "starName", "四钻"
        );
        //3、发送请求
        client.update(request,RequestOptions.DEFAULT);
    }

   

    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

运行该测试，控制台运行成功

删除文档

package cn.itcast.hotel;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.List;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

@SpringBootTest
public class HotelDocumentTest {
    @Autowired
    private IHotelService hotelService;
    private RestHighLevelClient client;

    @Test
    void testAddDocument() throws IOException {
        //根据id查询酒店数据
        Hotel hotel = hotelService.getById(61083L);
        //转换成文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        //1、准备Request对象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        //2、准备json文档
        request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);
        //3、发送请求
        client.index(request,RequestOptions.DEFAULT);
    }

    @Test
    void testGetDocumentById() throws IOException {
        //1、准备Request
        GetRequest request = new GetRequest("hotel", "61083");
        //2、发送请求，得到响应
        GetResponse response = client.get(request, RequestOptions.DEFAULT);
        //3、解析响应结果
        String json = response.getSourceAsString();
        //反序列化
        HotelDoc hotelDoc = JSON.parseObject(json,HotelDoc.class);
        System.out.println(hotelDoc);
    }

    @Test
    void testUpdateDocument() throws IOException {
        //1、准备Request
        UpdateRequest request = new UpdateRequest("hotel", "61083");
        //2、准备请求参数
        request.doc(
                "price", "952",
                "starName", "四钻"
        );
        //3、发送请求
        client.update(request,RequestOptions.DEFAULT);
    }

    @Test
    void testDeleteDocument() throws IOException {
        //1、准备Request
        DeleteRequest request = new DeleteRequest("hotel", "61083");
        //2、发送请求
        client.delete(request,RequestOptions.DEFAULT);
    }


    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

运行该测试，控制台运行成功

批量导入文档

我这里改成这样，是因为我要把该表的所有数据都导入文档

package cn.itcast.hotel;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.List;

import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE;

@SpringBootTest
public class HotelDocumentTest {
    @Autowired
    private IHotelService hotelService;
    private RestHighLevelClient client;

    @Test
    void testAddDocument() throws IOException {
        //根据id查询酒店数据
        Hotel hotel = hotelService.getById(61083L);
        //转换成文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        //1、准备Request对象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        //2、准备json文档
        request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);
        //3、发送请求
        client.index(request,RequestOptions.DEFAULT);
    }

    @Test
    void testGetDocumentById() throws IOException {
        //1、准备Request
        GetRequest request = new GetRequest("hotel", "61083");
        //2、发送请求，得到响应
        GetResponse response = client.get(request, RequestOptions.DEFAULT);
        //3、解析响应结果
        String json = response.getSourceAsString();
        //反序列化
        HotelDoc hotelDoc = JSON.parseObject(json,HotelDoc.class);
        System.out.println(hotelDoc);
    }

    @Test
    void testUpdateDocument() throws IOException {
        //1、准备Request
        UpdateRequest request = new UpdateRequest("hotel", "61083");
        //2、准备请求参数
        request.doc(
                "price", "952",
                "starName", "四钻"
        );
        //3、发送请求
        client.update(request,RequestOptions.DEFAULT);
    }

    @Test
    void testDeleteDocument() throws IOException {
        //1、准备Request
        DeleteRequest request = new DeleteRequest("hotel", "61083");
        //2、发送请求
        client.delete(request,RequestOptions.DEFAULT);
    }

    @Test
    void testBulkRequest() throws IOException {
        //批量查询酒店数据
        List<Hotel> hotels = hotelService.list();
        //1、创建Request
        BulkRequest request = new BulkRequest();
        //2、准备参数，添加多个新增的Request
        for(Hotel hotel:hotels){
            //转换为文档类型HotelDoc
            HotelDoc hotelDoc = new HotelDoc(hotel);
            //创建新增文档的Request对象
            request.add(new IndexRequest("hotel")
            .id(hotelDoc.getId().toString())
            .source(JSON.toJSONString(hotelDoc),XContentType.JSON));
        }
        //3、发送请求
        client.bulk(request,RequestOptions.DEFAULT);
    }

    @BeforeEach
    void setUp() {
        this.client = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://xxx.xxx.xxx.xxx:9200")
        ));
    }

    @AfterEach
    void tearDown() throws IOException {
        this.client.close();
    }
}

运行该测试，控制台运行成功，并且可以看到导入了201条数据，正好表的数据量