robertwb commented on code in PR #25394:
URL: https://github.com/apache/beam/pull/25394#discussion_r1102087462
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1852,3 +1853,46 @@ def visit_transform(self, transform_node):
self.deterministic_key_coders and transform_node.full_label)
return GroupByKeyInputVisitor(deterministic_key_coders)
+
+
+def validate_pipeline_graph(pipeline_proto):
+ """Ensures this is a correctly constructed Beam pipeline.
+ """
+ def get_coder(pcoll_id):
+ return pipeline_proto.components.coders[
+ pipeline_proto.components.pcollections[pcoll_id].coder_id]
+
+ def validate_transform(transform_id):
+ transform_proto = pipeline_proto.components.transforms[transform_id]
+
+ # Currently the only validation we perform is that GBK operations have
+ # their coders set properly.
+ if transform_proto.spec.urn == common_urns.primitives.GROUP_BY_KEY.urn:
Review Comment:
I don't think that's worth doing here, and it's not always even possible
(e.g. cross-langauge). We could consider elevating the notion of determinism to
the model level but that's a larger change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
