Import JSON from SQL Server to ElasticSearch

[qttv]

Hi, I'm importing some data from Microsoft SQL Server in ElasticSearch using Logstash. I'm dealing with a table in wich a column contains JSON logs, and i need to import these logs splitting the fields in order to create new items in my index.

I'm trying to use a JSON filter for doing these, but it seems not working. In past i used with success the JSON filter for import CSV files containing JSON columns, but in this case i'm not able to obtain the result that i want.

I show you my config file:

input {
  jdbc {
    jdbc_connection_string =>"jdbc:sqlserver://...\SQLExpress01:1433;databaseName=...;integratedSecurity=false"
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_user => "..."
	jdbc_password => "..."
    jdbc_driver_library => "C:\Users\Valerio\Desktop\sqljdbc_7.4\ita\mssql-jdbc-7.4.1.jre8.jar"
    schedule => "* * * * *" 	
    statement => ["SELECT * from dbo.Logs where Id > :sql_last_value  "]
	    use_column_value => true
        tracking_column => "id"
		tracking_column_type => "numeric" 
        clean_run => true 
        record_last_run => true
        last_run_metadata_path => "C:\Users\Valerio\Desktop\Logstash\logstash-7.12.1\bin\Storico\.logstash_jdbc_last_run"
  }
}



filter {
  json {
        source => "RawMessage"
  }
  } 
  
  
output {
  elasticsearch {
    hosts => "http://localhost:9200"
    index => "cs_users"
  }
}

This code give me no errors but it simply don't split the content of the JSON field as i wish.
The field containing the JSON logs is named "RawMessage" and the Logs presents this aspect:

{
  "message": "...",
  "level": "...",
  "logType": "...",
  "timeStamp": "...",
  "fingerprint": "...",
  "windowsIdentity": "...",
  "machineName": "...",
  "processName": "...",
  "processVersion": "...",
  "jobId": "...",
  "robotName": "...",
  "machineId": 75,
  "fileName": "...",
  "logF_BusinessProcessName": "...",
  "Data": {
    "processCode": "",
    "processName": "",
    "userCode": "...",
    "sessionId": "2",
    "computedParams": {
      "IdMask": "7",
      "DescMask": "Gestione tabella",
      "IdItem": "9",
      "CountProtectedIdMask": "0",
      "CountSendInputCommandIdMask": "0",
      "dataType": "OutputTable",
      "inputKey": "",
      "RowEmulatorMessage": "",
      "InputCoord": "",
      "InputValue": "",
      "CheckSingleInputFromCoord": "",
      "SendInputCommandValue": "",
      "CheckAfterSendInputFromCoord": "",
      "CheckAfterSendInputToCoord": "",
      "CheckAfterSendInputValue": "",
      "CheckAfterSendInputOperator": "",
      "SendInputCommandMaxRetry": ""
    },
    "result": {
      "OutputLabelString": "...",
      "OutputLabelFromCoord": "9,2",
      "OutputLabelToCoord": "9,8",
      "OutputValueTopLeft": "3,0",
      "OutputValueTopRight": "0,11",
      "OutputValueBottomLeft": "",
      "OutputValueBottomRight": "",
      "OutputFormatValue": "string",
      "OutputLabelOptional": "",
      "Eccetera": "..."
    }
  }
}

As you can see there is a nested JSON, maybe that's the problem?
I was thinking to add a multiline codec, but I don't know where to insert it, because it reguards only the RawMessage field and not the others.
Hope someone will help me.

Thanks!