claudevdm commented on code in PR #39442:
URL: https://github.com/apache/beam/pull/39442#discussion_r3703507039
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -597,6 +597,34 @@ def display_data(self):
def default_label(self):
return 'FixedSizePerKey(%d)' % self._n
+ @with_input_types(T)
+ @with_output_types(T)
+ class Any(ptransform.PTransform):
+ """Returns up to n arbitrary elements from the input PCollection.
+
+ This is the Python equivalent of Java's ``Sample.any``. Unlike
+ ``FixedSizeGlobally`` it does not sample uniformly at random, and it
returns
+ the selected elements rather than a single list. If the input has fewer
than
+ n elements, all of them are returned.
+ """
+ def __init__(self, n):
+ if n < 0:
+ raise ValueError('Expected non-negative n, received %s.' % n)
+ self._n = n
+
+ def expand(self, pcoll):
+ return (
+ pcoll
+ | core.CombineGlobally(_SampleAnyCombineFn(
+ self._n)).without_defaults()
+ | core.FlatMap(lambda elements: elements))
Review Comment:
Can you please add an explicit
.with_input_types(list[T]).with_output_types(T) to FlatMap?
--
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]
