安装好后在bin目录下新建mysql.conf，配置一下内容：

input {
    jdbc {
      # mysql数据库连接
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/esdemo?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai"
      # mysqly用户名和密码
      jdbc_user => "root"
      jdbc_password => "123456"
      # 驱动配置
      jdbc_driver_library => "D:\elasticsearch\logstash-8.2.0\mysql-connector-java-5.1.7-bin.jar"
      # 驱动类名
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      # 数据库重连尝试次数
      connection_retry_attempts => "3"
      # 判断数据库连接是否可用，默认false不开启
      jdbc_validate_connection => "true"
      # 数据库连接可用校验超时时间，默认3600S
      jdbc_validation_timeout => "3600"
      # 开启分页查询（默认false不开启）；
      jdbc_paging_enabled => "true"
      # 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
      jdbc_page_size => "50000"
      # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
      # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
      # 执行指定的sql文件
      #statement_filepath => "D:\service\logstash-7.7.0\mysql\sql\fa_search.sql"
      #执行的sql语句
      statement => "SELECT * FROM syshl1 where id > :sql_last_value order by id asc"
      # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
      lowercase_column_names => false
      # Value can be any of: fatal,error,warn,info,debug，默认info；
      sql_log_level => warn
      # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
      record_last_run => true
      # 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
      use_column_value => true
      # 需要记录的字段，用于增量同步，需是数据库字段
      tracking_column => "id"
      #追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型()
      tracking_column_type => "numeric"
      # record_last_run上次数据存放位置；
      last_run_metadata_path => "D:\elasticsearch\logstash-8.2.0\last_id.txt"
      # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
      clean_run => false
      # 设置监听 各字段含义 分 时 天 月  年 ，默认全部为*代表含义：每分钟都更新
      schedule => "* * * * *"
      # 索引类型
      #type => "blog"
    }
}

filter {
   json {
        source => "message"
        remove_field => ["message"]
    }
}

output {

    elasticsearch {
        #es服务器
        hosts => ["localhost:9200"]
        #ES索引名称
        index => "syshl"
        #自增ID
        document_id => "%{id}"
    }
    
    stdout {
        codec => json_lines
    }
}

然后通过 logstash -f bin/mysql.conf 去启动；
成功后开始同步数据