anchor.diff: -= Streaming data to a downstream Elasticsearch
  += Streaming Data Changes from Your Database to Elasticsearch
   jpechane
   :awestruct-tags: [ mysql, postgres, elasticsearch, smt, example ]
   :awestruct-layout: blog-post
   
  -We have seen how to stream a change data changes to a downstream database link:/blog/2017/09/25/streaming-to-another-database[few weeks ago].
  -
  -In this blog post we will follow the same approach to stream the data to an https://www.elastic.co/[Elasticsearch] server for potential further search processing. But to make the matter a little bit more interesting we will stream the data to both PostgreSQL database and Elasticsearch so we will optimize access to data both via SQL query language and a full-text search.
  +We have seen how to set up a change data stream to a downstream database link:/blog/2017/09/25/streaming-to-another-database[few weeks ago].
  +
  +In this blog post we will follow the same approach to stream the data to an https://www.elastic.co/[Elasticsearch] server for potential further search processing. But to make the matter a little bit more interesting, we will stream the data to both, a PostgreSQL database and Elasticsearch, so we will optimize access to the data via the SQL query language as well as via full-text search.
   
   == Topology
  -The general topology for this scenario is displayed on the following picture:
  +The general topology for this scenario is shown in the following picture:
   
   .A General topology
   [#img-general]
  
positive.code: We again use the `UnwrapFromEnvelope` transformation to automatically extract a new row record and thus effectively convert the Debezium change data message into a simple one consumable by other connectors. Both JDBC and Elasticsearch connectors expect the same format.
  
  == Example
  Let's move directly to our example as that's where the changes are visible.
  
  First of all we need to deploy all components.
  [source,bash,indent=0]
  ----
  export DEBEZIUM_VERSION=0.6
  
negative.code: 
  The request configures these options:
  
  1. extracting only the new row's state from Debezium's change data message
  2. extracting the `id` field from the key `struct`, then the same key is used for the source and both destinations.
  This is to address the fact that the Elasticsearch connector only supports numeric types and `string`s as keys. If we do not extract the `id` the messages will be filtered out by the connector because of unknown key type.
  3. use key from the event instead of generating a synthetic one
  4. type under which the events will be registered in Elasticsearch
  
  Next we are going to register the JDBC Sink connector writing into PostgreSQL database:
  
  [source,bash,indent=0]
  ----
  curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @jdbc-sink.json
  