diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 392867e5630..eda9e956a01 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -674,6 +674,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 
 - Upgrade to ECS 1.8.0. {pull}23783[23783]
+- Add `event.type: [connection]` to flow events and include `end` for final flows. {pull}24564[24564]
 
 *Functionbeat*
 
diff --git a/packetbeat/flows/worker.go b/packetbeat/flows/worker.go
index eebc3c260b5..1c9db02ff6f 100644
--- a/packetbeat/flows/worker.go
+++ b/packetbeat/flows/worker.go
@@ -220,6 +220,12 @@ func createEvent(
 		"category": []string{"network_traffic", "network"},
 		"action":   "network_flow",
 	}
+	eventType := []string{"connection"}
+	if isOver {
+		eventType = append(eventType, "end")
+	}
+	event["type"] = eventType
+
 	flow := common.MapStr{
 		"id":    common.NetString(f.id.Serialize()),
 		"final": isOver,
diff --git a/packetbeat/tests/system/test_0060_flows.py b/packetbeat/tests/system/test_0060_flows.py
index ab2e91bfbbc..ae6a4f923fb 100644
--- a/packetbeat/tests/system/test_0060_flows.py
+++ b/packetbeat/tests/system/test_0060_flows.py
@@ -47,6 +47,7 @@ def test_mysql_flow(self):
             'source.bytes': 1480,
             'destination.packets': 10,
             'destination.bytes': 181133,
+            'event.type': ['connection', 'end'],
         })
 
         start_ts = parse_timestamp(objs[0]['event.start'])