Logstash导入数据

电玩女神 2023-06-03 02:47 8阅读 0赞

开发中, 更多的是从现有数据库中导入数据  
Django中 python manage.py rebuild\_index 就是在导入数据

*  方式
    
     *  自己写一个程序, 按照之前的语法从数据库中读取数据并添加到es中
     *  也可以使用Logstash工具导入数据
 *  安装
 *  从mysql中导入数据
    
     *  创建配置文件

### Logstach安装 ###

sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

在 /etc/yum.repos.d/ 中创建logstash.repo文件

[logstash-6.x]
    name=Elastic repository for 6.x packages
    baseurl=https://artifacts.elastic.co/packages/6.x/yum
    gpgcheck=1
    gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
    enabled=1
    autorefresh=1
    type=rpm-md

执行

sudo yum install logstash

cd /usr/share/logstash/bin/
    sudo ./logstash-plugin install logstash-input-jdbc
    sudo ./logstash-plugin install logstash-output-elasticsearch
    scp mysql-connector-java-8.0.13.tar.gz python@10.211.55.7:~/
    tar -zxvf mysql-connector-java-8.0.13.tar.gz

创建配置文件logstash\_mysql.conf

input{
         jdbc {  # java数据库访问的API接口
             jdbc_driver_library => "/home/python/mysql-connector-java-8.0.13/mysql-connector-java-8.0.13.jar"
             jdbc_driver_class => "com.mysql.jdbc.Driver"
             jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/toutiao?tinyInt1isBit=false"
             jdbc_user => "root"
             jdbc_password => "mysql"
             jdbc_paging_enabled => "true"  # 数据分页, 一共14W数据
             jdbc_page_size => "1000"  # 每页1000条数据
             jdbc_default_timezone =>"Asia/Shanghai"
             statement => "select a.article_id as article_id,a.user_id as user_id, a.title as title, a.status as status, a.create_time as create_time,  b.content as content from news_article_basic as a inner join news_article_content as b on a.article_id=b.article_id"  # 联表查询, 尽量起别名,否则ES的字段名称会变为a.xx, 这样和mysql的字段名称会出现差异
         }
    }
    
    output{
          elasticsearch {
             hosts => "127.0.0.1:9200"
             index => "articles"
             document_id => "%{article_id}"  # 让文档id记录文章id, 方便进行数据库查询
             document_type => "article"
          }
          stdout {  # 导入过程中以json形式显式的输出导入的内容
             codec => json_lines  
         }
    }

*  增量更新的配置

input{
      record_last_run => "true"     # 记录最后一次运行时的数据点, 默认为最后一次更新的时间
        use_column_value => "true"  # 不再记录最后一次更新的时间, 而是记录最后一次更新时, 数据库某个字段的值(字段的值要求是递增的)
      tracking_column => "article_id"  # 设置记录的字段
    last_run_metadata_path => "/xx/data"    # 数据点的存储位置
      clean_run => "false"    # 从存储位置开始继续读取, 如果设置为true, 则清除数据点, 从头开始读取
    }

解压缩java类库

tar -zxvf mysqlxxx.tar.gz

执行导入命令

sudo /usr/share/logstash/bin/logstash -f ./logstash_mysql.conf

转载于:https://www.cnblogs.com/oklizz/p/11448261.html