opensearch-project / opensearch-project/data-prepper
Feature request for merge_all aggregate action
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 374
- Forks
- 354
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 8
Description
Is your feature request related to a problem? Please describe.
It would be nice to have a merge aggregate action. There is no way to preserve values from different events in aggregate action for non identification_keys fields though they may contain useful data.
Describe the solution you'd like
Proposing a merge_all action which needs data_types for fields which we want to merge. The data type is needed for doing object comparisons.
Pipeline
aggregation-pipeline:
workers: 2
delay: "1000"
source:
file:
path: /usr/share/data-prepper/files/example.log
format: json
record_type: event
processor:
- aggregate:
identification_keys: ["ipv4_src_addr", "l4_src_port", "ipv4_dst_addr", "l4_dst_port", "protocol", "in_bytes", "in_pkts"]
group_duration: 4s
action:
merge_all:
data_types:
output_snmp: "integer"
flow_seq_num: "integer"
src_tos: "integer"
input_snmp: "integer"
l4_dst_port: "integer"
tcp_flags: "integer"
in_bytes: "integer"
in_pkts: "integer"
protocol: "integer"
flowset_id: "integer"
version: "integer"
dst_as: "integer"
ip_dscp: "integer"
l4_src_port: "integer"
ipv4_src_addr: "string"
first_switched: "string"
last_switched: "string"
ipv4_dst_addr: "string"
sink:
- stdout:
Input
{"output_snmp": 2, "flow_seq_num": 700710, "src_tos": 64, "input_snmp": 10, "l4_dst_port": 22, "tcp_flags": 16, "in_bytes": 40, "in_pkts": 1, "ipv4_src_addr": "10.35.197.104", "first_switched": "2023-01-11t08:37:22.783z", "protocol": 6, "flowset_id": 257, "last_switched": "2023-01-11t08:37:22.783z", "version": 9, "dst_as": 0, "ip_dscp": 16, "ipv4_dst_addr": "10.46.1.138", "l4_src_port": 52114}
{"output_snmp": 9, "flow_seq_num": 48159, "src_tos": 64, "input_snmp": 1, "l4_dst_port": 22, "tcp_flags": 16, "in_bytes": 40, "in_pkts": 1, "ipv4_src_addr": "10.35.197.104", "first_switched": "2023-01-11t08:37:23.057z", "protocol": 6, "flowset_id": 256, "last_switched": "2023-01-11t08:37:23.057z", "version": 9, "dst_as": 0, "ip_dscp": 16, "ipv4_dst_addr": "10.46.1.138", "l4_src_port": 52114}
Output
{"output_snmp":[9,2],"ip_dscp":16,"dst_as":0,"in_pkts":1,"src_tos":64,"ipv4_dst_addr":"10.46.1.138","first_switched":["2023-01-11t08:37:23.057z","2023-01-11t08:37:22.783z"],"flowset_id":[256,257],"l4_src_port":52114,"version":9,"flow_seq_num":[48159,700710],"ipv4_src_addr":"10.35.197.104","in_bytes":40,"protocol":6,"input_snmp":[1,10],"tcp_flags":16,"last_switched":["2023-01-11t08:37:23.057z","2023-01-11t08:37:22.783z"],"l4_dst_port":22}
Describe alternatives you've considered (Optional)
The put_all aggregate action keeps values from the last event thus losing information.
Additional context
The use case is to stitch different netflow logs into one.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by locating the aggregate processor and the existing put_all action, then trace how action configuration and field values are handled. Use the supplied pipeline and input/output examples as the behavioral reference; done means a merge_all action can preserve configured fields as merged values while retaining other aggregate behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100