Pulling data from two different DB with Logstash JDBC Input and merging input events data

Hi and thanks for your great community.

I have two database that I want configuring a pipeline with Logstash that pulling data from these two database and merging the input events based on one same field in these events and pushing one merged event into Elasticsearch.

I think i must using the JDBC input plugin twice in Logstash input block like this config:

input {
  jdbc {
    jdbc_driver_library => "/Users/logstash/mysql-connector-java-5.1.39-bin.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "connection str for first db"
    jdbc_user => "root"
    jdbc_password => "password"
    schedule => "* * * * *"
    statement => "select * from table1"
    type => "table1"
  }
  jdbc {
    jdbc_driver_library => "/Users/logstash/mysql-connector-java-5.1.39-bin.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "connection str for second db"
    jdbc_user => "root"
    jdbc_password => "password"
    schedule => "* * * * *"
    statement => "select * from table2"
    type => "table2"
  }
  # add more jdbc inputs to suit your needs 
}

filter {
	.
	.
	.
}

output {
    elasticsearch {
        index => "testdb"
        document_type => "idx"
        hosts => "elstic host"
    }
}

but I haven't any idea about how can I merged the input events based on one same field like bellow:

#event 1
{
	"user_id": 90876,
	"user_register_ts": 1631449749
}
#event 2
{
	"user_id": 89090,
	"user_register_ts": 1631444700
}
#event 3
{
	"user_id": 90876,
	"store": "GooglePlay",
	"coin_num": "780"
}
#event 4
{
	"user_id": 89090,
	"store": "AppStore",
	"coin_num": "1037"
}

## after merging process

# merged event 1 (merged event 1 and 3 based on user_id)
{
	"user_id": 90876,
	"user_register_ts": 1631449749,
	"store": "GooglePlay",
	"coin_num": "780"
}

# merged event 2 (merged event 2 and 4 based on user_id)
{
	"user_id": 89090,
	"user_register_ts": 1631444700,
	"store": "AppStore",
	"coin_num": "1037"
}

Can you help me about methods that helps me to merging these events in logstash?

Thanks for any response.

Instead of using two jdbc inputs you might want to use one jdbc input and a jdbc_static filter to enrich it.

1 Like