forked from opensearch-project/data-prepper
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Load exported S3 files in RDS source (opensearch-project#4718)
* Add s3 file loader Signed-off-by: Hai Yan <[email protected]> * Make checkExportStatus a callable Signed-off-by: Hai Yan <[email protected]> * Fix unit tests Signed-off-by: Hai Yan <[email protected]> * Add load status and record converter Signed-off-by: Hai Yan <[email protected]> * Update unit tests Signed-off-by: Hai Yan <[email protected]> * Restore changes for test Signed-off-by: Hai Yan <[email protected]> * Address review comments Signed-off-by: Hai Yan <[email protected]> --------- Signed-off-by: Hai Yan <[email protected]>
- Loading branch information
Showing
23 changed files
with
1,133 additions
and
32 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
36 changes: 36 additions & 0 deletions
36
...n/java/org/opensearch/dataprepper/plugins/source/rds/converter/ExportRecordConverter.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,36 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.converter; | ||
|
|
||
| import org.opensearch.dataprepper.model.event.Event; | ||
| import org.opensearch.dataprepper.model.event.EventMetadata; | ||
| import org.opensearch.dataprepper.model.record.Record; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import static org.opensearch.dataprepper.plugins.source.rds.converter.MetadataKeyAttributes.EVENT_TABLE_NAME_METADATA_ATTRIBUTE; | ||
| import static org.opensearch.dataprepper.plugins.source.rds.converter.MetadataKeyAttributes.INGESTION_EVENT_TYPE_ATTRIBUTE; | ||
| import static org.opensearch.dataprepper.plugins.source.rds.converter.MetadataKeyAttributes.PRIMARY_KEY_DOCUMENT_ID_METADATA_ATTRIBUTE; | ||
|
|
||
| public class ExportRecordConverter { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(ExportRecordConverter.class); | ||
|
|
||
| static final String EXPORT_EVENT_TYPE = "EXPORT"; | ||
|
|
||
| public Event convert(Record<Event> record, String tableName, String primaryKeyName) { | ||
| Event event = record.getData(); | ||
|
|
||
| EventMetadata eventMetadata = event.getMetadata(); | ||
| eventMetadata.setAttribute(EVENT_TABLE_NAME_METADATA_ATTRIBUTE, tableName); | ||
| eventMetadata.setAttribute(INGESTION_EVENT_TYPE_ATTRIBUTE, EXPORT_EVENT_TYPE); | ||
|
|
||
| final Object primaryKeyValue = record.getData().get(primaryKeyName, Object.class); | ||
| eventMetadata.setAttribute(PRIMARY_KEY_DOCUMENT_ID_METADATA_ATTRIBUTE, primaryKeyValue); | ||
|
|
||
| return event; | ||
| } | ||
| } |
20 changes: 20 additions & 0 deletions
20
...n/java/org/opensearch/dataprepper/plugins/source/rds/converter/MetadataKeyAttributes.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,20 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.converter; | ||
|
|
||
| public class MetadataKeyAttributes { | ||
| static final String PRIMARY_KEY_DOCUMENT_ID_METADATA_ATTRIBUTE = "primary_key"; | ||
|
|
||
| static final String EVENT_VERSION_FROM_TIMESTAMP = "document_version"; | ||
|
|
||
| static final String EVENT_TIMESTAMP_METADATA_ATTRIBUTE = "event_timestamp"; | ||
|
|
||
| static final String EVENT_NAME_BULK_ACTION_METADATA_ATTRIBUTE = "opensearch_action"; | ||
|
|
||
| static final String EVENT_TABLE_NAME_METADATA_ATTRIBUTE = "table_name"; | ||
|
|
||
| static final String INGESTION_EVENT_TYPE_ATTRIBUTE = "ingestion_type"; | ||
| } |
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
77 changes: 77 additions & 0 deletions
77
...g/opensearch/dataprepper/plugins/source/rds/coordination/partition/DataFilePartition.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,77 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.coordination.partition; | ||
|
|
||
| import org.opensearch.dataprepper.model.source.coordinator.SourcePartitionStoreItem; | ||
| import org.opensearch.dataprepper.model.source.coordinator.enhanced.EnhancedSourcePartition; | ||
| import org.opensearch.dataprepper.plugins.source.rds.coordination.state.DataFileProgressState; | ||
|
|
||
| import java.util.Optional; | ||
|
|
||
| /** | ||
| * An DataFilePartition represents an export data file needs to be loaded. | ||
| * The source identifier contains keyword 'DATAFILE' | ||
| */ | ||
| public class DataFilePartition extends EnhancedSourcePartition<DataFileProgressState> { | ||
|
|
||
| public static final String PARTITION_TYPE = "DATAFILE"; | ||
|
|
||
| private final String exportTaskId; | ||
| private final String bucket; | ||
| private final String key; | ||
| private final DataFileProgressState state; | ||
|
|
||
| public DataFilePartition(final SourcePartitionStoreItem sourcePartitionStoreItem) { | ||
|
|
||
| setSourcePartitionStoreItem(sourcePartitionStoreItem); | ||
| String[] keySplits = sourcePartitionStoreItem.getSourcePartitionKey().split("\\|"); | ||
| exportTaskId = keySplits[0]; | ||
| bucket = keySplits[1]; | ||
| key = keySplits[2]; | ||
| state = convertStringToPartitionProgressState(DataFileProgressState.class, sourcePartitionStoreItem.getPartitionProgressState()); | ||
|
|
||
| } | ||
|
|
||
| public DataFilePartition(final String exportTaskId, | ||
| final String bucket, | ||
| final String key, | ||
| final Optional<DataFileProgressState> state) { | ||
| this.exportTaskId = exportTaskId; | ||
| this.bucket = bucket; | ||
| this.key = key; | ||
| this.state = state.orElse(null); | ||
| } | ||
|
|
||
| @Override | ||
| public String getPartitionType() { | ||
| return PARTITION_TYPE; | ||
| } | ||
|
|
||
| @Override | ||
| public String getPartitionKey() { | ||
| return exportTaskId + "|" + bucket + "|" + key; | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<DataFileProgressState> getProgressState() { | ||
| if (state != null) { | ||
| return Optional.of(state); | ||
| } | ||
| return Optional.empty(); | ||
| } | ||
|
|
||
| public String getExportTaskId() { | ||
| return exportTaskId; | ||
| } | ||
|
|
||
| public String getBucket() { | ||
| return bucket; | ||
| } | ||
|
|
||
| public String getKey() { | ||
| return key; | ||
| } | ||
| } |
44 changes: 44 additions & 0 deletions
44
...g/opensearch/dataprepper/plugins/source/rds/coordination/state/DataFileProgressState.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,44 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.coordination.state; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
|
|
||
| public class DataFileProgressState { | ||
|
|
||
| @JsonProperty("isLoaded") | ||
| private boolean isLoaded = false; | ||
|
|
||
| @JsonProperty("totalRecords") | ||
| private int totalRecords; | ||
|
|
||
| @JsonProperty("sourceTable") | ||
| private String sourceTable; | ||
|
|
||
| public int getTotalRecords() { | ||
| return totalRecords; | ||
| } | ||
|
|
||
| public void setTotalRecords(int totalRecords) { | ||
| this.totalRecords = totalRecords; | ||
| } | ||
|
|
||
| public boolean getLoaded() { | ||
| return isLoaded; | ||
| } | ||
|
|
||
| public void setLoaded(boolean loaded) { | ||
| this.isLoaded = loaded; | ||
| } | ||
|
|
||
| public String getSourceTable() { | ||
| return sourceTable; | ||
| } | ||
|
|
||
| public void setSourceTable(String sourceTable) { | ||
| this.sourceTable = sourceTable; | ||
| } | ||
| } |
Oops, something went wrong.