[
https://issues.apache.org/jira/browse/BEAM-3736?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17124000#comment-17124000
]
Kenneth Knowles commented on BEAM-3736:
---------------------------------------
For context on where CombineFn is invoked, I mostly know pre-portability. Each
runner will have something like the PartialGroupByKey and something like
CombineGroupedValues. Ignore the names. These are "combine before a shuffle"
and "combine after a shuffle", basically.
-
https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PartialGroupByKeyParDoFns.java
-
https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/SimplePartialGroupByKeyParDoFn.java
-
https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/CombineValuesFnFactory.java
-
https://github.com/apache/beam/blob/master/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkPartialReduceFunction.java
-
https://github.com/apache/beam/blob/master/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkReduceFunction.java
You will need to find a way to do this in a safe backwards-compatible way:
1. Add the capability for setup/teardown so it can be detected at construction
time, with all runners detecting and rejecting it (this is very important to do
in one atomic step)
2. Add capability for some runners to and properly invoke setup and teardown
> Add SetUp() and TearDown() for CombineFns
> -----------------------------------------
>
> Key: BEAM-3736
> URL: https://issues.apache.org/jira/browse/BEAM-3736
> Project: Beam
> Issue Type: Improvement
> Components: beam-model, sdk-py-core
> Reporter: Chuan Yu Foo
> Priority: P2
>
> I have a CombineFn that has a large amount of state that needs to be loaded
> once before it can add_input or merge_combiners (for example, the CombineFn
> might load up a large lookup table used for combining).
> Right now, to initialise this state, for each of the methods, I check if the
> state has already been initialised, and if not, I initialise it. It would be
> nice if CombineFn provided a SetUp() method that is called once to initialise
> this state (and a corresponding TearDown() method to clean up this state if
> necessary).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
