Thanks Constatin.
Your answer will work if we just use one index & type.
But, my current river using multiple index & type. ( Index is used for customer identity and type is used for element name )
So we have to find a solution how specify index & type at input data instead of setting at logstash config.
I also curious whether we could update data using logstash.
I include _Id field at input message, but auto generated _Id is set on the document.
So update document isn't happen in my test.