[HUDI-4445] S3 Incremental source improvements #6176
There are no files selected for viewing
This file contains hidden or 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 hidden or 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,70 @@ | ||
| package org.apache.hudi.common.testutils; | ||
|
|
||
| import org.apache.avro.Schema; | ||
| import org.apache.avro.SchemaBuilder; | ||
| import org.apache.avro.generic.GenericData; | ||
| import org.apache.avro.generic.GenericRecord; | ||
|
|
||
| // Utility for the schema of S3 events listed here (https://docs.aws.amazon.com/AmazonS3/latest/userguide/notification-content-structure.html) | ||
|
|
||
| public class S3EventsSchemaUtils { | ||
| public static final String DEFAULT_STRING_VALUE = "default_string"; | ||
|
|
||
| public static String generateSchemaString() { | ||
| return generateS3EventSchema().toString(); | ||
| } | ||
|
|
||
| public static Schema generateObjInfoSchema() { | ||
| Schema objInfo = SchemaBuilder.record("objInfo") | ||
| .fields() | ||
| .requiredString("key") | ||
| .requiredLong("size") | ||
| .endRecord(); | ||
| return objInfo; | ||
| } | ||
|
|
||
| public static GenericRecord generateObjInfoRecord(String key, Long size) { | ||
| GenericRecord rec = new GenericData.Record(generateObjInfoSchema()); | ||
| rec.put("key", key); | ||
| rec.put("size", size); | ||
| return rec; | ||
| } | ||
|
|
||
| public static Schema generateS3MetadataSchema() { | ||
| Schema s3Metadata = SchemaBuilder.record("s3Metadata") | ||
| .fields() | ||
| .requiredString("configurationId") | ||
| .name("object") | ||
| .type(generateObjInfoSchema()) | ||
| .noDefault() | ||
| .endRecord(); | ||
| return s3Metadata; | ||
| } | ||
|
|
||
| public static GenericRecord generateS3MetadataRecord(GenericRecord objRecord) { | ||
| GenericRecord rec = new GenericData.Record(generateS3MetadataSchema()); | ||
| rec.put("configurationId", DEFAULT_STRING_VALUE); | ||
| rec.put("object", objRecord); | ||
| return rec; | ||
| } | ||
|
|
||
| public static Schema generateS3EventSchema() { | ||
| Schema s3Event = SchemaBuilder.record("s3Event") | ||
| .fields() | ||
| .requiredString("eventSource") | ||
| .requiredString("eventName") | ||
| .name("s3") | ||
|
Comment on lines
+53
to
+93
Member
There was a problem hiding this comment. Let's extract all these strings to constants.
Member
There was a problem hiding this comment. Preferably extract all these as static string constants. |
||
| .type(generateS3MetadataSchema()) | ||
| .noDefault() | ||
| .endRecord(); | ||
| return s3Event; | ||
| } | ||
|
|
||
| public static GenericRecord generateS3EventRecord(String rowKey, GenericRecord objRecord) { | ||
| GenericRecord rec = new GenericData.Record(generateS3EventSchema()); | ||
| rec.put("_row_key", rowKey); | ||
| rec.put("eventSource", DEFAULT_STRING_VALUE); | ||
| rec.put("eventName", DEFAULT_STRING_VALUE); | ||
| rec.put("s3", generateS3MetadataRecord(objRecord)); | ||
| return rec; | ||
| } | ||
| } | ||
This file contains hidden or 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 hidden or 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,134 @@ | ||
| package org.apache.hudi.utilities.sources; | ||
|
|
||
| import org.apache.avro.Schema; | ||
| import org.apache.hudi.client.SparkRDDWriteClient; | ||
| import org.apache.hudi.client.WriteStatus; | ||
| import org.apache.hudi.common.config.HoodieMetadataConfig; | ||
| import org.apache.hudi.common.config.TypedProperties; | ||
| import org.apache.hudi.common.model.HoodieRecord; | ||
| import org.apache.hudi.common.table.HoodieTableMetaClient; | ||
| import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion; | ||
| import org.apache.hudi.common.testutils.HoodieTestDataGenerator; | ||
| import org.apache.hudi.common.testutils.S3EventsSchemaUtils; | ||
| import org.apache.hudi.common.util.Option; | ||
| import org.apache.hudi.common.util.collection.Pair; | ||
| import org.apache.hudi.config.HoodieArchivalConfig; | ||
| import org.apache.hudi.config.HoodieCleanConfig; | ||
| import org.apache.hudi.config.HoodieWriteConfig; | ||
| import org.apache.hudi.testutils.SparkClientFunctionalTestHarness; | ||
| import org.apache.hudi.utilities.schema.SchemaProvider; | ||
| import org.apache.hudi.utilities.sources.helpers.IncrSourceHelper; | ||
| import org.apache.spark.api.java.JavaRDD; | ||
| import org.apache.spark.sql.Dataset; | ||
| import org.apache.spark.sql.Row; | ||
| import org.junit.jupiter.api.Assertions; | ||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
| import java.util.Properties; | ||
|
|
||
| import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.S3_EVENTS_SCHEMA; | ||
| import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors; | ||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
|
|
||
| public class TestS3EventsHoodieIncrSource extends SparkClientFunctionalTestHarness { | ||
| private HoodieTestDataGenerator dataGen; | ||
| private HoodieTableMetaClient metaClient; | ||
|
|
||
| @BeforeEach | ||
| public void setUp() throws IOException { | ||
| dataGen = new HoodieTestDataGenerator(); | ||
| metaClient = getHoodieMetaClient(hadoopConf(), basePath()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testHoodieIncrSource() throws IOException { | ||
|
Member
There was a problem hiding this comment. Maybe rename to testS3EventsHoodieIncrSource? |
||
| HoodieWriteConfig writeConfig = getConfigBuilder(basePath(), metaClient) | ||
| .withArchivalConfig(HoodieArchivalConfig.newBuilder().archiveCommitsWith(2, 3).build()) | ||
| .withCleanConfig(HoodieCleanConfig.newBuilder().retainCommits(1).build()) | ||
| .withMetadataConfig(HoodieMetadataConfig.newBuilder() | ||
| .withMaxNumDeltaCommitsBeforeCompaction(1).build()) | ||
| .build(); | ||
|
|
||
| SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig); | ||
| Pair<String, List<HoodieRecord>> inserts = writeRecords(writeClient, null, "100"); | ||
| Pair<String, List<HoodieRecord>> inserts2 = writeRecords(writeClient, null, "200"); | ||
| Pair<String, List<HoodieRecord>> inserts3 = writeRecords(writeClient, null, "300"); | ||
| Pair<String, List<HoodieRecord>> inserts4 = writeRecords(writeClient, null, "400"); | ||
| Pair<String, List<HoodieRecord>> inserts5 = writeRecords(writeClient, null, "500"); | ||
|
|
||
| // read everything upto latest | ||
| readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.empty(), 500, inserts5.getKey()); | ||
|
|
||
| // even if the begin timestamp is archived (100), full table scan should kick in, but should filter for records having commit time > 100 | ||
| readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.of("100"), 400, inserts5.getKey()); | ||
|
|
||
| // even if the read upto latest is set, if begin timestamp is in active timeline, only incremental should kick in. | ||
| readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_UPTO_LATEST_COMMIT, Option.of("400"), 100, inserts5.getKey()); | ||
|
|
||
| // read just the latest | ||
| readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.empty(), 100, inserts5.getKey()); | ||
|
|
||
| // ensure checkpoint does not move | ||
| readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.of(inserts5.getKey()), 0, inserts5.getKey()); | ||
|
|
||
| Pair<String, List<HoodieRecord>> inserts6 = writeRecords(writeClient, null, "600"); | ||
|
|
||
| // insert new batch and ensure the checkpoint moves | ||
| readAndAssert(IncrSourceHelper.MissingCheckpointStrategy.READ_LATEST, Option.of(inserts5.getKey()), 100, inserts6.getKey()); | ||
| writeClient.close(); | ||
| } | ||
|
|
||
| private void readAndAssert(IncrSourceHelper.MissingCheckpointStrategy missingCheckpointStrategy, Option<String> checkpointToPull, int expectedCount, String expectedCheckpoint) { | ||
|
|
||
| Properties properties = new Properties(); | ||
| properties.setProperty("hoodie.deltastreamer.source.hoodieincr.path", basePath()); | ||
| properties.setProperty("hoodie.deltastreamer.source.hoodieincr.missing.checkpoint.strategy", missingCheckpointStrategy.name()); | ||
| TypedProperties typedProperties = new TypedProperties(properties); | ||
| S3EventsHoodieIncrSource s3IncrSource = new S3EventsHoodieIncrSource(typedProperties, jsc(), spark(), new DummySchemaProvider(S3EventsSchemaUtils.generateS3EventSchema())); | ||
|
|
||
| // read everything until latest | ||
| Pair<Option<Dataset<Row>>, String> batchCheckPoint = s3IncrSource.fetchNextBatch(checkpointToPull, 500); | ||
| Assertions.assertNotNull(batchCheckPoint.getValue()); | ||
| if (expectedCount == 0) { | ||
| assertFalse(batchCheckPoint.getKey().isPresent()); | ||
| } else { | ||
| assertEquals(batchCheckPoint.getKey().get().count(), expectedCount); | ||
| } | ||
| Assertions.assertEquals(batchCheckPoint.getRight(), expectedCheckpoint); | ||
| } | ||
|
|
||
| private Pair<String, List<HoodieRecord>> writeRecords(SparkRDDWriteClient writeClient, List<HoodieRecord> insertRecords, String commit) throws IOException { | ||
| writeClient.startCommitWithTime(commit); | ||
| List<HoodieRecord> records = dataGen.generateInsertsWithSchema(commit, 100, S3_EVENTS_SCHEMA); | ||
| JavaRDD<WriteStatus> result = writeClient.upsert(jsc().parallelize(records, 1), commit); | ||
| List<WriteStatus> statuses = result.collect(); | ||
| assertNoWriteErrors(statuses); | ||
| return Pair.of(commit, records); | ||
| } | ||
|
|
||
| private HoodieWriteConfig.Builder getConfigBuilder(String basePath, HoodieTableMetaClient metaClient) { | ||
| return HoodieWriteConfig.newBuilder().withPath(basePath).withSchema(S3_EVENTS_SCHEMA) | ||
| .withParallelism(2, 2).withBulkInsertParallelism(2).withFinalizeWriteParallelism(2).withDeleteParallelism(2) | ||
| .withTimelineLayoutVersion(TimelineLayoutVersion.CURR_VERSION) | ||
| .forTable(metaClient.getTableConfig().getTableName()); | ||
| } | ||
|
|
||
| private static class DummySchemaProvider extends SchemaProvider { | ||
|
|
||
| private final Schema schema; | ||
|
|
||
| public DummySchemaProvider(Schema schema) { | ||
| super(new TypedProperties()); | ||
| this.schema = schema; | ||
| } | ||
|
|
||
| @Override | ||
| public Schema getSourceSchema() { | ||
| return schema; | ||
| } | ||
| } | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
RawTripTestPayloadassumes some form of trips schema. If you look at its constructor, we don't use the schema. And its APIs assume a few things about the schema. Should we keep all this out ofHoodieTestDataGenerator?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is an old comment. Please check if it's still valid.