Jdbc input put data in all indexes

Elastic Stack Logstash
1

Hi all,

When I create a conf file with jdbc input, the data go to all my indexes.
For example I did two conf files with jdbc input :

input {
jdbc {
jdbc_connection_string => "jdbc:postgresql://localhost:5432/postgres"
jdbc_user => "user"
jdbc_password => "password"
jdbc_driver_library => "/path_to_postgre_jar/postgresql-9.2-1004.jdbc3.jar"
jdbc_driver_class => "org.postgresql.Driver"
schedule => "30 * * * * *"
statement => "select id_temp_index, nom, prenom, sexe, age from temp_index"
}
}

filter { }

output {
elasticsearch {
index => "index_sql_1"
document_type => "type_01"
hosts => ["localhost:9200"]
document_id => "%{id_temp_index}"
}
}

and

input {
jdbc {
jdbc_connection_string => "jdbc:postgresql://localhost:5432/postgres"
jdbc_user => "user"
jdbc_password => "password"
jdbc_driver_library => "/path_to_postgre_jar/postgresql-9.2-1004.jdbc3.jar"
jdbc_driver_class => "org.postgresql.Driver"
schedule => "30 * * * * *"
statement => "select id_temp_index as id_temp_index_2, nom as nom_2, prenom as prenom_2, sexe as sexe_2, age as age_2 from temp_index"
}
}

filter{ }

output {
elasticsearch {
protocol => http
index => "index_sql_2"
document_type => "type_2"
hosts => ["localhost:9200"]
document_id => "%{id_temp_index_2}"
}
}

So I should have two indexes quite identical exept the name of the columns with 5 fields and I got :

1796×748 133 KB

So, each index got the fields of each conf and one line come from the other.
Those lines appears in all indexes I will create, even csv input, beats input...

And it can replace some data if the document_id is somehting like document_id => "%{id}" in several conf files.

So, did I miss something in my conf files? Is there a way to prevent that?

2

@Amilyso Hi!

If I have not misunderstanding, are you trying to copy index_sql_1 document to index_sql_2?

3

Not at all, it's just I got a lot of error in one of my project due to this "bug". So i simplify the problem to try to undestand it.

I want to get somehting like that :

sql_index2.PNG1281×556 63.7 KB

I should have use two tables for my example (like custormer and products instead of using the same table twice), it should be more relevent...

4

@Amilyso
your situation is similar to mine. I have three table in my db. And I have to build three index to save it. But I got a solution of it. but it is not the best way.

5

may be you can try my solution

input{
	jdbc{
		# mysql jdbc connection string to our backup databse
		jdbc_connection_string => "jdbc:mysql://localhost:3306/price_prod_test"

		# the user we wish to excute our statement as
		jdbc_user => "user"
		jdbc_password => "pass"

		# the path to our downloaded jdbc driver
		jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"

		# the name of the driver class for mysql
		jdbc_driver_class => "com.mysql.jdbc.Driver"

		# the control statement
		record_last_run=>true
		type => "product"
		statement => "SELECT * FROM product"
	}
	jdbc{
		# mysql jdbc connection string to our backup databse
		jdbc_connection_string => "jdbc:mysql://localhost:3306/price_prod_test"

		# the user we wish to excute our statement as
		jdbc_user => "user"
		jdbc_password => "pass"

		# the path to our downloaded jdbc driver
		jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"

		# the name of the driver class for mysql
		jdbc_driver_class => "com.mysql.jdbc.Driver"

		# the control statement
		record_last_run=>true
		type => "merchant"
		statement => "SELECT * FROM merchant"
	}
	jdbc{
		# mysql jdbc connection string to our backup databse
		jdbc_connection_string => "jdbc:mysql://localhost:3306/price_prod_test"

		# the user we wish to excute our statement as
		jdbc_user => "user"
		jdbc_password => "pass"

		# the path to our downloaded jdbc driver
		jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"

		# the name of the driver class for mysql
		jdbc_driver_class => "com.mysql.jdbc.Driver"

		# the control statement
		record_last_run=>true
		type => "news"
		statement => "SELECT * FROM news"
	}
}

filter {
	if [type] == "product"{
		mutate {
			add_field => { "[product_suggest][input]" => "%{display_name}"}
			add_field => { "[product_suggest][weight]" => "%{id}"}
			add_field => { "id_sort" => "%{id}" }
			remove_field => ["attribute_value_en", "attribute_value_sc"]
		}
	}
	if [type] == "merchant"{
		mutate {
			add_field => { "[merchant_suggest][input]" => "%{merchant_name}"}
			add_field => { "[merchant_suggest][weight]" => "%{id}"}
		}
	}
	if [type] == "news"{
		mutate {
			add_field => { "id_sort" => "%{id}" }
		}
	}
}

output {
	stdout{codec => dots}
	if [type] == "product"{
		elasticsearch {
				hosts => "localhost"
				index => "product"
				document_id => "%{id}"
		}
	}
	if [type] == "merchant"{
		elasticsearch {
				hosts => "localhost"
				index => "merchant"
				document_id => "%{id}"
		}
	}
	if [type] == "news"{
		elasticsearch {
				hosts => "localhost"
				index => "news"
				document_id => "%{id}"
		}
	}
}
6

Thanks!
I'll try it and will tell you if it works for me.

7

It works fine!

Thanks @Greentea