[Feature Request]: Add GroupByEncryptedKey transform

[damccorm]

What would you like to happen?

Add a transform which allows users to use a secret manager to ensure that any GBK data is encrypted before being sent to a runner. This allows users to avoid unencrypted data at rest.

Exact support is still runner dependent (the runner needs to guarantee it won't otherwise materialize Pcollections), but this gives the ability to offer this capability.

Subparts:

• Add GBEK transform in Python - Add GroupByEncryptedKey transform #36213
• Add GBEK transform in Java - Java GroupByEncryptedKey #36217
• Support pipeline option to auto-replace GBEK with encrypted GBEK in Python - Add pipeline option to enforce gbek #36321
• Support pipeline option to auto-replace GBEK with encrypted GBEK in Java - Add pipeline option to force GBEK (Java) #36346
• CombinePerKey can break this workflow since some workflows will lift this to the top level - we should not allow CombineValues transform replacement when using GBEK, which can be done by not using the COMBINE_PER_KEY urn) - CombinePerKey with gbek (Python) #36382 and CombinePerKey with gbek (Java) #36408
• Make sure pipeline option works in x-lang setting (e.g. Java GBK in Python pipeline) - Use consistent encoding for GBEK across languages #36431 , x-lang GroupByEncryptedKey (Java to Python) #36418 , Add some x-lang gbek tests (Python to Java) #36457 , and Fix passing pipeline options to external transforms #36443
• Add gbek option to

  beam/runners/google-cloud-dataflow-java/build.gradle

  Line 179 in 2f9a910

  def runnerV2CommonPipelineOptions = [

  and

  beam/sdks/python/scripts/run_integration_test.sh

  Line 220 in e4c891f

  opts=(

  and run postcommit suites (no need to merge). Set up a similar suite which runs a few integration tests. Led to PRs - Handle null keys in gbek #36505 and Softens the GBEK determinism requirement #36495
• (Dataflow only) Set a service option to tell the service that gbek enforcement is in place - Add use_gbek service option when gbek option used #36452

Future work: Support a more universal pipeline option which runners can opt into/out of for this concept, like #36214 (comment)

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[hjtran]

Just passing by and saw this ticket. Out of curiosity, how are runners supposed to guarantee encryption at rest? Can't a runner arbitrarily decide when to materialize pcollections to disk? In which case, would the runner need to be able to encrypt at any point of the pipeline?

[damccorm]

Good question - I updated the description a bit. Basically, the goal here is to allow runners to add this capability; in order to do so, they'd need to guarantee that they're not going to materialize any PCollections in an undesirable way outside of the context of a GBK.

But using the transform with an arbitrary runner isn't necessarily enough by itself.

[hjtran]

Would it be feasible to implement this capability generically for all transforms using a wrapper secret coder? It'd be a type of BytesCoder that allows specifying a secret. Then runners could use this secret coder and wrap any final bytes coder they'd be using to read/write for the SKD harness and GBK without requiring users to adjust their pipelines to accommodate encryption-at-rest

[damccorm]

That would definitely be an option. I'll note that it is probably less efficient though, since you need to apply it to any encoding/decoding, as opposed to just:

1. Applying it at the GBK boundary
2. Enforcing that you only persist GBK'd data (this is how most runners work/checkpoint already AFAIK).

An alternative would be to eventually offer both and let the runner choose the mode that works for them.

[hjtran]

2. Enforcing that you only persist GBK'd data (this is how most runners work/checkpoint already AFAIK).

The Schrodinger SeamRunner doesn't work this way. Many of the SeamRunner stage boundaries are just from GBKs but there are many stage boundaries that arise from incompatible environments as well. Not sure how other runners handle stage boundaries - do they just pipe together data streams directly between workers?

An alternative would be to eventually offer both and let the runner choose the mode that works for them.

If we had an API that specifies how secrets are determined as a PipelineOption, then only the only-GBK-persisting runners could replace GBKs with GBEKs and other runners could use the secret in all places where they persist pcollections.

I think a main concern here for me is the addition of a new transform that blurs the SDK/runner boundary. When onboarding new developers onto Beam, the biggest hurdle I face is introducing Reshuffle since it breaks the promise of "As a pipeline author, you don't have to worry about the how of execution" and GBEK may be another transform like `Reshuffle.

[damccorm]

Not sure how other runners handle stage boundaries - do they just pipe together data streams directly between workers?

Generally this reduces to a reshuffle or similar which eventually uses GBK under the hood. But definitely not part of the Beam API.

If we had an API that specifies how secrets are determined as a PipelineOption, then only the only-GBK-persisting runners could replace GBKs with GBEKs and other runners could use the secret in all places where they persist pcollections.

Yeah, this is more or less what I'm suggesting here; this issue is more of an implementation detail (though potentially you could only encrypt some of your pipeline at the GBK step if you want to use it directly I guess).

Note that even with this transform and the GBK guarantees, it is still impossible for users to get encryption at rest guarantees since they would need to replace GBK in reshuffles (and other transforms) somehow, and they won't be able to do that. So a pipeline option will be necessary regardless.