logstash 怎么将数据导入ES

Posted 2023-05-03

参考技术A 在配置文件
input
file
type => "json"
path => "/home/hadoop/xinwang_XW351464_2110"


output
elasticsearch
cluster => "es_master"
#manage_template => false
embedded => true
index => "huhu"


版本是logstash-1.3.2，es版本是1.1.1
错误提示
java -jar logstash-1.3.2-flatjar.jar agent -f ogstash-syslog.conf
Using milestone 2 input plugin 'file'. This plugin should be stable, but if you see strange behavior, please let us know! For more information on plugin milestones, see http://logstash.net/docs/1.3.2/plugin-milestones :level=>:warn
log4j, [2014-10-31T11:55:01.977] WARN: org.elasticsearch.discovery: [Jocasta] waited for 30s and no initial state was set by the discovery
Unable to check template. Automatic template management disabled. :error=>"waited for [30s]",
Redhat 5.7 64bit / CentOS 5.x
JDK 1.6.0_45
logstash 1.3.2 (内带kibana)
elasticsearch 0.90.10
redis 2.8.4
对应下版本本回答被提问者和网友采纳 参考技术B

Redhat 5.7 64bit / CentOS 5.x

JDK 1.6.0_45

logstash 1.3.2 (内带kibana)

elasticsearch 0.90.10

redis 2.8.4

对应下版本即可实现将数据导入ES。

使用LOGSTASH 将数据导入到ES

logstash 执行过程

input -->filter -->output 

filter 可以对数据进行处理

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 输出插件

 

 

 codec plugin

 

 

 

 

使用脚本将数据导入到ES

input {
  jdbc {
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/db_example"
    jdbc_user => root
    jdbc_password => ymruan123
    #启用追踪，如果为true，则需要指定tracking_column
    use_column_value => true
    #指定追踪的字段，
    tracking_column => "last_updated"
    #追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
    tracking_column_type => "numeric"
    #记录最后一次运行的结果
    record_last_run => true
    #上面运行结果的保存位置
    last_run_metadata_path => "jdbc-position.txt"
    statement => "SELECT * FROM user where last_updated >:sql_last_value;"
    schedule => " * * * * * *"
  }
}
output {
  elasticsearch {
    document_id => "%{id}"
    document_type => "_doc"
    index => "users"
    hosts => ["http://localhost:9200"]
  }
  stdout{
    codec => rubydebug
  }
}

 使用 logstash 执行

logstash -f mysqltoes.conf

使用别名查询索引

POST /_aliases
{
  "actions": [
    {
      "add": {
        "index": "users",
        "alias": "view_users",
         "filter" : { "term" : { "is_deleted" : false } }
      }
    }
  ]
}

创建一个索引别名，过滤掉 只显示 is_deleted 为未删除的数据。

通过别名查询数据

POST view_users/_search
{
  "query": {
    "term": {
      "name.keyword": {
        "value": "Jack"
      }
    }
  }
}