使用logstash将整个数据库保存到elasticsearch答案

【问题标题】：save whole database to elasticsearch using logstash使用logstash将整个数据库保存到elasticsearch
【发布时间】：2018-12-13 14:28:14
【问题描述】：

我是 ELK 的新手，目前正在做的是：

使用configuration.conf文件在logstash中设置jdbc（输入>过滤>输出）
对于每个MySQL 查询在logstash 配置文件中具有单独的input{}
或使用pipilines.yml 分隔配置文件以在单独的线程中运行（即每个 MySQL 查询具有（存储在）不同的配置文件中）
运行命令logstash -f config.conf (Windows) 或只是logstash 用于管道

如何使用 logstash 获取数据库的所有表并将它们索引到 ElatsticSeach 一次性 其中每个表的索引与 MySQL 中的表名相同数据库 (Windows)。我可以将查询作为显示表运行，获取列表并使用 for 循环并为每个表定义 .conf 并将它们保存为 .conf 文件吗？但是我将如何修改 .yml 文件呢？因为文件是 .conf 和 .yml 而不是 .py？

Logstash configuration file image

【问题讨论】：

Logstash 无法运行“show tables”，需要自己做，并相应修改配置。您可以使用连接到数据库的 shell 脚本自动生成配置，运行 show tables 并为每个表打印出 jdbc 输入。
我认为logstash不可能做到这一点。您可能必须使用例如 Java API for Elasticsearch 编写自己的逻辑并遍历数据库中的所有表。

标签： python mysql elasticsearch logstash

【解决方案1】：

官方文档说“每个查询都必须有单独的 jdbc”：Configuring multiple SQL statements

这是脚本：

getTableNames.py

import MySQLdb
# custom made class, Generate
from package_name.generate_conf_yml_logstash import Generate
connection = MySQLdb.connect(host="localhost:3306",
                              user="root/sa", password="password", db="database_name")
cursor = connection.cursor()
cursor.execute("show tables")
tables = cursor.fetchall()
application_name_tables = []
for table in tables:
    application_name_tables.append(table[0])
cursor.close()
connection.close()
Generate.save_files(application_name_tables)

generate_conf_yml_logstash.py

import sys
import os.path


class Generate:
    @staticmethod
    def save_files(tables):

        # save config files
        save_path = "D:\folder_name"
        for table in tables:
            init_f = save_path + "\initial_logstash.conf"
            conf_f_name = table + ".conf"
            save_file = os.path.join(save_path, conf_f_name)
            with open(init_f, "r") as original:  # read only
                data = original.read()
                data = data.replace("table_name", table)
            with open(save_file, "w+") as conf:  # overwrite if exist
                conf.write(data)

        # save yml file
        yml_f_name = "init_logstash_pipelines.yml"
        save_file = os.path.join(save_path, yml_f_name)
        with open(save_file, "r") as original:
            data = original.read()
        with open(save_file, "a+") as yml:  # append
            for table in tables:
                data = data.replace("table", table)
                yml.write(data)
        sys.exit()

Sample Config 和 YML 文件是：

init_logstash_pipelines.yml

- pipeline.id: table
    path.config: "../config/table.conf"
    pipeline.workers: 1

initial_logstash.conf

input {
  jdbc {
    jdbc_driver_library => "../logstash-6.3.0/logstash-core/lib/jars/mysql-connector-java-5.1.46-bin.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://Mysql:3306/database_name"
    jdbc_user => "root"
    jdbc_password => "password"
    statement => "SELECT * from table_name"
    # schedule => "* * * * *"
  }
}
output {
    stdout { codec => json_lines }
    elasticsearch {
    hosts => ["localhost:9200"]
    index => "table_name"
    # as every table has diff. primary key, change this please
    document_id => "%{pk}"
    }
}

请随意更改初始配置、yml 文件和代码。满足您的需求

【讨论】：