通过Logstash复制MySQL数据到ElasticSearch

// docker启动es
docker run -d --name es -p 9200:9200 -p 9300:9300 -v D:\work\iio\dockerFile\es\data:\usr\share\elasticsearch -e "discovery.type=single-node" -m 512M docker.elastic.co/elasticsearch/elasticsearch:7.11.2

// 与容器交互 es为容器名称（也可以使用容器Id）
docker exec -it es /bin/bash

// 容器内创建目录
mkdir /usr/share/elasticsearch/plugins/ik

// 将文件导入指定容器内目录下 es为容器名称（也可以使用容器Id）
docker cp C:\Users\admin\Downloads\elasticsearch-analysis-ik-7.11.2.zip es:/usr/share/elasticsearch/plugins/ik

// 容器内解压压缩包
unzip elasticsearch-analysis-ik-7.11.2.zip

GET _analyze?pretty
 
{
 
  "analyzer": "ik_max_word",
 
  "text":"我们是中国人啊"
 
}

{
    "tokens": [
        {
            "token": "我们",
            "start_offset": 0,
            "end_offset": 2,
            "type": "CN_WORD",
            "position": 0
        },
        {
            "token": "是",
            "start_offset": 2,
            "end_offset": 3,
            "type": "CN_CHAR",
            "position": 1
        },
        {
            "token": "中国人",
            "start_offset": 3,
            "end_offset": 6,
            "type": "CN_WORD",
            "position": 2
        },
        {
            "token": "中国",
            "start_offset": 3,
            "end_offset": 5,
            "type": "CN_WORD",
            "position": 3
        },
        {
            "token": "国人",
            "start_offset": 4,
            "end_offset": 6,
            "type": "CN_WORD",
            "position": 4
        },
        {
            "token": "啊",
            "start_offset": 6,
            "end_offset": 7,
            "type": "CN_CHAR",
            "position": 5
        }
    ]
}

// docker启动logstash
docker run --name logstash -d -p 5044:5044 -v D:\work\iio\dockerFile\logstash\data:\usr\share\logstash logstash:7.11.2

1）修改/config/logstash.yml 中的es地址
2）修改/pipeline/logstash.conf 中的相关配置（input、output、filler）

input {
	jdbc {
        jdbc_connection_string => "jdbc:mysql://mysql地址:端口/数据库名"
        jdbc_user => "账号"
        jdbc_password => "密码"
        #mysql驱动位置
        jdbc_driver_library => "/usr/share/logstash/driver/mysql-connector-java-8.0.23.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        lowercase_column_names => false
        # statement_filepath => "filename.sql"
        statement => "sql语句"
        #定时器（每分钟执行一次）
        schedule => "* * * * *"
        type => "product"
    }
}

output {

     if[type]=="product"{
        elasticsearch {
            hosts => ["es地址:9200"]
            #manage_template => false
            #template_name => "myik"
            #template => "/usr/share/logstash/template/test_template.json"
            #template_overwrite => true
            #document_id => "%{id}"
            #索引名称
            index => "logstash-test"
        }

     }
}