es-5数据同步

向ES里面添加数据

1. logstash
2. kabina
3. javaapi
4. restful api

重点介绍一下logstash，是一个数据同步工具，可以动态地将来自不同数据源的数据统一起来，并将数据标准化到其他任何数据源。

其实最常见的使用场景是将各个数据源同时同步到es。有点像八爪鱼转换器：

举例说明：

1. mysql->mysql
   https://blog.csdn.net/xyh930929/article/details/10119773
2. mysql->es，只要select后面的字段与es里面field的字段对上就行。

input {
    #stdin {
    #}
    jdbc {
        jdbc_connection_string => "jdbc:mysql://localhost:3306/XXX?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=utf8&useSSL=false&allowMultiQueries=true"
        jdbc_user => "matrix_db"
        jdbc_password => "c3I44dfWkm&CyfdNiL2rp"
        jdbc_driver_library => "/opt/elastic/logstash-7.3.1/mysql-connector-java-5.1.35-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_default_timezone =>"Asia/Shanghai"
        statement => "SELECT id,Title,Description from OP_Interview WHERE LastUpdateDate > :sql_last_value"
        schedule => "* * * * *"
        type => "interview"
        record_last_run => true
        use_column_value => true
        tracking_column => "LastUpdateDate"
        tracking_column_type => "timestamp"
        last_run_metadata_path => "./last_record/mini_interview_last_v9_time"
        clean_run => false
        lowercase_column_names => false
    }
}

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

output {
    if[type] == "interview"{
        elasticsearch {
            hosts => "http://localhost:9200"
            index => "mini_matrix_v9"
            document_id => "%{id}"
        }
    }
    stdout {
        codec => json_lines
    }
}

3. csv->es

input
{
    file{
        path => ["/home/m2/yanhui/220506_dp_brandlist.csv"]
        start_position => "beginning"
    }
}
filter{
    csv{
        separator => ","
             columns => ["out_brand"]
       }
    mutate{
            convert => {
                "out_brand" => "string"
        }
    }
}
output{
    elasticsearch{
            hosts => ["localhost:9200"]
            index => "brand_dp"
    }