diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 3648e287779f..6b4dac641259 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -159,6 +159,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - Added separation of transform context object inside httpjson. Introduced new clause `.parent_last_response.*` {pull}33499[33499]
 - Cloud Foundry input uses server-side filtering when retrieving logs. {pull}33456[33456]
 - Add `parse_aws_vpc_flow_log` processor. {pull}33656[33656]
+- Update `aws.vpcflow` dataset in AWS module have a configurable log `format` and to produce ECS 8.x fields. {pull}33699[33699]
 - Modified `aws-s3` input to reduce mutex contention when multiple SQS message are being processed concurrently. {pull}33658[33658]
 - Disable "event normalization" processing for the aws-s3 input to reduce allocations. {pull}33673[33673]
 - Add Common Expression Language input. {pull}31233[31233]
diff --git a/x-pack/filebeat/filebeat.reference.yml b/x-pack/filebeat/filebeat.reference.yml
index 0e34ef771b88..cb543595c5d2 100644
--- a/x-pack/filebeat/filebeat.reference.yml
+++ b/x-pack/filebeat/filebeat.reference.yml
@@ -474,6 +474,9 @@ filebeat.modules:
     # Configures the SSL settings, ie. set trusted CAs, ignore certificate verification....
     #var.ssl:
 
+    # Specify a custom VPC flow log format.
+    #var.format:
+
 #----------------------------- AWS Fargate Module -----------------------------
 - module: awsfargate
   log:
diff --git a/x-pack/filebeat/module/aws/_meta/config.yml b/x-pack/filebeat/module/aws/_meta/config.yml
index 60213ba811a1..e92cb36e7b53 100644
--- a/x-pack/filebeat/module/aws/_meta/config.yml
+++ b/x-pack/filebeat/module/aws/_meta/config.yml
@@ -376,3 +376,6 @@
 
     # Configures the SSL settings, ie. set trusted CAs, ignore certificate verification....
     #var.ssl:
+
+    # Specify a custom VPC flow log format.
+    #var.format:
diff --git a/x-pack/filebeat/module/aws/vpcflow/config/input.yml b/x-pack/filebeat/module/aws/vpcflow/config/input.yml
index 5c08c0e6e386..ecb1842be7a8 100644
--- a/x-pack/filebeat/module/aws/vpcflow/config/input.yml
+++ b/x-pack/filebeat/module/aws/vpcflow/config/input.yml
@@ -79,12 +79,14 @@ ssl: {{ .ssl | tojson }}
 {{ else if eq .input "file" }}
 
 type: log
-paths:
-  {{ range $i, $path := .paths }}
-  - {{$path}}
-    {{ end }}
+paths: {{ .paths | tojson }}
 exclude_files: [".gz$"]
 
 {{ end }}
 tags: {{.tags | tojson}}
 publisher_pipeline.disable_host: {{ inList .tags "forwarded" }}
+
+processors:
+  - parse_aws_vpc_flow_log:
+      format: {{ .format | tojson }}
+  - community_id: ~
diff --git a/x-pack/filebeat/module/aws/vpcflow/ingest/pipeline.yml b/x-pack/filebeat/module/aws/vpcflow/ingest/pipeline.yml
index d4d98f083966..7837ba07dc64 100644
--- a/x-pack/filebeat/module/aws/vpcflow/ingest/pipeline.yml
+++ b/x-pack/filebeat/module/aws/vpcflow/ingest/pipeline.yml
@@ -2,171 +2,43 @@
 description: Pipeline for AWS VPC Flow Logs
 
 processors:
+  - drop:
+      if: 'ctx.message.startsWith("version") || ctx.message.startsWith("instance-id")'
   - set:
       field: event.ingested
       value: '{{_ingest.timestamp}}'
   - set:
       field: ecs.version
-      value: '1.12.0'
+      value: '8.0.0'
   - rename:
       field: message
       target_field: event.original
       ignore_missing: true
   - set:
-      field: event.type
-      value: flow
+      field: event.kind
+      value: event
   - set:
       field: event.category
-      value: network_traffic
-  - drop:
-      if: 'ctx.event?.original.startsWith("version") || ctx.event?.original.startsWith("instance-id")'
-  - script:
-      lang: painless
-      if: ctx.event?.original != null
-      source: >-
-        ctx._temp_ = new HashMap();
-        ctx._temp_.message_token_count = ctx.event?.original.splitOnToken(" ").length;
-  - dissect:
-      field: event.original
-      pattern: '%{aws.vpcflow.version} %{aws.vpcflow.account_id} %{aws.vpcflow.interface_id} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.srcport} %{aws.vpcflow.dstport} %{aws.vpcflow.protocol} %{aws.vpcflow.packets} %{aws.vpcflow.bytes} %{aws.vpcflow.start} %{aws.vpcflow.end} %{aws.vpcflow.action} %{aws.vpcflow.log_status}'
-      if: ctx?._temp_?.message_token_count == 14
-  - dissect:
-      field: event.original
-      pattern: '%{aws.vpcflow.instance_id} %{aws.vpcflow.interface_id} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.pkt_srcaddr} %{aws.vpcflow.pkt_dstaddr}'
-      if: ctx?._temp_?.message_token_count == 6
-  - dissect:
-      field: event.original
-      pattern: '%{aws.vpcflow.version} %{aws.vpcflow.interface_id} %{aws.vpcflow.account_id} %{aws.vpcflow.vpc_id} %{aws.vpcflow.subnet_id} %{aws.vpcflow.instance_id} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.srcport} %{aws.vpcflow.dstport} %{aws.vpcflow.protocol} %{aws.vpcflow.tcp_flags} %{aws.vpcflow.type} %{aws.vpcflow.pkt_srcaddr} %{aws.vpcflow.pkt_dstaddr} %{aws.vpcflow.action} %{aws.vpcflow.log_status}'
-      if: ctx?._temp_?.message_token_count == 17
-  - dissect:
-      field: event.original
-      pattern: '%{aws.vpcflow.version} %{aws.vpcflow.vpc_id} %{aws.vpcflow.subnet_id} %{aws.vpcflow.instance_id} %{aws.vpcflow.interface_id} %{aws.vpcflow.account_id} %{aws.vpcflow.type} %{aws.vpcflow.srcaddr} %{aws.vpcflow.dstaddr} %{aws.vpcflow.srcport} %{aws.vpcflow.dstport} %{aws.vpcflow.pkt_srcaddr} %{aws.vpcflow.pkt_dstaddr} %{aws.vpcflow.protocol} %{aws.vpcflow.bytes} %{aws.vpcflow.packets} %{aws.vpcflow.start} %{aws.vpcflow.end} %{aws.vpcflow.action} %{aws.vpcflow.tcp_flags} %{aws.vpcflow.log_status}'
-      if: ctx?._temp_?.message_token_count == 21
-
-  # Convert Unix epoch to timestamp
-  - date:
-      field: aws.vpcflow.end
-      target_field: '@timestamp'
-      ignore_failure: true
-      formats:
-        - UNIX
-  - date:
-      field: aws.vpcflow.start
-      target_field: event.start
-      ignore_failure: true
-      formats:
-        - UNIX
-  - date:
-      field: aws.vpcflow.end
-      target_field: event.end
-      ignore_failure: true
-      formats:
-        - UNIX
-  - remove:
-      field:
-        - aws.vpcflow.start
-        - aws.vpcflow.end
-      ignore_missing: true
-  - script:
-      lang: painless
-      ignore_failure: true
-      if: ctx.aws != null
-      source: >-
-        void handleMap(Map map) {
-          for (def x : map.values()) {
-            if (x instanceof Map) {
-                handleMap(x);
-            } else if (x instanceof List) {
-                handleList(x);
-            }
-          }
-          map.values().removeIf(v -> v instanceof String && v == "-");
-        }
-        void handleList(List list) {
-          for (def x : list) {
-              if (x instanceof Map) {
-                  handleMap(x);
-              } else if (x instanceof List) {
-                  handleList(x);
-              }
-          }
-        }
-        handleMap(ctx.aws);
+      value: [network]
   - set:
-      field: event.outcome
-      value: allow
-      if: ctx.aws?.vpcflow?.action == "ACCEPT"
-  - set:
-      field: event.outcome
-      value: deny
-      if: ctx.aws?.vpcflow?.action == "REJECT"
-  - rename:
-      field: aws.vpcflow.srcaddr
-      target_field: source.address
-      ignore_missing: true
+      field: cloud.provider
+      value: aws
+
+  # parse_aws_vpc_flow_log removes the original values to avoid duplication
+  # but to avoid a breaking change continue to populate these fields.
   - set:
-      field: source.ip
-      copy_from: source.address
-      if: ctx.source?.address != null
-  - convert:
-      field: aws.vpcflow.srcport
-      target_field: source.port
-      type: integer
-      ignore_missing: true
-  - rename:
-      field: aws.vpcflow.dstaddr
-      target_field: destination.address
-      ignore_missing: true
+      copy_from: cloud.account.id
+      field: aws.vpcflow.account_id
+      ignore_empty_value: true
   - set:
-      field: destination.ip
-      copy_from: destination.address
-      if: ctx.destination?.address != null
-  - convert:
-      field: aws.vpcflow.dstport
-      target_field: destination.port
-      type: integer
+      copy_from: cloud.instance.id
+      field: aws.vpcflow.instance_id
+      ignore_empty_value: true
+  - uppercase:
+      field: event.action
+      target_field: aws.vpcflow.action
       ignore_missing: true
-  - rename:
-      field: aws.vpcflow.protocol
-      target_field: network.iana_number
-      ignore_missing: true
-  - convert:
-      field: aws.vpcflow.packets
-      target_field: source.packets
-      type: long
-      ignore_missing: true
-  - convert:
-      field: aws.vpcflow.bytes
-      target_field: source.bytes
-      type: long
-      ignore_missing: true
-  - set:
-      field: network.bytes
-      copy_from: source.bytes
-      if: ctx.source?.bytes != null
-  - set:
-      field: network.packets
-      copy_from: source.packets
-      if: ctx.source?.packets != null
-  - set:
-      field: network.type
-      value: ipv4
-      if: 'ctx.source?.ip != null && ctx.source?.ip.contains(".")'
-  - set:
-      field: network.type
-      value: ipv6
-      if: 'ctx.source?.ip != null && ctx.source?.ip.contains(":")'
-  - set:
-      field: network.transport
-      value: tcp
-      if: ctx.network?.iana_number == "6"
-  - set:
-      field: network.transport
-      value: udp
-      if: ctx.network?.iana_number == "17"
-  - community_id:
-      target_field: network.community_id
-      ignore_failure: true
+
   # IP Geolocation Lookup
   - geoip:
       field: source.ip
@@ -176,6 +48,7 @@ processors:
       field: destination.ip
       target_field: destination.geo
       ignore_missing: true
+
   # IP Autonomous System (AS) Lookup
   - geoip:
       database_file: GeoLite2-ASN.mmdb
@@ -209,66 +82,7 @@ processors:
       field: destination.as.organization_name
       target_field: destination.as.organization.name
       ignore_missing: true
-  # Generate related.ip field
-  - append:
-      if: 'ctx.source?.ip != null && ctx.destination?.ip != null'
-      field: related.ip
-      value: ["{{source.ip}}", "{{destination.ip}}"]
-  - set:
-      field: cloud.provider
-      value: aws
-  - set:
-      if: ctx.aws?.vpcflow?.account_id != null
-      field: cloud.account.id
-      value: '{{aws.vpcflow.account_id}}'
-  - set:
-      if: 'ctx?.aws?.vpcflow?.instance_id != null && ctx.aws.vpcflow.instance_id != "-"'
-      field: cloud.instance.id
-      value: '{{aws.vpcflow.instance_id}}'
-  - set:
-      field: event.kind
-      value: event
-  - script:
-      lang: painless
-      ignore_failure: true
-      if: "ctx.aws?.vpcflow?.tcp_flags != null"
-      source: |
-        if (ctx.aws.vpcflow.tcp_flags_array == null) {
-          ArrayList al = new ArrayList();
-          ctx.aws.vpcflow.put("tcp_flags_array", al);
-        }
 
-        def flags = Integer.parseUnsignedInt(ctx.aws.vpcflow.tcp_flags);
-
-        if ((flags & 0x01) != 0) {
-          ctx.aws.vpcflow.tcp_flags_array.add('fin');
-        }
-        if ((flags & 0x02) != 0) {
-          ctx.aws.vpcflow.tcp_flags_array.add('syn');
-        }
-        if ((flags & 0x04) != 0) {
-          ctx.aws.vpcflow.tcp_flags_array.add('rst');
-        }
-        if ((flags & 0x08) != 0) {
-          ctx.aws.vpcflow.tcp_flags_array.add('psh');
-        }
-        if ((flags & 0x10) != 0) {
-          ctx.aws.vpcflow.tcp_flags_array.add('ack');
-        }
-        if ((flags & 0x20) != 0) {
-          ctx.aws.vpcflow.tcp_flags_array.add('urg');
-        }
-  - remove:
-      field:
-        - _temp_
-        - aws.vpcflow.srcaddr
-        - aws.vpcflow.srcport
-        - aws.vpcflow.dstaddr
-        - aws.vpcflow.dstport
-        - aws.vpcflow.bytes
-        - aws.vpcflow.packets
-        - aws.vpcflow.protocol
-      ignore_missing: true
   - remove:
       field: event.original
       if: "ctx?.tags == null || !(ctx.tags.contains('preserve_original_event'))"
@@ -277,4 +91,4 @@ processors:
 on_failure:
   - set:
       field: 'error.message'
-      value: '{{ _ingest.on_failure_message }}'
+      value: '{{{ _ingest.on_failure_message }}}'
diff --git a/x-pack/filebeat/module/aws/vpcflow/manifest.yml b/x-pack/filebeat/module/aws/vpcflow/manifest.yml
index b329c7264f19..de772408a868 100644
--- a/x-pack/filebeat/module/aws/vpcflow/manifest.yml
+++ b/x-pack/filebeat/module/aws/vpcflow/manifest.yml
@@ -24,6 +24,12 @@ var:
   - name: proxy_url
   - name: max_number_of_messages
   - name: ssl
+  - name: format
+    default:
+      - version account-id interface-id srcaddr dstaddr srcport dstport protocol packets bytes start end action log-status
+      - instance-id interface-id srcaddr dstaddr pkt-srcaddr pkt-dstaddr
+      - version interface-id account-id vpc-id subnet-id instance-id srcaddr dstaddr srcport dstport protocol tcp-flags type pkt-srcaddr pkt-dstaddr action log-status
+      - version vpc-id subnet-id instance-id interface-id account-id type srcaddr dstaddr srcport dstport pkt-srcaddr pkt-dstaddr protocol bytes packets start end action tcp-flags log-status
 
 ingest_pipeline: ingest/pipeline.yml
 input: config/input.yml
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/accept-reject-traffic.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/accept-reject-traffic.log-expected.json
index 3f2adbd186d6..5a2e3b77b78b 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/accept-reject-traffic.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/accept-reject-traffic.log-expected.json
@@ -5,7 +5,7 @@
         "aws.vpcflow.action": "ACCEPT",
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "OK",
-        "aws.vpcflow.version": "2",
+        "aws.vpcflow.version": 2,
         "cloud.account.id": "123456789010",
         "cloud.provider": "aws",
         "destination.address": "175.16.199.1",
@@ -19,15 +19,21 @@
         "destination.geo.region_name": "Jilin Sheng",
         "destination.ip": "175.16.199.1",
         "destination.port": 22,
-        "event.category": "network_traffic",
+        "event.action": "accept",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2014-12-14T04:07:50.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "2 123456789010 eni-1235b8ca123456789 175.16.199.1 175.16.199.1 20641 22 6 20 4249 1418530010 1418530070 ACCEPT OK",
-        "event.outcome": "allow",
+        "event.outcome": "success",
         "event.start": "2014-12-14T04:06:50.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "allowed",
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 115,
@@ -38,7 +44,6 @@
         "network.transport": "tcp",
         "network.type": "ipv4",
         "related.ip": [
-            "175.16.199.1",
             "175.16.199.1"
         ],
         "service.type": "aws",
@@ -66,7 +71,7 @@
         "aws.vpcflow.action": "REJECT",
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "OK",
-        "aws.vpcflow.version": "2",
+        "aws.vpcflow.version": 2,
         "cloud.account.id": "123456789010",
         "cloud.provider": "aws",
         "destination.address": "175.16.199.1",
@@ -80,15 +85,21 @@
         "destination.geo.region_name": "Jilin Sheng",
         "destination.ip": "175.16.199.1",
         "destination.port": 3389,
-        "event.category": "network_traffic",
+        "event.action": "reject",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2014-12-14T04:07:50.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "2 123456789010 eni-1235b8ca123456789 175.16.199.1 175.16.199.1 49761 3389 6 20 4249 1418530010 1418530070 REJECT OK",
-        "event.outcome": "deny",
+        "event.outcome": "failure",
         "event.start": "2014-12-14T04:06:50.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "connection",
+            "denied"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 229,
@@ -99,7 +110,6 @@
         "network.transport": "tcp",
         "network.type": "ipv4",
         "related.ip": [
-            "175.16.199.1",
             "175.16.199.1"
         ],
         "service.type": "aws",
@@ -127,21 +137,27 @@
         "aws.vpcflow.action": "ACCEPT",
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "OK",
-        "aws.vpcflow.version": "2",
+        "aws.vpcflow.version": 2,
         "cloud.account.id": "123456789010",
         "cloud.provider": "aws",
         "destination.address": "172.31.16.139",
         "destination.ip": "172.31.16.139",
         "destination.port": 0,
-        "event.category": "network_traffic",
+        "event.action": "accept",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2015-05-29T16:32:22.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "2 123456789010 eni-1235b8ca123456789 203.0.113.12 172.31.16.139 0 0 1 4 336 1432917027 1432917142 ACCEPT OK",
-        "event.outcome": "allow",
+        "event.outcome": "success",
         "event.start": "2015-05-29T16:30:27.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "allowed",
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 345,
@@ -149,6 +165,7 @@
         "network.community_id": "1:H//CCQJhRqDUJ9c23S0VrQ+drxU=",
         "network.iana_number": "1",
         "network.packets": 4,
+        "network.transport": "icmp",
         "network.type": "ipv4",
         "related.ip": [
             "172.31.16.139",
@@ -171,21 +188,27 @@
         "aws.vpcflow.action": "REJECT",
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "OK",
-        "aws.vpcflow.version": "2",
+        "aws.vpcflow.version": 2,
         "cloud.account.id": "123456789010",
         "cloud.provider": "aws",
         "destination.address": "203.0.113.12",
         "destination.ip": "203.0.113.12",
         "destination.port": 0,
-        "event.category": "network_traffic",
+        "event.action": "reject",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2015-05-29T16:32:22.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "2 123456789010 eni-1235b8ca123456789 172.31.16.139 203.0.113.12 0 0 1 4 336 1432917094 1432917142 REJECT OK",
-        "event.outcome": "deny",
+        "event.outcome": "failure",
         "event.start": "2015-05-29T16:31:34.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "connection",
+            "denied"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 453,
@@ -193,6 +216,7 @@
         "network.community_id": "1:cfQqw/Kh6+4yqhEKgkCw/m3WoJM=",
         "network.iana_number": "1",
         "network.packets": 4,
+        "network.transport": "icmp",
         "network.type": "ipv4",
         "related.ip": [
             "172.31.16.139",
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/bad.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/bad.log-expected.json
index 7e762a71c7ef..e139eb6603f1 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/bad.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/bad.log-expected.json
@@ -1,12 +1,13 @@
 [
     {
         "cloud.provider": "aws",
-        "event.category": "network_traffic",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "Phony unsupported log format.",
-        "event.type": "flow",
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 0,
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/custom-nat-gateway.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/custom-nat-gateway.log-expected.json
index 11b49a3af565..337936a86059 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/custom-nat-gateway.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/custom-nat-gateway.log-expected.json
@@ -6,19 +6,24 @@
         "cloud.provider": "aws",
         "destination.address": "10.0.0.220",
         "destination.ip": "10.0.0.220",
-        "event.category": "network_traffic",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "- eni-1235b8ca123456789 10.0.1.5 10.0.0.220 10.0.1.5 203.0.113.5",
-        "event.type": "flow",
+        "event.type": [
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 65,
         "network.type": "ipv4",
         "related.ip": [
             "10.0.0.220",
-            "10.0.1.5"
+            "10.0.1.5",
+            "203.0.113.5"
         ],
         "service.type": "aws",
         "source.address": "10.0.1.5",
@@ -37,12 +42,16 @@
         "cloud.provider": "aws",
         "destination.address": "203.0.113.5",
         "destination.ip": "203.0.113.5",
-        "event.category": "network_traffic",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "i-01234567890123456 eni-1111aaaa2222bbbb3 10.0.1.5 203.0.113.5 10.0.1.5 203.0.113.5",
-        "event.type": "flow",
+        "event.type": [
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 130,
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/custom-transit-gateway.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/custom-transit-gateway.log-expected.json
index c0258dd5f8e5..fdb57a846d15 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/custom-transit-gateway.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/custom-transit-gateway.log-expected.json
@@ -8,13 +8,13 @@
         "aws.vpcflow.pkt_dstaddr": "10.40.2.236",
         "aws.vpcflow.pkt_srcaddr": "10.20.33.164",
         "aws.vpcflow.subnet_id": "subnet-22222222bbbbbbbbb",
-        "aws.vpcflow.tcp_flags": "3",
+        "aws.vpcflow.tcp_flags": 3,
         "aws.vpcflow.tcp_flags_array": [
             "fin",
             "syn"
         ],
         "aws.vpcflow.type": "IPv4",
-        "aws.vpcflow.version": "3",
+        "aws.vpcflow.version": 3,
         "aws.vpcflow.vpc_id": "vpc-abcdefab012345678",
         "cloud.account.id": "123456789010",
         "cloud.instance.id": "i-01234567890123456",
@@ -22,13 +22,19 @@
         "destination.address": "10.40.2.236",
         "destination.ip": "10.40.2.236",
         "destination.port": 80,
-        "event.category": "network_traffic",
+        "event.action": "accept",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "3 eni-33333333333333333 123456789010 vpc-abcdefab012345678 subnet-22222222bbbbbbbbb i-01234567890123456 10.20.33.164 10.40.2.236 39812 80 6 3 IPv4 10.20.33.164 10.40.2.236 ACCEPT OK",
-        "event.outcome": "allow",
-        "event.type": "flow",
+        "event.outcome": "success",
+        "event.type": [
+            "allowed",
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 159,
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/ipv6.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/ipv6.log-expected.json
index e487f326cf92..43cfadb44504 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/ipv6.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/ipv6.log-expected.json
@@ -5,21 +5,27 @@
         "aws.vpcflow.action": "ACCEPT",
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "OK",
-        "aws.vpcflow.version": "2",
+        "aws.vpcflow.version": 2,
         "cloud.account.id": "123456789010",
         "cloud.provider": "aws",
         "destination.address": "2001:db8:1234:a102:3304:8879:34cf:4071",
         "destination.ip": "2001:db8:1234:a102:3304:8879:34cf:4071",
         "destination.port": 22,
-        "event.category": "network_traffic",
+        "event.action": "accept",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2016-10-31T11:37:00.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "2 123456789010 eni-1235b8ca123456789 2001:db8:1234:a100:8d6e:3477:df66:f105 2001:db8:1234:a102:3304:8879:34cf:4071 34892 22 6 54 8855 1477913708 1477913820 ACCEPT OK",
-        "event.outcome": "allow",
+        "event.outcome": "success",
         "event.start": "2016-10-31T11:35:08.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "allowed",
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 0,
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/no-data-skip-data.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/no-data-skip-data.log-expected.json
index d055f270eccb..21968f5a52a8 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/no-data-skip-data.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/no-data-skip-data.log-expected.json
@@ -4,17 +4,21 @@
         "aws.vpcflow.account_id": "123456789010",
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "NODATA",
-        "aws.vpcflow.version": "2",
+        "aws.vpcflow.version": 2,
         "cloud.account.id": "123456789010",
         "cloud.provider": "aws",
-        "event.category": "network_traffic",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2015-05-10T18:02:14.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "2 123456789010 eni-1235b8ca123456789 - - - - - - - 1431280876 1431280934 - NODATA",
         "event.start": "2015-05-10T18:01:16.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 0,
@@ -29,17 +33,21 @@
         "aws.vpcflow.account_id": "123456789010",
         "aws.vpcflow.interface_id": "eni-11111111aaaaaaaaa",
         "aws.vpcflow.log_status": "SKIPDATA",
-        "aws.vpcflow.version": "2",
+        "aws.vpcflow.version": 2,
         "cloud.account.id": "123456789010",
         "cloud.provider": "aws",
-        "event.category": "network_traffic",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2015-05-10T18:02:14.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "2 123456789010 eni-11111111aaaaaaaaa - - - - - - - 1431280876 1431280934 - SKIPDATA",
         "event.start": "2015-05-10T18:01:16.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 82,
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence-skip-data.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence-skip-data.log-expected.json
index db3c56d1969d..d0d71dee2d5b 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence-skip-data.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence-skip-data.log-expected.json
@@ -6,19 +6,23 @@
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "SKIPDATA",
         "aws.vpcflow.subnet_id": "subnet-aaaaaaaa012345678",
-        "aws.vpcflow.version": "3",
+        "aws.vpcflow.version": 3,
         "aws.vpcflow.vpc_id": "vpc-abcdefab012345678",
         "cloud.account.id": "123456789010",
         "cloud.instance.id": "i-01234567890123456",
         "cloud.provider": "aws",
-        "event.category": "network_traffic",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2019-08-26T19:48:53.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "3 vpc-abcdefab012345678 subnet-aaaaaaaa012345678 i-01234567890123456 eni-1235b8ca123456789 123456789010 - - - - - - - - - - 1566848875 1566848933 - - SKIPDATA",
         "event.start": "2019-08-26T19:47:55.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 183,
@@ -35,19 +39,23 @@
         "aws.vpcflow.interface_id": "eni-1235b8ca123456789",
         "aws.vpcflow.log_status": "NODATA",
         "aws.vpcflow.subnet_id": "subnet-aaaaaaaa012345678",
-        "aws.vpcflow.version": "3",
+        "aws.vpcflow.version": 3,
         "aws.vpcflow.vpc_id": "vpc-abcdefab012345678",
         "cloud.account.id": "123456789010",
         "cloud.instance.id": "i-01234567890123456",
         "cloud.provider": "aws",
-        "event.category": "network_traffic",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2019-08-26T19:48:53.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "3 vpc-abcdefab012345678 subnet-aaaaaaaa012345678 i-01234567890123456 eni-1235b8ca123456789 123456789010 - - - - - - - - - - 1566848875 1566848933 - - NODATA",
         "event.start": "2019-08-26T19:47:55.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 526,
diff --git a/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence.log-expected.json b/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence.log-expected.json
index 26243b2065b6..bd2cf013344a 100644
--- a/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence.log-expected.json
+++ b/x-pack/filebeat/module/aws/vpcflow/test/tcp-flag-sequence.log-expected.json
@@ -9,12 +9,12 @@
         "aws.vpcflow.pkt_dstaddr": "10.0.0.62",
         "aws.vpcflow.pkt_srcaddr": "216.160.83.57",
         "aws.vpcflow.subnet_id": "subnet-aaaaaaaa012345678",
-        "aws.vpcflow.tcp_flags": "2",
+        "aws.vpcflow.tcp_flags": 2,
         "aws.vpcflow.tcp_flags_array": [
             "syn"
         ],
         "aws.vpcflow.type": "IPv4",
-        "aws.vpcflow.version": "3",
+        "aws.vpcflow.version": 3,
         "aws.vpcflow.vpc_id": "vpc-abcdefab012345678",
         "cloud.account.id": "123456789010",
         "cloud.instance.id": "i-01234567890123456",
@@ -22,15 +22,21 @@
         "destination.address": "10.0.0.62",
         "destination.ip": "10.0.0.62",
         "destination.port": 5001,
-        "event.category": "network_traffic",
+        "event.action": "accept",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2019-08-26T19:48:53.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "3 vpc-abcdefab012345678 subnet-aaaaaaaa012345678 i-01234567890123456 eni-1235b8ca123456789 123456789010 IPv4 216.160.83.57 10.0.0.62 43416 5001 216.160.83.57 10.0.0.62 6 568 8 1566848875 1566848933 ACCEPT 2 OK",
-        "event.outcome": "allow",
+        "event.outcome": "success",
         "event.start": "2019-08-26T19:47:55.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "allowed",
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 183,
@@ -74,13 +80,13 @@
         "aws.vpcflow.pkt_dstaddr": "10.0.0.62",
         "aws.vpcflow.pkt_srcaddr": "216.160.83.57",
         "aws.vpcflow.subnet_id": "subnet-aaaaaaaa012345678",
-        "aws.vpcflow.tcp_flags": "3",
+        "aws.vpcflow.tcp_flags": 3,
         "aws.vpcflow.tcp_flags_array": [
             "fin",
             "syn"
         ],
         "aws.vpcflow.type": "IPv4",
-        "aws.vpcflow.version": "3",
+        "aws.vpcflow.version": 3,
         "aws.vpcflow.vpc_id": "vpc-abcdefab012345678",
         "cloud.account.id": "123456789010",
         "cloud.instance.id": "i-01234567890123456",
@@ -88,15 +94,21 @@
         "destination.address": "10.0.0.62",
         "destination.ip": "10.0.0.62",
         "destination.port": 5001,
-        "event.category": "network_traffic",
+        "event.action": "accept",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2019-08-27T19:13:13.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "3 vpc-abcdefab012345678 subnet-aaaaaaaa012345678 i-01234567890123456 eni-1235b8ca123456789 123456789010 IPv4 216.160.83.57 10.0.0.62 43638 5001 216.160.83.57 10.0.0.62 6 1260 17 1566933133 1566933193 ACCEPT 3 OK",
-        "event.outcome": "allow",
+        "event.outcome": "success",
         "event.start": "2019-08-27T19:12:13.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "allowed",
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 393,
@@ -140,14 +152,14 @@
         "aws.vpcflow.pkt_dstaddr": "216.160.83.57",
         "aws.vpcflow.pkt_srcaddr": "10.0.0.62",
         "aws.vpcflow.subnet_id": "subnet-aaaaaaaa012345678",
-        "aws.vpcflow.tcp_flags": "19",
+        "aws.vpcflow.tcp_flags": 19,
         "aws.vpcflow.tcp_flags_array": [
             "ack",
             "fin",
             "syn"
         ],
         "aws.vpcflow.type": "IPv4",
-        "aws.vpcflow.version": "3",
+        "aws.vpcflow.version": 3,
         "aws.vpcflow.vpc_id": "vpc-abcdefab012345678",
         "cloud.account.id": "123456789010",
         "cloud.instance.id": "i-01234567890123456",
@@ -164,15 +176,21 @@
         "destination.geo.region_name": "Washington",
         "destination.ip": "216.160.83.57",
         "destination.port": 43638,
-        "event.category": "network_traffic",
+        "event.action": "accept",
+        "event.category": [
+            "network"
+        ],
         "event.dataset": "aws.vpcflow",
         "event.end": "2019-08-27T19:13:13.000Z",
         "event.kind": "event",
         "event.module": "aws",
         "event.original": "3 vpc-abcdefab012345678 subnet-aaaaaaaa012345678 i-01234567890123456 eni-1235b8ca123456789 123456789010 IPv4 10.0.0.62 216.160.83.57 5001 43638 10.0.0.62 216.160.83.57 6 967 14 1566933133 1566933193 ACCEPT 19 OK",
-        "event.outcome": "allow",
+        "event.outcome": "success",
         "event.start": "2019-08-27T19:12:13.000Z",
-        "event.type": "flow",
+        "event.type": [
+            "allowed",
+            "connection"
+        ],
         "fileset.name": "vpcflow",
         "input.type": "log",
         "log.offset": 605,
diff --git a/x-pack/filebeat/modules.d/aws.yml.disabled b/x-pack/filebeat/modules.d/aws.yml.disabled
index d58c52640025..c730b8aea074 100644
--- a/x-pack/filebeat/modules.d/aws.yml.disabled
+++ b/x-pack/filebeat/modules.d/aws.yml.disabled
@@ -379,3 +379,6 @@
 
     # Configures the SSL settings, ie. set trusted CAs, ignore certificate verification....
     #var.ssl:
+
+    # Specify a custom VPC flow log format.
+    #var.format: