Skip to content

Commit bb121e8

Browse files
srikanthjgsrikanthjg
authored
Add support for spilt event processor (#4166)
* Add support for spilt event processor Signed-off-by: srigovs <srigovs@amazon.com> * Add support for spilt event processor(#4089) Signed-off-by: srigovs <srigovs@amazon.com> * Add support for spilt event processor(#4089) Signed-off-by: srigovs <srigovs@amazon.com> * Add support for spilt event processor(#4089) Signed-off-by: srigovs <srigovs@amazon.com> --------- Signed-off-by: srigovs <srigovs@amazon.com> Signed-off-by: Srikanth Govindarajan <srikanthjg123@gmail.com>
1 parent 8c1e8cd commit bb121e8

File tree

5 files changed

+585
-1
lines changed

5 files changed

+585
-1
lines changed

‎data-prepper-plugins/split-event-processor/build.gradle‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
jacocoTestCoverageVerification {
7+
dependsOn jacocoTestReport
8+
violationRules {
9+
rule {
10+
limit {
11+
minimum = 1.0
12+
}
13+
}
14+
}
15+
}
16+
17+
18+
dependencies {
19+
implementation project(':data-prepper-api')
20+
implementation project(':data-prepper-plugins:common')
21+
implementation 'com.fasterxml.jackson.core:jackson-databind'
22+
}

‎data-prepper-plugins/split-event-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/splitevent/SplitEventProcessor.java‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*
9+
*/
10+
11+
package org.opensearch.dataprepper.plugins.processor.splitevent;
12+
13+
import org.opensearch.dataprepper.metrics.PluginMetrics;
14+
import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
15+
import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
16+
import org.opensearch.dataprepper.model.event.DefaultEventHandle;
17+
import org.opensearch.dataprepper.model.event.Event;
18+
import org.opensearch.dataprepper.model.event.JacksonEvent;
19+
import org.opensearch.dataprepper.model.processor.AbstractProcessor;
20+
import org.opensearch.dataprepper.model.processor.Processor;
21+
import org.opensearch.dataprepper.model.record.Record;
22+
23+
import java.util.ArrayList;
24+
import java.util.Collection;
25+
import java.util.function.Function;
26+
import java.util.regex.Pattern;
27+
28+
29+
@DataPrepperPlugin(name = "split_event", pluginType = Processor.class, pluginConfigurationType = SplitEventProcessorConfig.class)
30+
public class SplitEventProcessor extends AbstractProcessor<Record<Event>, Record<Event>>{
31+
final String delimiter;
32+
final String delimiterRegex;
33+
final String field;
34+
final Pattern pattern;
35+
private final Function<String, String[]> splitter;
36+
37+
@DataPrepperPluginConstructor
38+
public SplitEventProcessor(final PluginMetrics pluginMetrics, final SplitEventProcessorConfig config) {
39+
super(pluginMetrics);
40+
this.delimiter = config.getDelimiter();
41+
this.delimiterRegex = config.getDelimiterRegex();
42+
this.field = config.getField();
43+
44+
if(delimiterRegex != null && !delimiterRegex.isEmpty()
45+
&& delimiter != null && !delimiter.isEmpty()) {
46+
throw new IllegalArgumentException("delimiter and delimiter_regex cannot be defined at the same time");
47+
} else if((delimiterRegex == null || delimiterRegex.isEmpty()) &&
48+
(delimiter == null || delimiter.isEmpty())) {
49+
throw new IllegalArgumentException("delimiter or delimiter_regex needs to be defined");
50+
}
51+
52+
if(delimiterRegex != null && !delimiterRegex.isEmpty()) {
53+
pattern = Pattern.compile(delimiterRegex);
54+
splitter = pattern::split;
55+
} else {
56+
splitter = inputString -> inputString.split(delimiter);
57+
pattern = null;
58+
}
59+
}
60+
61+
@Override
62+
public Collection<Record<Event>> doExecute(final Collection<Record<Event>> records) {
63+
Collection<Record<Event>> newRecords = new ArrayList<>();
64+
for(final Record<Event> record : records) {
65+
final Event recordEvent = record.getData();
66+
67+
if (!recordEvent.containsKey(field)) {
68+
newRecords.add(record);
69+
continue;
70+
}
71+
72+
final Object value = recordEvent.get(field, Object.class);
73+
74+
//split record according to delimiter
75+
final String[] splitValues = splitter.apply((String) value);
76+
77+
// when no splits or empty value use the original record
78+
if(splitValues.length <= 1) {
79+
newRecords.add(record);
80+
continue;
81+
}
82+
83+
//create new events for the splits
84+
for (int i = 0; i < splitValues.length-1 ; i++) {
85+
Record newRecord = createNewRecordFromEvent(recordEvent, splitValues[i]);
86+
addToAcknowledgementSetFromOriginEvent((Event) newRecord.getData(), recordEvent);
87+
newRecords.add(newRecord);
88+
}
89+
90+
// Modify original event to hold the last split
91+
recordEvent.put(field, splitValues[splitValues.length-1]);
92+
newRecords.add(record);
93+
}
94+
return newRecords;
95+
}
96+
97+
protected Record createNewRecordFromEvent(final Event recordEvent, String splitValue) {
98+
Record newRecord;
99+
JacksonEvent newRecordEvent;
100+
101+
newRecordEvent = JacksonEvent.fromEvent(recordEvent);
102+
newRecordEvent.put(field,(Object) splitValue);
103+
newRecord = new Record<>(newRecordEvent);
104+
return newRecord;
105+
}
106+
107+
protected void addToAcknowledgementSetFromOriginEvent(Event recordEvent, Event originRecordEvent) {
108+
DefaultEventHandle eventHandle = (DefaultEventHandle) originRecordEvent.getEventHandle();
109+
if (eventHandle != null && eventHandle.getAcknowledgementSet() != null) {
110+
eventHandle.getAcknowledgementSet().add(recordEvent);
111+
}
112+
}
113+
114+
@Override
115+
public void prepareForShutdown() {
116+
}
117+
118+
@Override
119+
public boolean isReadyForShutdown() {
120+
return true;
121+
}
122+
123+
@Override
124+
public void shutdown() {
125+
}
126+
}

‎data-prepper-plugins/split-event-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/splitevent/SplitEventProcessorConfig.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*
9+
*/
10+
11+
package org.opensearch.dataprepper.plugins.processor.splitevent;
12+
13+
import com.fasterxml.jackson.annotation.JsonProperty;
14+
import jakarta.validation.constraints.NotEmpty;
15+
import jakarta.validation.constraints.NotNull;
16+
import jakarta.validation.constraints.Size;
17+
18+
19+
public class SplitEventProcessorConfig {
20+
@NotEmpty
21+
@NotNull
22+
@JsonProperty("field")
23+
private String field;
24+
25+
@JsonProperty("delimiter_regex")
26+
private String delimiterRegex;
27+
28+
@Size(min = 1, max = 1)
29+
private String delimiter;
30+
31+
public String getField() {
32+
return field;
33+
}
34+
35+
public String getDelimiterRegex() {
36+
return delimiterRegex;
37+
}
38+
39+
public String getDelimiter() {
40+
return delimiter;
41+
}
42+
}

0 commit comments

Comments
 (0)