Error with NSG sample

I am looking at the NSG config here
https://github.com/Azure/azure-diagnostics-tools/tree/master/Logstash/logstash-input-azureblob
I copied it and updated it with my keys but this is what i am using
input {
azureblob
{
storage_account_name => ""
storage_access_key => ""
container => "insights-logs-networksecuritygroupflowevent"
codec => "json"
# Refer https://docs.microsoft.com/en-us/azure/network-watcher/network-watcher-read-nsg-flow-logs
# Typical numbers could be 21/9 or 12/2 depends on the nsg log file types
file_head_bytes => 21
file_tail_bytes => 9
}
}

filter {
split { field => "[records]" }
split { field => "[records][properties][flows]"}
split { field => "[records][properties][flows][flows]"}
split { field => "[records][properties][flows][flows][flowTuples]"}

mutate{
split => { "[records][resourceId]" => "/"}
add_field => {"Subscription" => "%{[records][resourceId][2]}"
"ResourceGroup" => "%{[records][resourceId][4]}"
"NetworkSecurityGroup" => "%{[records][resourceId][8]}"}
convert => {"Subscription" => "string"}
convert => {"ResourceGroup" => "string"}
convert => {"NetworkSecurityGroup" => "string"}
split => { "[records][properties][flows][flows][flowTuples]" => ","}
add_field => {
"unixtimestamp" => "%{[records][properties][flows][flows][flowTuples][0]}"
"srcIp" => "%{[records][properties][flows][flows][flowTuples][1]}"
"destIp" => "%{[records][properties][flows][flows][flowTuples][2]}"
"srcPort" => "%{[records][properties][flows][flows][flowTuples][3]}"
"destPort" => "%{[records][properties][flows][flows][flowTuples][4]}"
"protocol" => "%{[records][properties][flows][flows][flowTuples][5]}"
"trafficflow" => "%{[records][properties][flows][flows][flowTuples][6]}"
"traffic" => "%{[records][properties][flows][flows][flowTuples][7]}"
}
convert => {"unixtimestamp" => "integer"}
convert => {"srcPort" => "integer"}
convert => {"destPort" => "integer"}
}

date{
match => ["unixtimestamp" , "UNIX"]
}
}

output {
stdout { codec => rubydebug }
}

When log stash tryes to parse the json i get an error
JJSON parse error, original data now in message field {:error=>##