How to deal with Data changing

[Aymen_Ben_Moussa]

I have kafka pushing logs from mongodb to logstash and logstash outputting to elasticsearch, i have fields added like the after and patch fields, they are both in the string format, the first one contains all the fields that are on my mongodb instance, everything in the document but none of them are accessible in es and if the document get's updated i get a other line pushed to es where after is nill and patch contains my $set command that i updated the document with in mongodb that's a other problem as it does not update the line that already exist or is there a way to link the patch line added to it's document ? here is a example of a line pushed to es in json format

{
  "_index": "yoyo4y",
  "_type": "doc",
  "_id": "vTFU_moBQHn1BMrfqi3m",
  "_version": 1,
  "_score": null,
  "_source": {
    "patch": null,
    "after": "{\"_id\" : {\"$oid\" : \"5ce81ce00770e00f1cbe9b49\"},\"name\" : \"mongo3\",\"season\" : \"cold2\",\"number\" : 999.0}",
    "op": "r",
    "@version": "1",
    "source": {
      "version": "0.9.5.Final",
      "ord": 1,
      "ns": "test.testCollection",
      "initsync": true,
      "sec": 1559044767,
      "h": 1595360082745638400,
      "rs": "rs0",
      "name": "mongo_conn",
      "connector": "mongodb"
    },
    "@timestamp": "2019-05-28T12:05:21.906Z",
    "ts_ms": 1559044773272
  },
  "fields": {
    "@timestamp": [
      "2019-05-28T12:05:21.906Z"
    ]
  },
  "sort": [
    1559045121906
  ]
}

and here is a line pushed to es after a db.collection.update()

  {
   "_index": "yoyo4y",
   "_type": "doc",
   "_id": "ujFb-WoBQHn1BMrfcy1W",
  "_version": 1,
  "_score": null,
  "_source": {
      "patch": "{\"$v\" : 1,\"$set\" : {\"number\" : 95.0}}",
      "op": "u",
      "@version": "1",
      "@timestamp": "2019-05-27T12:54:40.966Z",
      "after": null,
      "source": {
      "connector": "mongodb",
      "version": "0.9.5.Final",
      "name": "mongo_conn",
      "ns": "test.testCollection",
      "sec": 1558961680,
      "rs": "rs0",
      "h": -4743345325051307000,
      "initsync": false,
      "ord": 1
      },
      "ts_ms": 1558961680735
       },
      "fields": {
      "@timestamp": [
      "2019-05-27T12:54:40.966Z"
       ]
     },
     "sort": [
       1558961680966
      ]
    }

and this is my default mapping,

> {
>   "firsttry" : {
>     "mappings" : {
>       "doc" : {
>         "properties" : {
>           "@timestamp" : {
>             "type" : "date"
>           },
>           "@version" : {
>             "type" : "text",
>             "fields" : {
>               "keyword" : {
>                 "type" : "keyword",
>                 "ignore_above" : 256
>               }
>             }
>           },
>           "after" : {
>             "type" : "text",
>             "fields" : {
>               "keyword" : {
>                 "type" : "keyword",
>                 "ignore_above" : 256
>               }
>             }
>           },
>           "op" : {
>             "type" : "text",
>             "fields" : {
>               "keyword" : {
>                 "type" : "keyword",
>                 "ignore_above" : 256
>               }
>             }
>           },
>           "patch" : {
>             "type" : "text",
>             "fields" : {
>               "keyword" : {
>                 "type" : "keyword",
>                 "ignore_above" : 256
>               }
>             }
>           },
>           "source" : {
>             "properties" : {
>               "connector" : {
>                 "type" : "text",
>                 "fields" : {
>                   "keyword" : {
>                     "type" : "keyword",
>                     "ignore_above" : 256
>                   }
>                 }
>               },
>               "h" : {
>                 "type" : "long"
>               },
>               "initsync" : {
>                 "type" : "boolean"
>               },
>               "name" : {
>                 "type" : "text",
>                 "fields" : {
>                   "keyword" : {
>                     "type" : "keyword",
>                     "ignore_above" : 256
>                   }
>                 }
>               },
>               "ns" : {
>                 "type" : "text",
>                 "fields" : {
>                   "keyword" : {
>                     "type" : "keyword",
>                     "ignore_above" : 256
>                   }
>                 }
>               },
>               "ord" : {
>                 "type" : "long"
>               },
>               "rs" : {
>                 "type" : "text",
>                 "fields" : {
>                   "keyword" : {
>                     "type" : "keyword",
>                     "ignore_above" : 256
>                   }
>                 }
>               },
>               "sec" : {
>                 "type" : "long"
>               },
>               "version" : {
>                 "type" : "text",
>                 "fields" : {
>                   "keyword" : {
>                     "type" : "keyword",
>                     "ignore_above" : 256
>                   }
>                 }
>               }
>             }
>           },
>           "ts_ms" : {
>             "type" : "long"
>           }
>         }
>       }
>     }
>   }
> }

my main concern is to get the es mapping working recognizing those fields nested in {after} or {patch} and also a way to link that update with it's document , that mapping came with error type": "action_request_validation_exception",
"reason": "Validation Failed: 1: mapping type is missing;"
can not upload my new mapping as the Body is limited to only 7000 characters