Add --transform hook to copy verb for filter/reshape/redact (#93) - #288
Closed
monikagadage wants to merge 1 commit into
Closed
Add --transform hook to copy verb for filter/reshape/redact (#93)#288monikagadage wants to merge 1 commit into
monikagadage wants to merge 1 commit into
Conversation
…loses awslabs#93) The copy verb ships segmented scan -> batch-write between two tables but had no way to touch items in flight (issue awslabs#93). Add an optional --transform argument that dynamically loads a module from a new copy_transform/ package, following the same convention load-export and revert-export already use. Each module exposes transform_item(item), which can return the modified item, None/[] to filter it out, or a list to fan one item into several. Includes two example transforms (pii_redact, attribute_filter), unit coverage for the transform application logic (modify/filter/fan-out/ exception handling) and the example modules, and an e2e smoke test that proves the hook actually runs.
Author
|
Closed accidentally via a branch rename that orphaned the head ref. Replaced by #289. |
monikagadage
added a commit
to monikagadage/amazon-dynamodb-tools
that referenced
this pull request
Sep 3, 2026
…ut (awslabs#93) The copy verb ships a segmented scan -> batch-write between two tables but had no way to touch items in flight (issue awslabs#93). Add an optional --transform argument that dynamically loads a module from copy/transform/, following the same convention load-export and revert-export use. Each module exposes transform_item(item), which returns a list: [item] to write it (modified or not), [] to skip it, or several items to fan out. A bare item is coerced to a one-element list; None is not a skip signal. Design notes / review response: - The transform module is loaded on the driver in run() before the RDD is submitted, so a bad --transform name fails with one BulkExecutorError sentence rather than 400 Spark tasks x 4 retries. The per-worker load stays. - Transform exceptions go through record_worker_failure(understood=False), preserving the worker traceback needed to debug a user transform_item. - Per-worker error accumulation is bounded: the first transform failure and the first post-transform key failure are recorded, then the worker stays quiet, so a broken transform on a huge table cannot flood the driver. - Post-transform key validation: an item that lost its pk/sk in the transform is dropped and reported, not sent into a batch_writer flush that would fail mid-segment. - run() reports "Items excluded by transform: N" and prints a caveat that the up-front cost estimate assumes a 1:1 copy and no longer holds. - copy.py becomes a copy/ package with a transform/ subpackage and a default.py passthrough, matching fill/, load/, update/, scancount/. - transform_loader moves to python_modules/shared/, so the copy path does not import the export pipeline. Includes example transforms (default, pii_redact, attribute_filter), unit coverage for the transform logic (modify / [] skip / fan-out / bounded exception / key validation) and the example modules, and a module_zipper test that proves copy/transform/ and shared/transform_loader.py reach the bootstrap archive. make test: 1751 passed, 48 skipped. Verified end to end against a real Glue job: --transform pii_redact (200 copied), --transform attribute_filter (0 copied, "Items excluded by transform: 200"), a bad --transform name (job fails in ~48s on the driver with "Cannot import transform module ...", before the scan), and a transform that drops the sort key (job fails with "missing key attribute(s) ['sk']", zero rows written to the target). The e2e copy smoke in tests/e2e/commands/test_copy_smoke.py is included but not wired into make test. Supersedes awslabs#288 (closed due to a branch rename that orphaned its head ref).
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
Summary
--transform <name>argument to thecopyverb. It dynamically loads a module from a newcopy_transform/package, following the same conventionload-export/revert-exportalready use for their--transformargument.transform_item(item), which can:Noneor[]to skip the item (not written to the target table)pii_redact(redacts PII attributes) andattribute_filter(only copies items matching an attribute value).Closes #93.
Test plan
make install && make testpasses with no regressions: 1570 passed, 49 skipped.This includes new unit coverage for the transform application logic (modify, filter, fan-out, exception handling) and for both example transform modules.
A new e2e smoke test (
tests/e2e/commands/test_copy_smoke.py) is included but has not been run.