logstash从oracle12c导入数据到elasticsearch

Posted 2023-05-29 奋斗的小鸟

在logstash的config中添加配置信息

cd logstash-5.2.0/config/

vim 12ctoEs.conf

input 
jdbc 
jdbc_driver_library => "/hadoop/logstash-5.2.0/ojdbc8.jar"
jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
jdbc_connection_string => "jdbc:oracle:thin:@192.168.100.20:1521/orclpdb"
jdbc_user => "test"
jdbc_password => "test"
jdbc_paging_enabled => "true"
jdbc_page_size => "50000"
statement => "select id,name,age from sys.test1"


filter 
mutate 
remove_field => ["@version","@timestamp"]


output 
elasticsearch 
hosts => "http://192.168.100.10:9200"
index => "test1"
document_id => "%id"
document_type => "test1"



##执行命令

 cd ../bin/
 ./logstash -f ../config/12ctoEs.conf

  注：jdbc_driver_library jar包可在oracle安装目录下查询到

 

Logstash：把MySQL数据导入到Elasticsearch中

Logstash：把MySQL数据导入到Elasticsearch中

前提条件

需要安装好Elasticsearch及Kibana。

MySQL安装

根据不同的操作系统我们分别对MySQL进行安装。我们可以访问网页来对MySQL进行安装。等我们安装完我们的MySQL后，在我们的terminal中，打入如下的命令来检查MySQL的版本：

$ /usr/local/mysql/bin/mysql -V
/usr/local/mysql/bin/mysql  Ver 8.0.17 for macos10.14 on x86_64 (MySQL Community Server - GPL)

Logstash安装

在上一步中，已经知道了mysql的版本信息。需要下载相应的JDBC connector。在地址https://dev.mysql.com/downloads/connector/j/
下载最新的Connector。下载完这个Connector后，把这个connector存入到Logstash安装目录下的如下子目录中。

$ ls logstash-core/lib/jars/mysql-connector-java-8.0.17.jar 
logstash-core/lib/jars/mysql-connector-java-8.0.17.jar

这样我们的安装就完成了。

准备练习数据

采用把一个CSV文件导入到MySQL中的办法来形成一个MySQL的数据库。CSV文件下载地址：

https://github.com/liu-xiao-guo/sample_csv

在上面的sample_csv中，有一个SalesJan2009.csv文件。通过MySQL的前端工具把这个导入到MySQL数据库中。

这样MySQL的数据库data里含有一个叫做SalesJan2009的数据就建立好了。

Logstash 配置

对Logstash做如下的配置sales.conf:

    input {
        jdbc {
           jdbc_connection_string => "jdbc:mysql://localhost:3306/data"
           jdbc_user => "root"
           jdbc_password => "YourMyQLPassword"
           jdbc_validate_connection => true
           jdbc_driver_library => ""
           jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
           parameters => { "Product_id" => "Product1" }
           statement => "SELECT * FROM SalesJan2009 WHERE Product = :Product_id"
        }    
    }
     
    filter {
        mutate {
            rename => {
                "longitude" => "[location][lon]"
                "latitude" => "[location][lat]"
            }
        }
    }
     
    output {
        stdout {
        }
     
        elasticsearch {
            index => "sales"
            hosts => "localhost:9200"
            document_type => "_doc"
        } 
    }

在这里，必须替换jdbc_user和jdbc_password为自己的MySQL账号的用户名及密码。特别值得指出的是jdbc_driver_library按elastic的文档是可以放入JDBC驱动的路径及驱动名称。实践证明如果这个驱动不在JAVA的classpath里，也是不能被正确地加载。正因为这样的原因，在上一步里把驱动mysql-connector-java-8.0.17.jar放入到Logstash的jar目录里，所以这里就直接填入空字符串。

运行Logstash加载数据

接下来我们运行Logstash来加载我们的MySQL里的数据到Elasticsearch中：

./bin/logstash --debug -f ~/data/sales.conf

在这里把sales.conf置于用户home目录下的data子目录中。

我们可以在Kibana中查看到最新的导入到Elasticsearch中的数据

这里显示在sales索引中有847个文档。一旦数据进入到我们的Elastic，我们可以对数据进行分析：

参考：https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html#plugins-inputs-jdbc-parameters