配置logstash拉取myql8.0.18数据库中的数据到elastcisearch中

input {
    stdin {
    }
    jdbc {
      jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false"
      jdbc_user => "mysqluser"
      jdbc_password => "mysqlpass" 
      jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      
      statement => "SELECT * FROM font_element WHERE updated_at > :sql_last_value"
 
      jdbc_default_timezone =>"Asia/Shanghai"
      record_last_run => "true"

      use_column_value => "true"
      tracking_column => "updated_at"
      tracking_column_type => "timestamp"

      last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_font_element"  
      clean_run => "false"
      schedule => "*/3 * * * *"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      type => "font-element"
    }
    jdbc {
        jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false"
        jdbc_user => "mysqluser"
        jdbc_password => "mysqlpass" 
        jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_default_timezone =>"Asia/Shanghai"
        
        statement => "SELECT i.id, i.guid, i.thumbnail, i.property, i.updated_at, i.created_at, kg.keywords, ec.category_name FROM image_element i left join
(
  select element_id, GROUP_CONCAT(keyword separator ' ') as keywords from image_element_keywords as ie INNER JOIN keywords as k on ie.keywords_id = k.id GROUP BY element_id
) as kg on i.id = kg.element_id left JOIN element_category as ec on i.category_id = ec.id WHERE i.updated_at > :sql_last_value"
 
        record_last_run => "true"
        use_column_value => "true"
        tracking_column => "updated_at"
        tracking_column_type => "timestamp"

        last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_image_element"  
        clean_run => "false"
        schedule => "*/3 * * * *"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        type => "image-element"
    }
    jdbc {
        jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false"
        jdbc_user => "mysqluser"
        jdbc_password => "mysqlpass" 
        jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_default_timezone =>"Asia/Shanghai"
        
        statement => "SELECT i.id, i.guid, i.thumbnail, i.property, i.updated_at, i.created_at, kg.keywords, '' as category_name FROM background_element as i left join
(
  select element_id, GROUP_CONCAT(keyword separator ' ') as keywords from background_element_keywords as ie INNER JOIN keywords as k on ie.keywords_id = k.id GROUP BY element_id
) as kg on i.id = kg.element_id WHERE i.updated_at > :sql_last_value"
 
        record_last_run => "true"
        use_column_value => "true"
        tracking_column => "updated_at"
        tracking_column_type => "timestamp"

        last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_background_element"  
        clean_run => "false"
        schedule => "*/3 * * * *"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        type => "background-element"
    }
    jdbc {
        jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false"
        jdbc_user => "mysqluser"
        jdbc_password => "mysqlpass" 
        jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_default_timezone =>"Asia/Shanghai"
        
        statement => "SELECT i.id, i.guid, i.thumbnail, i.property, i.updated_at, i.created_at, kg.keywords, ec.category_name FROM shape_element as i left join
(
  select element_id, GROUP_CONCAT(keyword separator ' ') as keywords from shape_element_keywords as ie INNER JOIN keywords as k on ie.keywords_id = k.id GROUP BY element_id
) as kg on i.id = kg.element_id left JOIN element_category as ec on i.category_id = ec.id WHERE i.updated_at > :sql_last_value"
 
        record_last_run => "true"
        use_column_value => "true"
        tracking_column => "updated_at"
        tracking_column_type => "timestamp"

        last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_shape_element"  
        clean_run => "false"
        schedule => "*/3 * * * *"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        type => "shape-element"
    }
    jdbc {
        jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false"
        jdbc_user => "mysqluser"
        jdbc_password => "mysqlpass" 
        jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_default_timezone =>"Asia/Shanghai"
        
        statement => "SELECT id, keyword, category_name, updated_at FROM keywords WHERE updated_at > :sql_last_value"
 
        record_last_run => "true"
        use_column_value => "true"
        tracking_column => "updated_at"
        tracking_column_type => "timestamp"

        last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_keywords"
        clean_run => "false"
        schedule => "*/3 * * * *"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        type => "keywords"
    }
    jdbc {
        jdbc_connection_string => "jdbc:mysql://172.30.0.109:3306/graphics?useSSL=false"
        jdbc_user => "mysqluser"
        jdbc_password => "mysqlpass" 
        jdbc_driver_library => "/usr/local/elk/logstash-7.4.2/config/mysql-connector-java-8.0.18.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_default_timezone =>"Asia/Shanghai"
        
        statement => "select t.id, t.guid, t.title, t.property, t.user_id, t.user_name, t.thumbnail, t.description, t.updated_at, kg.keywords, tc.category_name from template as t left join 
(
 select template_id, GROUP_CONCAT(keyword separator ' ') as keywords from template_keywords as tk INNER JOIN keywords as k on tk.keywords_id = k.id GROUP BY template_id
) as kg on kg.template_id = t.id inner JOIN template_category as tc on t.category_id = tc.id where t.on_market = 1 and t.updated_at > :sql_last_value"
 
        record_last_run => "true"
        use_column_value => "true"
        tracking_column => "updated_at"
        tracking_column_type => "timestamp"

        last_run_metadata_path => "/usr/local/elk/logstash-7.4.2/config/logstash_jdbc_last_run_template"  
        clean_run => "false"
        schedule => "*/3 * * * *"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        type => "template"
    }
}
 
filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}
 
output {
    if[type] == "font-element" {
        elasticsearch {
            hosts => "172.30.0.106:9200"
            index => "logstash-ex-font_element"
            document_id => "%{id}"
            user => "elastic"
            password => "pass"
        }
    }
    if[type] == "image-element" {
        elasticsearch {
            hosts => "172.30.0.106:9200"
            index => "element"
            document_id => "image_element_%{id}"
            user => "elastic"
            password => "pass"
        }
    }
    if[type] == "background-element" {
        elasticsearch {
            hosts => "172.30.0.106:9200"
            index => "element"
            document_id => "background_element_%{id}"
            user => "elastic"
            password => "pass"
        }
    }
    if[type] == "shape-element" {
        elasticsearch {
            hosts => "172.30.0.106:9200"
            index => "element"
            document_id => "shape_element_%{id}"
            user => "elastic"
            password => "pass"
        }
    }
    if[type] == "keywords" {
        elasticsearch {
            hosts => "172.30.0.106:9200"
            index => "keywords"
            document_id => "keywords_%{id}"
            user => "elastic"
            password => "pass"
        }
    }
    if[type] == "template" {
        elasticsearch {
            hosts => "172.30.0.106:9200"
            index => "template"
            document_id => "template_%{id}"
            user => "elastic"
            password => "pass"
            template_overwrite => true
            template => "/usr/local/elk/logstash-7.4.2/config/template-es.json"
        }
    }
    stdout {
        codec => json_lines
    }
}

[Unit]
Description=logstash
[Service]
Type=simple
User=logstash
Group=logstash
Environment=JAVA_HOME=/usr/local/elk/elasticsearch-7.4.2/jdk
Environment=LS_HOME=/usr/local/elk/logstash-7.4.2
Environment=LS_SETTINGS_DIR=/usr/local/elk/logstash-7.4.2/config/
Environment=LS_PIDFILE=/usr/local/elk/logstash-7.4.2/logstash.pid
Environment=LS_USER=logstash
Environment=LS_GROUP=logstash
Environment=LS_GC_LOG_FILE=/usr/local/elk/logstash-7.4.2/logs/gc.log
Environment=LS_OPEN_FILES=16384
Environment=LS_NICE=19
Environment=SERVICE_NAME=logstash
Environment=SERVICE_DESCRIPTION=logstash
ExecStart=/usr/local/elk/logstash-7.4.2/bin/logstash "--path.settings" "/usr/local/elk/logstash-7.4.2/config/"
Restart=always
WorkingDirectory=/usr/local/elk/logstash-7.4.2
Nice=19
LimitNOFILE=16384


[Install]
WantedBy=multi-user.target

{
    "template": "*",
    "version": 50001,
    "settings": {
        "index.refresh_interval": "5s",
        "analysis": {
            "analyzer": {
                "comma": {
                     "type": "pattern",
                     "pattern":","
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "id": {
                "type": "long"
            },
            "template_name": {
                "type": "text",
                "analyzer": "standard"
            },
            "template_data": {
                "type": "text",
                "analyzer": "standard"
            },
            "author": {
                "type": "text",
                "analyzer": "standard"
            },
            "keywords": {
                "type": "text",
                "analyzer": "comma"
            },
            "category_ids": {
                "type": "text",
                "analyzer": "comma"
            },
            "thumb_id": {
                "type": "long"
            },
            "description": {
                "type": "text",
                "analyzer": "standard"
            },
            "updated_at": {
                "type": "date"
            },
            "created_at": {
                "type": "date"
            },
            "@timestamp": {  
               "type": "date"
            },
            "@version": {
              "type": "keyword"
            }
        }
    }
}