azureblobtologzio.conf

input {
azureblob
  {
      storage_account_name => "yourStorageAccount"
      storage_access_key => "yourStorageAccessKey"
      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 => 12
      file_tail_bytes => 2
      # Enable / tweak these settings when event is too big for codec to handle.
      # break_json_down_policy => "with_head_tail"
      # break_json_batch_count => 2
  }
}

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"}        
}
mutate {
    add_field => { "token" => "yourLogzioToken" }
}

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

output {
  tcp {
    host => "listener.logz.io"
    port => 5050
    codec => json_lines
  }
}