Repository navigation
Add format auto-detection for zero-config S3 ingestion - #6990
Davidding4718 wants to merge 1 commit into
Conversation
✅ License Header Check PassedAll newly added files have proper license headers. Great work! 🎉 |
dlvenable
left a comment
dlvenable
left a comment
There was a problem hiding this comment.
Thanks @Davidding4718 for this contribution!
There was a problem hiding this comment.
Let's not check in the compressed files. You can use the existing files like sample.ndjson and then compress them as part of the test setup.
There was a problem hiding this comment.
Can we generate the Parquet as part of the test setup as well? Binary files are hard to manage in a project.
| throw context.weirdStringException(value, Map.class, | ||
| "Empty string is not allowed for plugin '" + pluginName + "'. Use null, empty (no value), or {} instead."); | ||
| // Treat empty string same as null (YAML bare keys like "stdout:" parse as "" in Jackson 2.13+) | ||
| isNull = true; |
There was a problem hiding this comment.
It looks like you have some files from other PRs. Please be sure this PR only has the code for auto-detection. You may need to rebase from upstream/main.
| } | ||
|
|
||
|
|
||
| task runDataPrepper(type: JavaExec) { |
|
|
||
| private static final int SAMPLE_SIZE = 65536; | ||
|
|
||
| public static void main(final String[] args) throws IOException { |
There was a problem hiding this comment.
Overall, this should provide a new codec rather than a new application.
Also, we should not have another main() method in the plugins.
| // Auto-detect mode: no codec specified, detect format per-object | ||
| LOG.info("No codec configured — using auto-detect mode"); | ||
| final AutoDetectCodecFactory autoDetectCodecFactory = new AutoDetectCodecFactory(pluginFactory); | ||
| final S3ObjectRequest s3ObjectRequest = s3ObjectRequestBuilder |
There was a problem hiding this comment.
Let's not create a split.
A better design would be to include a Function<S3InputFile, InputCodec>. Then change the builder to take that instead.
Your code would look more like this:
Function<S3InputFile, InputCodec> inputCodecProvider;
if (codecConfiguration != null) {
inputCodecProvider = new AutoDetectCodecFactory(pluginFactory);
} else {
final InputCodec codec = pluginFactory.loadPlugin(InputCodec.class, codecPluginSettings);
inputCodecProvider = s3InputFile -> codec;
}
final S3ObjectRequest s3ObjectRequest = s3ObjectRequestBuilder
.bucketOwnerProvider(bucketOwnerProvider)
.codecProvider(inputCodecProvider)
...
There was a problem hiding this comment.
I took another look and see that S3InputFile has already loaded the object. Maybe we should make this a Function<S3ObjectReference, InputCodec>.
| * S3ObjectHandler that auto-detects format and compression before processing. | ||
| * Used when no codec is explicitly configured in the pipeline. | ||
| */ | ||
| class AutoDetectS3ObjectWorker implements S3ObjectHandler { |
There was a problem hiding this comment.
We should not fork this. The primary difference is that we need to sample the document.
Instead of create a new S3ObjectHandler using a Function<S3InputFile, InputCodec> should work. I gave some ideas on this below.
| * @param sample raw bytes from the beginning of the data source | ||
| * @return detection result with compression, format, and confidence | ||
| */ | ||
| public FormatDetectionResult detect(final byte[] sample) { |
There was a problem hiding this comment.
We should use Apache Tika to detect formats.
| final AcknowledgementSet acknowledgementSet, | ||
| final SourceCoordinator<S3SourceProgressState> sourceCoordinator, | ||
| final String partitionKey) throws IOException { | ||
| final S3InputFile inputFile = new S3InputFile( |
There was a problem hiding this comment.
We may want to load the object using two requests. Otherwise you've pushed the stream forward. You could use a buffered input stream.
If you do two requests, you should specify a byte range. This may be better anyway to constrain the request length given.
PR Code Analyzer ❗AI-powered 'Code-Diff-Analyzer' found issues on commit 4d2a1bc. ⛔ Hard block: Issues at High severity or above will block this PR from merging.
The table above displays the top 10 most important findings. Pull Requests Author(s): Please update your Pull Request according to the report above. Repository Maintainer(s): You can Thanks. |
Add a format auto-detection library that identifies data format and compression from S3 object bytes, so customers do not need to specify a codec in their pipeline configuration. - New format-detection module using Apache Tika for MIME/format detection, with lightweight text heuristics to distinguish JSON arrays, ND-JSON, CSV, and TSV. Detects gzip/snappy/zstd compression and decompresses gzip/snappy samples. - S3 source integration via a Function<S3ObjectReference, InputCodec> codec provider. When no codec is configured, the source reads a byte-range sample (bytes=0-65535) per object, detects the format, and selects the matching codec. No forked object worker; the existing S3ObjectWorker resolves the codec through the provider. - Codec is now optional in S3SourceConfig (auto-detect when absent). Includes unit tests covering compression detection, binary/text formats, edge cases (BOM, quoted CSV, truncated JSON, binary garbage), and gzip integration. Compressed test inputs are generated at test time. Signed-off-by: Siqi Ding <dingdd@amazon.com>
a661f78 to
4d2a1bc
Compare
Description
Add a format auto-detection library for Intelligent Ingestion that automatically identifies data format and compression from raw S3 object bytes, eliminating the need for customers to specify a
codecin their pipeline configuration.Key features:
data-prepper-plugins/format-detection/) with minimal dependencies (snappy-java for decompression)S3 source integration:
codecis specified in pipeline config, the S3 source enters auto-detect modeAutoDetectS3ObjectWorkerreads first 64KB of each object, detects format, and selects the appropriate codec dynamicallyAutoDetectCodecFactorymaps detected format to codec plugin (ndjson, json, csv, parquet, avro, newline)DetectionMetricsemits CloudWatch-compatible metrics via Micrometer with combined format+compression names and per-type duration timersPerformance: Average detection time is 1.1ms per file (tested with 5,000 files). Detection overhead is < 0.1% of total pipeline processing time.
Testing
Unit Tests (57 tests):
End-to-End Pipeline Testing (S3 → auto-detect → OpenSearch):
Tested with real S3 bucket using both S3 scan and SQS event-driven modes, with results indexed into an OpenSearch domain.
.ndjson.ndjson.gz.ndjson.snappy.json(array).csv.csv.gz.csv.snappy.tsv.psv(pipe).parquet.pdf.log(plain text)Load Test (5,000 files, ~500MB):
Mixed Format Test (3,600 files — NDJSON + CSV + JSON + compressed variants):
datapreppernamespaceCloudWatch Metrics & Observability
Format detection emits metrics via Micrometer (Prometheus + CloudWatch compatible):
Per format+compression type:
formatDetection.ndjson/formatDetection.ndjson.gzip/formatDetection.csv.snappy— count + duration per typeformatDetection.total— aggregate countformatDetection.failed— unsupported/unknown filesformatDetection.confidence.high/medium/low— confidence breakdownformatDetection.duration— overall detection timerPrometheus endpoint sample (after 5,000 file load test):
Periodic log summary (every 60 seconds during activity, with final summary when ingestion stops):
Issues Resolved
N/A
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.