解压：
unzip logstash-7.3.2.zip

cd /usr/local/logstash-7.3.2/bin
mkdir MysqL
cd MysqL

第一个：数据传输调度配置
vim dataTran.conf

input {
 stdin { }
    jdbc {
        type => "lgb_test"
        codec => plain { charset => "UTF-8"}
        lowercase_column_names => false
        jdbc_connection_string => "jdbc:MysqL://127.0.0.1:3306/test"

        jdbc_user => "root"

        jdbc_password => "123456"

        jdbc_driver_library => "/usr/local/logstash-7.3.2/lib/MysqL-connector-java-8.0.13.jar"

        jdbc_driver_class => "com.MysqL.cj.jdbc.Driver"

        jdbc_paging_enabled => "true"
        #添加增量类配置
        record_last_run => "true"
        use_column_value => "true"
        tracking_column => "event_id"
        tracking_column_type => numeric
        clean_run => "false"
        last_run_Metadata_path => "/usr/local/logstash-7.3.2/bin/MysqL/sql_last_value.txt"
        jdbc_page_size => "50000"
        #获取到记录的SQL查询语句
        statement_filepath => "/usr/local/logstash-7.3.2/bin/MysqL/jdbc.sql"
        #定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每>分钟都更新
        schedule => "10 * * * *"
    }

 }

 output {
     stdout {
        codec => json_lines
    }

    if[type]=="lgb_test"{
        elasticsearch {
          #ESIP地址与端口
          hosts => "localhost:9200"
          #ES索引名称（自己定义的）
          index => "lgb_index_test"
          #文档类型
          document_type => "lgb_type_test"
          #文档类型id
          document_id => "%{event_id}"
      }
    }

  }

第二个：查询sql语句
vim jdbc.sql

select event_id,timestmp,formatted_message,level_string,caller_class from logging_event where event_id > :sql_last_value ORDER BY timestmp

第三个,创建sql_last_value.txt，用于保存追踪列的最后一条记录

touch sql_last_value.txt

切换目录：
cd /usr/local/logstash-7.3.2/bin

启动：
./logstash -f MysqL/dataTran.conf --path.data=/root/