Class Combine
- java.lang.Object
-
- org.apache.beam.sdk.transforms.Combine
-
public class Combine extends java.lang.Object
PTransform
s for combiningPCollection
elements globally and per-key.See the documentation for how to use the operations in this class.
-
-
Nested Class Summary
Nested Classes Modifier and Type Class Description static class
Combine.AccumulatingCombineFn<InputT,AccumT extends Combine.AccumulatingCombineFn.Accumulator<InputT,AccumT,OutputT>,OutputT>
ACombineFn
that uses a subclass ofCombine.AccumulatingCombineFn.Accumulator
as its accumulator type.static class
Combine.BinaryCombineDoubleFn
An abstract subclass ofCombine.CombineFn
for implementing combiners that are more easily and efficiently expressed as binary operations ondouble
s.static class
Combine.BinaryCombineFn<V>
An abstract subclass ofCombine.CombineFn
for implementing combiners that are more easily expressed as binary operations.static class
Combine.BinaryCombineIntegerFn
An abstract subclass ofCombine.CombineFn
for implementing combiners that are more easily and efficiently expressed as binary operations onint
sstatic class
Combine.BinaryCombineLongFn
An abstract subclass ofCombine.CombineFn
for implementing combiners that are more easily and efficiently expressed as binary operations onlong
s.static class
Combine.CombineFn<InputT extends @Nullable java.lang.Object,AccumT extends @Nullable java.lang.Object,OutputT extends @Nullable java.lang.Object>
ACombineFn<InputT, AccumT, OutputT>
specifies how to combine a collection of input values of typeInputT
into a single output value of typeOutputT
.static class
Combine.Globally<InputT,OutputT>
Combine.Globally<InputT, OutputT>
takes aPCollection<InputT>
and returns aPCollection<OutputT>
whose elements are the result of combining all the elements in each window of the inputPCollection
, using a specifiedCombineFn<InputT, AccumT, OutputT>
.static class
Combine.GloballyAsSingletonView<InputT,OutputT>
Combine.GloballyAsSingletonView<InputT, OutputT>
takes aPCollection<InputT>
and returns aPCollectionView<OutputT>
whose elements are the result of combining all the elements in each window of the inputPCollection
, using a specifiedCombineFn<InputT, AccumT, OutputT>
.static class
Combine.GroupedValues<K,InputT,OutputT>
GroupedValues<K, InputT, OutputT>
takes aPCollection<KV<K, Iterable<InputT>>>
, such as the result ofGroupByKey
, applies a specifiedCombineFn<InputT, AccumT, OutputT>
to each of the inputKV<K, Iterable<InputT>>
elements to produce a combined outputKV<K, OutputT>
element, and returns aPCollection<KV<K, OutputT>>
containing all the combined output elements.static class
Combine.Holder<V>
Holds a single value value of typeV
which may or may not be present.static class
Combine.IterableCombineFn<V>
static class
Combine.PerKey<K,InputT,OutputT>
PerKey<K, InputT, OutputT>
takes aPCollection<KV<K, InputT>>
, groups it by key, applies a combining function to theInputT
values associated with each key to produce a combinedOutputT
value, and returns aPCollection<KV<K, OutputT>>
representing a map from each distinct key of the inputPCollection
to the corresponding combined value.static class
Combine.PerKeyWithHotKeyFanout<K,InputT,OutputT>
LikeCombine.PerKey
, but sharding the combining of hot keys.static class
Combine.SimpleCombineFn<V>
Deprecated.
-
Method Summary
All Methods Static Methods Concrete Methods Modifier and Type Method Description static <InputT,OutputT>
Combine.Globally<InputT,OutputT>globally(CombineFnBase.GlobalCombineFn<? super InputT,?,OutputT> fn)
Returns aCombine.Globally
PTransform
that uses the givenGloballyCombineFn
to combine all the elements in each window of the inputPCollection
into a single value in the outputPCollection
.static <V> Combine.Globally<V,V>
globally(SerializableBiFunction<V,V,V> combiner)
Returns aCombine.Globally
PTransform
that uses the givenSerializableBiFunction
to combine all the elements in each window of the inputPCollection
into a single value in the outputPCollection
.static <V> Combine.Globally<V,V>
globally(SerializableFunction<java.lang.Iterable<V>,V> combiner)
Returns aCombine.Globally
PTransform
that uses the givenSerializableFunction
to combine all the elements in each window of the inputPCollection
into a single value in the outputPCollection
.static <K,InputT,OutputT>
Combine.GroupedValues<K,InputT,OutputT>groupedValues(CombineFnBase.GlobalCombineFn<? super InputT,?,OutputT> fn)
Returns aCombine.GroupedValues
PTransform
that takes aPCollection
ofKV
s where a key maps to anIterable
of values, e.g., the result of aGroupByKey
, then uses the givenCombineFn
to combine all the values associated with a key, ignoring the key.static <K,V>
Combine.GroupedValues<K,V,V>groupedValues(SerializableBiFunction<V,V,V> fn)
Returns aCombine.GroupedValues
PTransform
that takes aPCollection
ofKV
s where a key maps to anIterable
of values, e.g., the result of aGroupByKey
, then uses the givenSerializableFunction
to combine all the values associated with a key, ignoring the key.static <K,V>
Combine.GroupedValues<K,V,V>groupedValues(SerializableFunction<java.lang.Iterable<V>,V> fn)
Returns aCombine.GroupedValues
PTransform
that takes aPCollection
ofKV
s where a key maps to anIterable
of values, e.g., the result of aGroupByKey
, then uses the givenSerializableFunction
to combine all the values associated with a key, ignoring the key.static <K,InputT,OutputT>
Combine.PerKey<K,InputT,OutputT>perKey(CombineFnBase.GlobalCombineFn<? super InputT,?,OutputT> fn)
Returns aCombine.PerKey
PTransform
that first groups its inputPCollection
ofKV
s by keys and windows, then invokes the given function on each of the values lists to produce a combined value, and then returns aPCollection
ofKV
s mapping each distinct key to its combined value for each window.static <K,V>
Combine.PerKey<K,V,V>perKey(SerializableBiFunction<V,V,V> fn)
Returns aCombine.PerKey
PTransform
that first groups its inputPCollection
ofKV
s by keys and windows, then invokes the given function on each of the values lists to produce a combined value, and then returns aPCollection
ofKV
s mapping each distinct key to its combined value for each window.static <K,V>
Combine.PerKey<K,V,V>perKey(SerializableFunction<java.lang.Iterable<V>,V> fn)
Returns aCombine.PerKey
PTransform
that first groups its inputPCollection
ofKV
s by keys and windows, then invokes the given function on each of the values lists to produce a combined value, and then returns aPCollection
ofKV
s mapping each distinct key to its combined value for each window.
-
-
-
Method Detail
-
globally
public static <V> Combine.Globally<V,V> globally(SerializableFunction<java.lang.Iterable<V>,V> combiner)
Returns aCombine.Globally
PTransform
that uses the givenSerializableFunction
to combine all the elements in each window of the inputPCollection
into a single value in the outputPCollection
. The types of the input elements and the output elements must be the same.If the input
PCollection
is windowed intoGlobalWindows
, a default value in theGlobalWindow
will be output if the inputPCollection
is empty. To use this with inputs with other windowing, eitherCombine.Globally.withoutDefaults()
orCombine.Globally.asSingletonView()
must be called.See
Combine.Globally
for more information.
-
globally
public static <V> Combine.Globally<V,V> globally(SerializableBiFunction<V,V,V> combiner)
Returns aCombine.Globally
PTransform
that uses the givenSerializableBiFunction
to combine all the elements in each window of the inputPCollection
into a single value in the outputPCollection
. The types of the input elements and the output elements must be the same.If the input
PCollection
is windowed intoGlobalWindows
, a default value in theGlobalWindow
will be output if the inputPCollection
is empty. To use this with inputs with other windowing, eitherCombine.Globally.withoutDefaults()
orCombine.Globally.asSingletonView()
must be called.See
Combine.Globally
for more information.
-
globally
public static <InputT,OutputT> Combine.Globally<InputT,OutputT> globally(CombineFnBase.GlobalCombineFn<? super InputT,?,OutputT> fn)
Returns aCombine.Globally
PTransform
that uses the givenGloballyCombineFn
to combine all the elements in each window of the inputPCollection
into a single value in the outputPCollection
. The types of the input elements and the output elements can differ.If the input
PCollection
is windowed intoGlobalWindows
, a default value in theGlobalWindow
will be output if the inputPCollection
is empty. To use this with inputs with other windowing, eitherCombine.Globally.withoutDefaults()
orCombine.Globally.asSingletonView()
must be called.See
Combine.Globally
for more information.
-
perKey
public static <K,V> Combine.PerKey<K,V,V> perKey(SerializableFunction<java.lang.Iterable<V>,V> fn)
Returns aCombine.PerKey
PTransform
that first groups its inputPCollection
ofKV
s by keys and windows, then invokes the given function on each of the values lists to produce a combined value, and then returns aPCollection
ofKV
s mapping each distinct key to its combined value for each window.Each output element is in the window by which its corresponding input was grouped, and has the timestamp of the end of that window. The output
PCollection
has the sameWindowFn
as the input.See
Combine.PerKey
for more information.
-
perKey
public static <K,V> Combine.PerKey<K,V,V> perKey(SerializableBiFunction<V,V,V> fn)
Returns aCombine.PerKey
PTransform
that first groups its inputPCollection
ofKV
s by keys and windows, then invokes the given function on each of the values lists to produce a combined value, and then returns aPCollection
ofKV
s mapping each distinct key to its combined value for each window.Each output element is in the window by which its corresponding input was grouped, and has the timestamp of the end of that window. The output
PCollection
has the sameWindowFn
as the input.See
Combine.PerKey
for more information.
-
perKey
public static <K,InputT,OutputT> Combine.PerKey<K,InputT,OutputT> perKey(CombineFnBase.GlobalCombineFn<? super InputT,?,OutputT> fn)
Returns aCombine.PerKey
PTransform
that first groups its inputPCollection
ofKV
s by keys and windows, then invokes the given function on each of the values lists to produce a combined value, and then returns aPCollection
ofKV
s mapping each distinct key to its combined value for each window.Each output element is in the window by which its corresponding input was grouped, and has the timestamp of the end of that window. The output
PCollection
has the sameWindowFn
as the input.See
Combine.PerKey
for more information.
-
groupedValues
public static <K,V> Combine.GroupedValues<K,V,V> groupedValues(SerializableFunction<java.lang.Iterable<V>,V> fn)
Returns aCombine.GroupedValues
PTransform
that takes aPCollection
ofKV
s where a key maps to anIterable
of values, e.g., the result of aGroupByKey
, then uses the givenSerializableFunction
to combine all the values associated with a key, ignoring the key. The type of the input and output values must be the same.Each output element has the same timestamp and is in the same window as its corresponding input element, and the output
PCollection
has the sameWindowFn
associated with it as the input.See
Combine.GroupedValues
for more information.Note that
perKey(SerializableFunction)
is typically more convenient to use thanGroupByKey
followed bygroupedValues(...)
.
-
groupedValues
public static <K,V> Combine.GroupedValues<K,V,V> groupedValues(SerializableBiFunction<V,V,V> fn)
Returns aCombine.GroupedValues
PTransform
that takes aPCollection
ofKV
s where a key maps to anIterable
of values, e.g., the result of aGroupByKey
, then uses the givenSerializableFunction
to combine all the values associated with a key, ignoring the key. The type of the input and output values must be the same.Each output element has the same timestamp and is in the same window as its corresponding input element, and the output
PCollection
has the sameWindowFn
associated with it as the input.See
Combine.GroupedValues
for more information.Note that
perKey(SerializableBiFunction)
is typically more convenient to use thanGroupByKey
followed bygroupedValues(...)
.
-
groupedValues
public static <K,InputT,OutputT> Combine.GroupedValues<K,InputT,OutputT> groupedValues(CombineFnBase.GlobalCombineFn<? super InputT,?,OutputT> fn)
Returns aCombine.GroupedValues
PTransform
that takes aPCollection
ofKV
s where a key maps to anIterable
of values, e.g., the result of aGroupByKey
, then uses the givenCombineFn
to combine all the values associated with a key, ignoring the key. The types of the input and output values can differ.Each output element has the same timestamp and is in the same window as its corresponding input element, and the output
PCollection
has the sameWindowFn
associated with it as the input.See
Combine.GroupedValues
for more information.Note that
perKey(CombineFnBase.GlobalCombineFn)
is typically more convenient to use thanGroupByKey
followed bygroupedValues(...)
.
-
-