Logstash入门

た入场券 2023-02-20 02:53 1阅读 0赞

# 1. Logstash介绍 #

Logstash用于数据的采集、转换和输出，采集是指从MySQLD等任意的数据源采集数据，然后输出到ES中

*  以id或update\_time作为同步边界
 *  使用logstash-input-jdbc插件进行同步
 *  进行数据同步之前，我们要预先创建索引

## 2. Logstash安装与配置，并同步Mysql数据到ES中 ##

1.  解压`tar -zxvf logstash-7.8.0.tar.gz`
2.  移动`mv logstash-7.8.0 /usr/local/`
3.  进入到logstash-7.8.0文件夹并创建自定义文件夹`mkdir sync`，所有的配置文件都存放在sync文件夹中
4.  在sync文件夹中创建文件`vi logstash-db-sync.conf`，配置文件内容如下：
    
        input { 
            jdbc { 	
        		jdbc_connection_string => "jdbc:mysql://10.4.45.85:3306/vant_mall?useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8"
                jdbc_user => "root"
                jdbc_password => "root"
                jdbc_driver_library => "/usr/local/logstash-7.8.0/sync/mysql-connector-java-8.0.20.jar"
                jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
                jdbc_paging_enabled => "true"
                jdbc_page_size => "1000"
                statement_filepath => "/usr/local/logstash-7.8.0/sync/search.sql"
                schedule => "* * * * *"
        	    type => "_doc"
        	    use_column_value => true
        	    last_run_metadata_path => "/usr/local/logstash-7.8.0/sync/track_time"
                tracking_column => "update_time"
        	    tracking_column_type => "timestamp"
        	    clean_run => false
        	    lowercase_column_names => false
            }
         }
        
        output { 
        	elasticsearch { 
        		hosts => ["10.4.53.89:9200"]
        	    index => "vant-mall"
        	    document_id => "%{id}"
             }
            stdout { 
        	    codec => json_lines
            }
        }
5.  拷贝mysql-connector-java-8.0.20.jar驱动到sync目录中
6.  在search.sql中写好你要采集的数据的sql语句，语句中必须包含 >= :sql\_last\_value，这个值是给Logstash看的，logstash可以根据这个边界值去查询最新的数据，如下所示：
    
        SELECT
        	p.id,
        	p.title,
        	p.sub_title AS subTitle,
        	p.img_main_url AS imgMainUrl,
        	k.min_price AS minPrice,
        	p.total_sales AS totalSales,
        	p.tag,
        	p.update_time AS updateTime 
        FROM
        	mall_spu p
        	LEFT JOIN ( SELECT spu_id, MIN( price ) AS min_price FROM mall_sku groupby spu_id ) k ON p.id = k.spu_id 
        WHERE
        	p.is_sale = 1 
        	AND p.update_time >= :sql_last_value
7.  自定义模板配置ik分词器，在sync文件夹下创建logstash-ik.json文件，文件内容如下：
    
        { 
                "order": 0,
                "version": 1,
                "index_patterns": ["*"],
                "settings": { 
                    "index": { 
                        "refresh_interval": "5s"
                    }
                },
                "mappings": { 
                    "_default_": { 
                        "dynamic_templates": [
                            { 
                                "message_field": { 
                                    "path_match": "message",
                                    "mapping": { 
                                        "norms": false,
                                        "type": "text"
                                    },
                                    "match_mapping_type": "string"
                                }
                            },
                            { 
                                "string_fields": { 
                                    "mapping": { 
                                        "norms": false,
                                        "type": "text",
                                        “analyzer”: "ik_max_word",
                                        "fields": { 
                                            "keyword": { 
                                                "ignore_above": 256,
                                                "type": "keyword"
                                            }
                                        }
                                    },
                                    "match_mapping_type": "string",
                                    "match": "*"
                                }
                            }
                        ],
                        "properties": { 
                            "@timestamp": { 
                                "type": "date"
                            },
                            "geoip": { 
                                "dynamic": true,
                                "properties": { 
                                    "ip": { 
                                        "type": "ip"
                                    },
                                    "latitude": { 
                                        "type": "half_float"
                                    },
                                    "location": { 
                                        "type": "geo_point"
                                    },
                                    "longitude": { 
                                        "type": "half_float"
                                    }
                                }
                            },
                            "@version": { 
                                "type": "keyword"
                            }
                        }
                    }
                },
                "aliases": { }
        }
8.  在logstash-db-sync.conf中output中的elasticsearch大括号内加入如下内容：
    
        template_name => "logstash-ik"
        template => "/usr/local/logstash-6.8.10/sync/logstash-ik.json"
        template_overwrite => true
        manage_template => false
9.  启动`./logstash -f /usr/local/logstash-6.8.10/sync/logstash-db-sync.conf`