Springboot整合Elasticsearch并用Logstash实现数据库数据同步-开发者社区-阿里云

Springboot整合Elasticsearch、搭建Logstash同步数据

2021-09-03 2020

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 个人学习笔记——ES

整合Springboot

通过虚拟机搭建ES，这里使用的版本是6.4.3，引入相应依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
    # 使用2.2.2是为了对应ES的版本
    <version>2.2.2.RELEASE</version>
</dependency>

引入依赖之后可以自行查看

spring:
  data:
    elasticsearch:
      # 在es中配置的名称
      cluster-name: es
      # 如果是集群，用,分隔
      cluster-nodes: 192.168.1.7:9300

测试实体类

package com.csea.entity;

import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;

/**
 * @author Csea
 * @title
 */
@Data
@Document(indexName = "merchant", type = "doc")
public class Merchant {

    @Id
    private Long merchantId;
    // store = true 表示这是要存储的字段
    @Field(store = true)
    private String name;

    @Field(store = true)
    private String mob;

    @Field(store = true)
    private String address;

    @Field(store = true)
    private String descr;
}

创建索引（文档）

当索引不存在时候，会先创建索引，并将数据插入。

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;
    @Test
    public void createIndex() {
        Merchant merchant = new Merchant();
        merchant.setMerchantId(1001L);
        merchant.setName("Csea-杂货铺1");
        merchant.setAddress("大道8888号");
        merchant.setDescr("好吃的不得了");
        merchant.setMob("13899999999");
        IndexQuery indexQuery = new IndexQueryBuilder().withObject(merchant).build();
        elasticsearchTemplate.index(indexQuery);
    }

执行完之后可以看到merchant索引已经创建，并且文档也新增了。

更新文档

    @Test
    public void updateMerchantDoc() {
        Map<String, Object> sourceMap = new HashMap<>();
        sourceMap.put("name", "Csea-杂货铺-update");
        sourceMap.put("adress", "光明大道8888号");
        sourceMap.put("descr", "好吃你就多吃点！~");
        IndexRequest indexRequest = new IndexRequest();
        indexRequest.source(sourceMap);

        UpdateQuery query = new UpdateQueryBuilder().withClass(Merchant.class)
                .withId("1001")
                .withIndexRequest(indexRequest)
                .build();
        elasticsearchTemplate.update(query);
    }

查询文档

    @Test
    public void queryMerchantDoc() {
        GetQuery query = new GetQuery();
        query.setId("1001");
        Merchant merchant = elasticsearchTemplate.queryForObject(query, Merchant.class);
        log.info("查询到的数据是={}", merchant);
    }

com.test.ESTest : 查询到的数据是=Merchant(merchantId=1001, name=Csea-杂货铺-update, mob=13899999999, address=大道8888号, descr=好吃你就多吃点！~)

删除文档数据

    @Test
    public void delMerchantDoc() {
        elasticsearchTemplate.delete(Merchant.class, "1001");
    }

分页文档查询

    @Test
    public void searchPageMerchantDoc() {

        Pageable pageable = PageRequest.of(0, 20);

        SearchQuery query = new NativeSearchQueryBuilder()
                .withQuery(QueryBuilders.matchQuery("description", "God Engineer "))
                .withPageable(pageable)
                .build();
        AggregatedPage<Merchant> merchants = elasticsearchTemplate.queryForPage(query, Merchant.class);
        List<Merchant> content = merchants.getContent();
        for (Merchant merchant : content) {
            log.info("查询到的文档为：{}", merchant);
        }

    }

高亮分页查询

    @Test
    public void searchHightMerchantDoc() {

        String preTag = "<font color='red'>";
        String postTg = "</font>";

        Pageable pageable = PageRequest.of(0, 20);
        // 排序
        SortBuilder sortBuilder = new FieldSortBuilder("name")
                .order(SortOrder.DESC);


        SearchQuery query = new NativeSearchQueryBuilder()
                // 查询的字段
                .withQuery(QueryBuilders.matchQuery("mob", "88"))
                .withHighlightFields(new HighlightBuilder.Field("mob")
                        .preTags(preTag)
                        .postTags(postTg))
                .withSort(sortBuilder)
                .withPageable(pageable)
                .build();
        AggregatedPage<Merchant> merchants = elasticsearchTemplate.queryForPage(query, Merchant.class,
                new SearchResultMapper() {
                    @Override
                    public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> aClass, Pageable pageable) {
                        SearchHits hits = response.getHits();

                        List<Merchant> stuList = new ArrayList<>();
                        for (SearchHit hit : hits) {
                            HighlightField highlightField = hit.getHighlightFields().get("mob");
                            String mob = highlightField.getFragments()[0].toString();

                            Object merchantId = hit.getSourceAsMap().get("merchantId");
                            String name = (String) hit.getSourceAsMap().get("name");
                            String descr = (String) hit.getSourceAsMap().get("descr");
                            String address = (String) hit.getSourceAsMap().get("address");

                            Merchant merchantHL = new Merchant();
                            merchantHL.setDescr(descr);
                            merchantHL.setMerchantId(Long.valueOf(merchantId.toString()));
                            merchantHL.setName(name);
                            merchantHL.setMob(mob);
                            merchantHL.setAddress(address);
                            stuList.add(merchantHL);

                        }

                        if (!stuList.isEmpty()) {
                            return new AggregatedPageImpl<>((List<T>) stuList);
                        }

                        return null;
                    }
                });
        log.info("分页总数：{}", merchants.getTotalPages());
        List<Merchant> content = merchants.getContent();
        for (Merchant merchant : content) {
            log.info("分页文档数据：{}", merchant);
        }

    }

删除索引

    @Test
    public void deleteDelMerchantIndex() {
        elasticsearchTemplate.deleteIndex(Merchant.class);
    }

小结

不建议使用ElasticsearchTemplate对索引进行管理（创建索引、更新映射、删除索引），因为就像使用Mysql，不会通过java代码去改变表结构，ES也是如此，更多是用来进行CRUD操作。

Logstash

概念

作为数据采集的工具；
以id或update_time作为同步边界；

以id同步：初次同步的时候会将所有数据同步过来，之后logstash的定时任务回去检查，比如上次同步到id为2000的数据，那么这次就同步2000之后的数据，使用id作为同步有很大的弊端，就只只能新增数据，无法更新。
以update_time同步：初次同步就将所有的数据同步过来，之后如果有新增或更新的操作，那么就把以上一次更新时间为界，之后的数据全部做一次新增或更新。

使用logstash-input-jdbd插件同步；
使用logstash需要跟ES的版本保持一致，比如两者都要是6.4.3版本；
同步数据时，需要事先创建索引。

安装配置

logstash下载地址
上传并解压，再上传mysql的驱动jar包，以及jdk（logstash需要jdk）

cd进入logstash目录之后，创建文件夹

mkdir sync

进入sync目录，创建配置文件

# 先创建同步配置文件
vim logstash-db-sync.conf
# 拷贝数据库驱动到该路径下
cp /usr/local/mysql-connector-java-8.0.13.jar .

修改logstash-db-sync.conf

input {
    jdbc {
        # 设置Mysql/MariaDB 数据库url以及数据库名称
        jdbc_connection_string => "jdbc:mysql://192.168.1.6:3306/xxg?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false"
        # 数据库用户名密码
        jdbc_user => "root"
        jdbc_password => "123456"
        # 数据库驱动所在位置，可以是绝对路径或相对路径
        #jdbc_driver_library => "/usr/local/logstash-6.4.3/sync/mysql-connector-java-5.1.49.jar"
        jdbc_driver_library => "/usr/local/logstash-6.4.3/sync/mysql-connector-java-8.0.13.jar"
        # 驱动类名
        #jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
        # 开启分页
        jdbc_paging_enabled => "true"
        # 分页每页数量，可以自定义
        jdbc_page_size => "10000"
        # 执行的sql文件路径
        statement_filepath => "/usr/local/logstash-6.4.3/sync/product.sql"
        # 设置定时任务间隔
        schedule => "* * * * *"
        # 索引类型
        type => "_doc"
        # 是否开启记录上次追踪的结构，也就是上次更新的实际，这个会记录到 last_run_metadata_path 的文件
        use_column_value => true
        # 记录上一次追踪的结果值
        last_run_metadata_path => "/usr/local/logstash-6.4.3/sync/track_time"
        # 如果 use_column_value 为true，配置本参数，追踪的colume名，可以是自增id或者时间
        tracking_column => "update_time"
        # tracking_column 对应字段的类型
        tracking_column_type => "timestamp"
        # 是否清除 last_run_metadata_path 的记录，true则每次都从头开始查询所有
        clean_run => false
        # 数据库字段名称大写转小写
        lowercase_column_names => false
    }
}

output {
    elasticsearch {
        # es地址
        hosts => ["192.168.1.7:9200"]
        # 同步索引名
        index => "product"
        # 设置_docID和数据相同，按照执行sql中的字段来
        document_id => "%{product_id}"
    }
    # 日志输出
    stdout {
        codec => json_lines
    }
}

添加执行同步查询的sql脚本

vim product.sql

SELECT
    * 
FROM
    jh_product 
WHERE
    is_delete = 0 
    AND update_time >= :sql_last_value

:sql_last_value是logstash中要赋值的地方，可以理解为占位符

启动logstash，先cd到目录下的bin目录

./logstash -f /usr/local/logstash-6.4.3/sync/logstash-db-sync.conf

启动之后就可以看到数据同步过去，在ES的索引中查看数据

自定义模板

因为需要用到中文分词，所以需要自定义模板
Get：http://{IP}:{port}/_template/logstash
查看模板信息

将拿到的template复制出来进行自定义修改

{
    "order": 10,
    "version": 1,
    "index_patterns": ["*"],
    "settings": {
        "index": {
            "refresh_interval": "5s"
        }
    },
    "mappings": {
        "_default_": {
            "dynamic_templates": [
                {
                    "message_field": {
                        "path_match": "message",
                        "match_mapping_type": "string",
                        "mapping": {
                            "type": "text",
                            "norms": false
                        }
                    }
                },
                {
                    "string_fields": {
                        "match": "*",
                        "match_mapping_type": "string",
                        "mapping": {
                            "type": "text",
                            "norms": false,
                            "analyzer": "ik_max_word",
                            "fields": {
                                "keyword": {
                                    "type": "keyword",
                                    "ignore_above": 256
                                }
                            }
                        }
                    }
                }
            ],
            "properties": {
                "@timestamp": {
                    "type": "date"
                },
                "@version": {
                    "type": "keyword"
                },
                "geoip": {
                    "dynamic": true,
                    "properties": {
                        "ip": {
                            "type": "ip"
                        },
                        "location": {
                            "type": "geo_point"
                        },
                        "latitude": {
                            "type": "half_float"
                        },
                        "longitude": {
                            "type": "half_float"
                        }
                    }
                }
            }
        }
    },
    "aliases": {}
}

创建完毕之后，在logstash目录的sync下创建新的json文件

vim logstash-ik.json

将自定义模板内容放入json文件中

然后修改logstash-db-sync.conf 配置，在output中添加如下配置，保存之后，重启logstash即可

output {
    elasticsearch {
        # 定义模板名称
        template_name => "myik"
        # 模板所在位置
        template => "/usr/local/logstash-6.4.3/sync/logstash-ik.json"
        # 重写模板
        template_overwrite => true
        #默认为true，false关闭logstash自动管理模板功能，如果自定义模板，则设置false
        manage_template => false
    }
}

中文分词不生效

如果出现中文分词"analyzer": "ik_max_word"不生效的情况可以先将manage_template改为true，然后启动logstash同步，同步之后，使用：
Get：http://{IP}:{port}/_template/{template名称}
来查看模板是否已经上传到es

同步之后将manage_template改为false，重新同步即可。
主要就是模板没有被上传到ES导致的。
另外可以修改模板中order的值，值越大，在 merge 规则的时候优先级越高。

Springboot整合Elasticsearch、搭建Logstash同步数据

整合Springboot

测试实体类

创建索引（文档）

更新文档

查询文档

删除文档数据

分页文档查询

高亮分页查询

删除索引

小结

Logstash

概念

安装配置

自定义模板

中文分词不生效

热门文章

最新文章

相关课程

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

Springboot整合Elasticsearch、搭建Logstash同步数据

整合Springboot

测试实体类

创建索引（文档）

更新文档

查询文档

删除文档数据

分页文档查询

高亮分页查询

删除索引

小结

Logstash

概念

安装配置

自定义模板

中文分词不生效

热门文章

最新文章

相关课程

相关电子书