-
Notifications
You must be signed in to change notification settings - Fork 198
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Improve OpenSearch sink performance by creating a customer JsonpMappe…
…r which avoids re-serializing bulk documents. (#1391) Signed-off-by: David Venable <[email protected]>
- Loading branch information
Showing
16 changed files
with
395 additions
and
284 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
84 changes: 84 additions & 0 deletions
84
...in/java/com/amazon/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package com.amazon.dataprepper.plugins.sink.opensearch.bulk; | ||
|
|
||
| import jakarta.json.spi.JsonProvider; | ||
| import jakarta.json.stream.JsonGenerator; | ||
| import jakarta.json.stream.JsonParser; | ||
| import org.opensearch.client.json.JsonpMapper; | ||
| import org.opensearch.client.json.jackson.JacksonJsonProvider; | ||
| import org.opensearch.client.json.jackson.JacksonJsonpGenerator; | ||
| import org.opensearch.client.json.jackson.JacksonJsonpMapper; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.OutputStream; | ||
|
|
||
| /** | ||
| * An implementation of the opensearch-java client's {@link JsonpMapper}. It can avoid duplicate | ||
| * serialization by use of the {@link SerializedJson} interface. For values that inherit from | ||
| * {@link SerializedJson}, it will not re-serialize these. For other values, it uses Jackson | ||
| * serialization via the {@link JacksonJsonpMapper}. | ||
| */ | ||
| public class PreSerializedJsonpMapper implements JsonpMapper { | ||
|
|
||
| private final JsonpMapper innerMapper; | ||
| private final PreSerializedJsonProvider jsonProvider; | ||
|
|
||
| public PreSerializedJsonpMapper() { | ||
| innerMapper = new JacksonJsonpMapper(); | ||
| jsonProvider = new PreSerializedJsonProvider(); | ||
| } | ||
|
|
||
| @Override | ||
| public JsonProvider jsonProvider() { | ||
| return jsonProvider; | ||
| } | ||
|
|
||
| @Override | ||
| public <T> T deserialize(JsonParser parser, Class<T> clazz) { | ||
| return innerMapper.deserialize(parser, clazz); | ||
| } | ||
|
|
||
| @Override | ||
| public <T> void serialize(T value, JsonGenerator generator) { | ||
| if(value instanceof SerializedJson) { | ||
| if (! (generator instanceof PreSerializedJsonGenerator)) | ||
| throw new IllegalArgumentException("Unsupported JsonGenerator"); | ||
|
|
||
| final OutputStream outputStream = ((PreSerializedJsonGenerator) generator).outputStream; | ||
|
|
||
| try { | ||
| outputStream.write(((SerializedJson) value).getSerializedJson()); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
|
|
||
| } else { | ||
| innerMapper.serialize(value, generator); | ||
| } | ||
| } | ||
|
|
||
| private static class PreSerializedJsonGenerator extends JacksonJsonpGenerator { | ||
| private OutputStream outputStream; | ||
|
|
||
| public PreSerializedJsonGenerator(com.fasterxml.jackson.core.JsonGenerator generator, OutputStream outputStream) { | ||
| super(generator); | ||
| this.outputStream = outputStream; | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private static class PreSerializedJsonProvider extends JacksonJsonProvider { | ||
| @Override | ||
| public JsonGenerator createGenerator(OutputStream out) { | ||
| try { | ||
| return new PreSerializedJsonGenerator(jacksonJsonFactory().createGenerator(out), out); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
| } | ||
| } |
28 changes: 28 additions & 0 deletions
28
...rch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/bulk/SerializedJson.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package com.amazon.dataprepper.plugins.sink.opensearch.bulk; | ||
|
|
||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * Represents JSON which is already serialized for use in the {@link PreSerializedJsonpMapper}. | ||
| */ | ||
| public interface SerializedJson extends SizedDocument { | ||
|
|
||
| byte[] getSerializedJson(); | ||
|
|
||
| /** | ||
| * Creates a new {@link SerializedJson} from a JSON string. | ||
| * | ||
| * @param jsonString The serialized JSON string which forms this JSON data. | ||
| * @return A new {@link SerializedJson}. | ||
| */ | ||
| static SerializedJson fromString(String jsonString) { | ||
| Objects.requireNonNull(jsonString); | ||
| return new SerializedJsonImpl(jsonString.getBytes(StandardCharsets.UTF_8)); | ||
| } | ||
| } |
24 changes: 24 additions & 0 deletions
24
...src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/bulk/SerializedJsonImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package com.amazon.dataprepper.plugins.sink.opensearch.bulk; | ||
|
|
||
| class SerializedJsonImpl implements SerializedJson { | ||
| private byte[] document; | ||
|
|
||
| public SerializedJsonImpl(final byte[] document) { | ||
| this.document = document; | ||
| } | ||
|
|
||
| @Override | ||
| public long getDocumentSize() { | ||
| return document.length; | ||
| } | ||
|
|
||
| @Override | ||
| public byte[] getSerializedJson() { | ||
| return document; | ||
| } | ||
| } |
18 changes: 18 additions & 0 deletions
18
...arch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/bulk/SizedDocument.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package com.amazon.dataprepper.plugins.sink.opensearch.bulk; | ||
|
|
||
| /** | ||
| * Represents an OpenSearch document with a serialized document size. | ||
| */ | ||
| public interface SizedDocument { | ||
| /** | ||
| * The size of this document. | ||
| * | ||
| * @return The document size in bytes | ||
| */ | ||
| long getDocumentSize(); | ||
| } |
41 changes: 0 additions & 41 deletions
41
...arch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/bulk/SizedJsonData.java
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.