logstash 同步 mysql 多张表到 es

【摘要】上一篇文章讲了 logstash 自动同步单张数据表到 es 中，本文将同步至多张表，多张表的基本步骤是和单张表是一样的，区别在于配置文件的配置不同。

接下来我们给出同步两张表的小例子，数据准备还是参考上一篇文章中的数据：

/438.html

logstash.conf 配置文件：

1. input {

2.     stdin {

4.     }

6.     # jdbc-1.sql 的 jdbc 配置

7.     jdbc {

8.         jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/jy?useUnicode=true&characterEncoding=utf-8&useSSL=false"

9.         jdbc_user => "root"

10.         jdbc_password => "root"

11.         jdbc_driver_library => "bin/mysql-connector-java-5.1.7-bin.jar"

12.         jdbc_driver_class => "com.mysql.jdbc.Driver"

13.         jdbc_paging_enabled => "true"

14.         jdbc_page_size => "50000"

15.         jdbc_default_timezone =>"Asia/Shanghai"

17.         statement => "select * from student"

18.         # 也可以在对应的文件中写 sql 

19.         # statement_filepath => "jdbc.sql"

21.         schedule => "* * * * *"

22.         lowercase_column_names => false

24.         # type 用来区分不同的同步任务

25.         type => "jdbc_student"

26.     }

28.     # jdbc-2.sql 的 jdbc 配置

29.     jdbc {

30.         jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/jy?useUnicode=true&characterEncoding=utf-8&useSSL=false"

31.         jdbc_user => "root"

32.         jdbc_password => "root"

33.         jdbc_driver_library => "bin/mysql-connector-java-5.1.7-bin.jar"

34.         jdbc_driver_class => "com.mysql.jdbc.Driver"

35.         jdbc_paging_enabled => "true"

36.         jdbc_page_size => "50000"

37.         jdbc_default_timezone =>"Asia/Shanghai"

38.         statement => "select * from class"

39.         schedule => "* * * * *"

40.         lowercase_column_names => false

41.         type => "jdbc_class"

42.     }

43. }

45. output {

47.     # 上面 type = "jdbc_student" 对应的同步任务 

48.     if[type] == "jdbc_student"{

49.         elasticsearch {

50.             hosts => ["localhost:9200"]

51.             # 索引名称

52.             index => "index_student"

53.             document_id => "%{id}"

54.             document_type => "test"

55.         }

56.     }

58.     if[type] == "jdbc_class"{

59.         elasticsearch {

60.             hosts => ["localhost:9200"]

61.             index => "index_class"

62.             document_id => "%{id}"

63.             document_type => "test"

64.         }

65.     }

67.     stdout {

68.         codec => json_lines

69.     }

70. }

如图所示，已经同步成功啦！

我要打赏