[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=413338&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-413338 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 31/Mar/20 16:36 Start Date: 31/Mar/20 16:36 Worklog Time Spent: 10m Work Description: boyuanzz commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 413338) Time Spent: 6h 10m (was: 6h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 6h 10m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=412766&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-412766 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 31/Mar/20 01:52 Start Date: 31/Mar/20 01:52 Worklog Time Spent: 10m Work Description: boyuanzz commented on issue #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#issuecomment-606349734 Run Python PreCommit 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 412766) Time Spent: 6h (was: 5h 50m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 6h > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=411873&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-411873 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 29/Mar/20 05:05 Start Date: 29/Mar/20 05:05 Worklog Time Spent: 10m Work Description: boyuanzz commented on issue #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#issuecomment-605559852 Run Python PreCommit 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 411873) Time Spent: 5h 50m (was: 5h 40m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 5h 50m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=411426&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-411426
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 27/Mar/20 22:06
Start Date: 27/Mar/20 22:06
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r399559906
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,133 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+import typing
+
+from apache_beam import typehints
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import core
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.utils import timestamp
+
+__all__ = [
+'Deduplicate',
+'DeduplicatePerKey',
+]
+
+K = typing.TypeVar('K')
+V = typing.TypeVar('V')
+
+
+@typehints.with_input_types(typing.Tuple[K, V])
+@typehints.with_output_types(typing.Tuple[K, V])
+class DeduplicatePerKey(ptransform.PTransform):
+ """ A PTransform which deduplicates pair over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(self, processing_time_duration=None, event_time_duration=None):
+if processing_time_duration is None and event_time_duration is None:
+ raise ValueError(
+ 'DeduplicatePerKey requires at lease provide either'
+ 'processing_time_duration or event_time_duration.')
+self.processing_time_duration = processing_time_duration
+self.event_time_duration = event_time_duration
+
+ def _create_deduplicate_fn(self):
+processing_timer_spec = userstate.TimerSpec(
+'processing_timer', TimeDomain.REAL_TIME)
+event_timer_spec = userstate.TimerSpec('event_timer', TimeDomain.WATERMARK)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
Review comment:
To follow up from what we discussed off-line, I think combining state
semantically makes slightly more sense here, but as we're not adding to the
state unless it's the first element there's no performance concerns, and I'm
also fine with this as is.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 411426)
Time Spent: 5h 40m (was: 5.5h)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 5h 40m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message wa
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=408298&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408298
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 23/Mar/20 20:50
Start Date: 23/Mar/20 20:50
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r396747769
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,133 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+import typing
+
+from apache_beam import typehints
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import core
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.utils import timestamp
+
+__all__ = [
+'Deduplicate',
+'DeduplicatePerKey',
+]
+
+K = typing.TypeVar('K')
+V = typing.TypeVar('V')
+
+
+@typehints.with_input_types(typing.Tuple[K, V])
+@typehints.with_output_types(typing.Tuple[K, V])
+class DeduplicatePerKey(ptransform.PTransform):
+ """ A PTransform which deduplicates pair over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(self, processing_time_duration=None, event_time_duration=None):
+if processing_time_duration is None and event_time_duration is None:
+ raise ValueError(
+ 'DeduplicatePerKey requires at lease provide either'
+ 'processing_time_duration or event_time_duration.')
+self.processing_time_duration = processing_time_duration
+self.event_time_duration = event_time_duration
+
+ def _create_deduplicate_fn(self):
+processing_timer_spec = userstate.TimerSpec(
+'processing_timer', TimeDomain.REAL_TIME)
+event_timer_spec = userstate.TimerSpec('event_timer', TimeDomain.WATERMARK)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
Review comment:
The `seen_state` is only set once per key during that duration. I'm not sure
whether combining state is more suitable here. What do you think?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408298)
Time Spent: 5.5h (was: 5h 20m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 5.5h
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=408286&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408286
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 23/Mar/20 20:38
Start Date: 23/Mar/20 20:38
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r396742097
##
File path: sdks/python/apache_beam/transforms/deduplicate_test.py
##
@@ -0,0 +1,168 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""Unit tests for deduplicate transform by using TestStream."""
+
+from __future__ import absolute_import
+
+import unittest
+
+from nose.plugins.attrib import attr
+
+import apache_beam as beam
+from apache_beam.coders import coders
+from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.testing.test_stream import TestStream
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+from apache_beam.testing.util import equal_to_per_window
+from apache_beam.transforms import deduplicate
+from apache_beam.transforms import window
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import Timestamp
+
+
+# TestStream is only supported in streaming pipeline. The Deduplicate transform
+# also requires Timer support. Sickbaying this testsuite until dataflow runner
+# supports both TestStream and user timer.
+@attr('ValidatesRunner', 'sickbay-batch', 'sickbay-streaming')
Review comment:
The 'sickbay-batch' and 'sickbay-streaming' is only used by dataflow suite
now. And unfortunately, I don't we have runners supporting these python test
now.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408286)
Time Spent: 5h 20m (was: 5h 10m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 5h 20m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=408148&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408148
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 23/Mar/20 18:50
Start Date: 23/Mar/20 18:50
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r396677254
##
File path: sdks/python/apache_beam/transforms/deduplicate_test.py
##
@@ -0,0 +1,168 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""Unit tests for deduplicate transform by using TestStream."""
+
+from __future__ import absolute_import
+
+import unittest
+
+from nose.plugins.attrib import attr
+
+import apache_beam as beam
+from apache_beam.coders import coders
+from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.testing.test_stream import TestStream
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+from apache_beam.testing.util import equal_to_per_window
+from apache_beam.transforms import deduplicate
+from apache_beam.transforms import window
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import Timestamp
+
+
+# TestStream is only supported in streaming pipeline. The Deduplicate transform
+# also requires Timer support. Sickbaying this testsuite until dataflow runner
+# supports both TestStream and user timer.
+@attr('ValidatesRunner', 'sickbay-batch', 'sickbay-streaming')
Review comment:
Is it not possible to sickbay it for only dataflow?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408148)
Time Spent: 5h (was: 4h 50m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 5h
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=408147&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408147
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 23/Mar/20 18:50
Start Date: 23/Mar/20 18:50
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r396675954
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,133 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+import typing
+
+from apache_beam import typehints
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import core
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.utils import timestamp
+
+__all__ = [
+'Deduplicate',
+'DeduplicatePerKey',
+]
+
+K = typing.TypeVar('K')
+V = typing.TypeVar('V')
+
+
+@typehints.with_input_types(typing.Tuple[K, V])
+@typehints.with_output_types(typing.Tuple[K, V])
+class DeduplicatePerKey(ptransform.PTransform):
+ """ A PTransform which deduplicates pair over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(self, processing_time_duration=None, event_time_duration=None):
+if processing_time_duration is None and event_time_duration is None:
+ raise ValueError(
+ 'DeduplicatePerKey requires at lease provide either'
+ 'processing_time_duration or event_time_duration.')
+self.processing_time_duration = processing_time_duration
+self.event_time_duration = event_time_duration
+
+ def _create_deduplicate_fn(self):
+processing_timer_spec = userstate.TimerSpec(
+'processing_timer', TimeDomain.REAL_TIME)
+event_timer_spec = userstate.TimerSpec('event_timer', TimeDomain.WATERMARK)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
Review comment:
Perhaps this should be a combining state?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408147)
Time Spent: 4h 50m (was: 4h 40m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 4h 50m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=408150&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408150
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 23/Mar/20 18:50
Start Date: 23/Mar/20 18:50
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r396680799
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,133 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+import typing
+
+from apache_beam import typehints
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import core
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.utils import timestamp
+
+__all__ = [
+'Deduplicate',
+'DeduplicatePerKey',
+]
+
+K = typing.TypeVar('K')
+V = typing.TypeVar('V')
+
+
+@typehints.with_input_types(typing.Tuple[K, V])
+@typehints.with_output_types(typing.Tuple[K, V])
+class DeduplicatePerKey(ptransform.PTransform):
+ """ A PTransform which deduplicates pair over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
Review comment:
It's not really best effort, rather it is only respected within the time
domains (and there you can count on it).
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408150)
Time Spent: 5h 10m (was: 5h)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 5h 10m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=408149&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408149
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 23/Mar/20 18:50
Start Date: 23/Mar/20 18:50
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r396680622
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,133 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+import typing
+
+from apache_beam import typehints
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import core
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.utils import timestamp
+
+__all__ = [
+'Deduplicate',
+'DeduplicatePerKey',
+]
+
+K = typing.TypeVar('K')
+V = typing.TypeVar('V')
+
+
+@typehints.with_input_types(typing.Tuple[K, V])
+@typehints.with_output_types(typing.Tuple[K, V])
+class DeduplicatePerKey(ptransform.PTransform):
+ """ A PTransform which deduplicates pair over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
Review comment:
I might phrase this as "Time durations are required so as to avoid unbounded
memory and/or storage requirements on the runner..."
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408149)
Time Spent: 5h (was: 4h 50m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 5h
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=406618&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-406618 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 19/Mar/20 21:54 Start Date: 19/Mar/20 21:54 Worklog Time Spent: 10m Work Description: boyuanzz commented on issue #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#issuecomment-601435026 @robertwb Please take another look. Thanks for your help : ) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 406618) Time Spent: 4h 40m (was: 4.5h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 4h 40m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405097&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405097 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 18/Mar/20 01:18 Start Date: 18/Mar/20 01:18 Worklog Time Spent: 10m Work Description: boyuanzz commented on issue #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#issuecomment-600376012 Run Python Dataflow ValidatesRunner 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 405097) Time Spent: 4h 20m (was: 4h 10m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 4h 20m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405098&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405098 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 18/Mar/20 01:18 Start Date: 18/Mar/20 01:18 Worklog Time Spent: 10m Work Description: boyuanzz commented on issue #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#issuecomment-600376033 Run Python 3.5 Flink ValidatesRunner 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 405098) Time Spent: 4.5h (was: 4h 20m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 4.5h > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405056&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405056 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 23:57 Start Date: 17/Mar/20 23:57 Worklog Time Spent: 10m Work Description: boyuanzz commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r394036805 ## File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py ## @@ -65,6 +65,7 @@ from apache_beam.transforms import environments from apache_beam.transforms import userstate from apache_beam.transforms import window +from apache_beam.transforms.deduplicate import DeduplictaionWithinDuration Review comment: Thanks for the pointer! 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 405056) Time Spent: 4h 10m (was: 4h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 4h 10m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405047&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405047 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 23:37 Start Date: 17/Mar/20 23:37 Worklog Time Spent: 10m Work Description: boyuanzz commented on issue #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#issuecomment-600350868 Run Python 3.5 Flink ValidatesRunner 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 405047) Time Spent: 4h (was: 3h 50m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 4h > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405046&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405046 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 23:36 Start Date: 17/Mar/20 23:36 Worklog Time Spent: 10m Work Description: boyuanzz commented on issue #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#issuecomment-600350776 Run Python Dataflow ValidatesRunner 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 405046) Time Spent: 3h 50m (was: 3h 40m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 3h 50m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405045&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405045
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 17/Mar/20 23:35
Start Date: 17/Mar/20 23:35
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r394029968
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,108 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.core import DoFn
+from apache_beam.transforms.core import Map
+from apache_beam.transforms.core import ParDo
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.typehints import trivial_inference
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import Timestamp
+
+__all__ = [
+'DeduplictaionWithinDuration',
+]
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicates input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self, time_domain=TimeDomain.REAL_TIME, duration=Duration(10 * 60)):
+if time_domain not in (TimeDomain.WATERMARK, TimeDomain.REAL_TIME):
+ raise ValueError(
+ 'Unsupported TimeDomain: %r. DeduplictaionWithinDuration'
+ 'expects TimeDomain.WATERMARK or TimeDomain.REAL_TIME' %
+ (time_domain, ))
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True in seen_state.read():
+ return
+
+if domain == TimeDomain.REAL_TIME:
+ expiry_timer.set(Timestamp.now() + duration)
+elif domain == TimeDomain.WATERMARK:
+ expiry_timer.set(ts + duration)
Review comment:
Are you talking about the watermark case? yes the timer will be fires when
the watermark advance to 11 if the timer is set by the value with timestamp 1
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 405045)
Time Spent: 3h 40m (was: 3.5h)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Componen
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405040&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405040
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 17/Mar/20 23:25
Start Date: 17/Mar/20 23:25
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r394026985
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,108 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.core import DoFn
+from apache_beam.transforms.core import Map
+from apache_beam.transforms.core import ParDo
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.typehints import trivial_inference
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import Timestamp
+
+__all__ = [
+'DeduplictaionWithinDuration',
+]
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicates input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self, time_domain=TimeDomain.REAL_TIME, duration=Duration(10 * 60)):
+if time_domain not in (TimeDomain.WATERMARK, TimeDomain.REAL_TIME):
+ raise ValueError(
+ 'Unsupported TimeDomain: %r. DeduplictaionWithinDuration'
+ 'expects TimeDomain.WATERMARK or TimeDomain.REAL_TIME' %
+ (time_domain, ))
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True in seen_state.read():
+ return
+
+if domain == TimeDomain.REAL_TIME:
+ expiry_timer.set(Timestamp.now() + duration)
+elif domain == TimeDomain.WATERMARK:
+ expiry_timer.set(ts + duration)
+else:
+ raise ValueError(
+ 'Unsupported TimeDomain: %r. DeduplictaionWithinDuration'
+ 'expects TimeDomain.WATERMARK or TimeDomain.REAL_TIME' %
+ (domain, ))
+seen_state.add(True)
+value, _ = element
+yield value
+
+ def infer_output_type(self, input_type):
+key_type, _ = trivial_inference.key_value_types(input_type)
+return key_type
+
+ @userstate.on_timer(timer_spec)
+ def process_timer(self, seen_state=DoFn.StateParam(state_spec)):
+seen_state.clear()
+
+return DeduplicationFn()
+
+ def expand(self, pcoll):
+return (
+pcoll
+| 'KeyByElement' >> Map(lambda x: (x, None))
+| 'DeduplicateFn' >> ParDo(self._create_deduplicate_fn()))
Review comment:
Are you suggesting that we provide both `DeduplicatePerKey`
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=405038&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-405038 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 23:21 Start Date: 17/Mar/20 23:21 Worklog Time Spent: 10m Work Description: boyuanzz commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r394025988 ## File path: sdks/python/apache_beam/transforms/deduplicate.py ## @@ -0,0 +1,108 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# pytype: skip-file + +"""a collection of ptransforms for deduplicating elements.""" + +from __future__ import absolute_import +from __future__ import division + +from apache_beam.coders.coders import BooleanCoder +from apache_beam.transforms import ptransform +from apache_beam.transforms import userstate +from apache_beam.transforms.core import DoFn +from apache_beam.transforms.core import Map +from apache_beam.transforms.core import ParDo +from apache_beam.transforms.timeutil import TimeDomain +from apache_beam.typehints import trivial_inference +from apache_beam.utils.timestamp import Duration +from apache_beam.utils.timestamp import Timestamp + +__all__ = [ +'DeduplictaionWithinDuration', +] + + +class DeduplictaionWithinDuration(ptransform.PTransform): + """ A PTransform which deduplicates input records over a time domain and + threshold. Values in different windows will NOT be considered duplicates of + each other. Deduplication is best effort. + + The durations specified may impose memory and/or storage requirements within + a runner and care might need to be used to ensure that the deduplication time + limit is long enough to remove duplicates but short enough to not cause + performance problems within a runner. Each runner may provide an optimized + implementation of their choice using the deduplication time domain and + threshold specified. + + Does not preserve any order the input PCollection might have had. + """ + def __init__( + self, time_domain=TimeDomain.REAL_TIME, duration=Duration(10 * 60)): Review comment: > Also, is it not possible to have both a real and watermark delay? This API forces you to do one or the other. Currently the API only allows either watermark based or walltime based. I'm not sure what kind of strategy we can do if allowing both. Do we want to do no matter which fires first, just clear another? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 405038) Time Spent: 3h 20m (was: 3h 10m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 3h 20m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404984&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404984 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 21:25 Start Date: 17/Mar/20 21:25 Worklog Time Spent: 10m Work Description: robertwb commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r393972784 ## File path: sdks/python/apache_beam/transforms/deduplicate.py ## @@ -0,0 +1,108 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# pytype: skip-file + +"""a collection of ptransforms for deduplicating elements.""" + +from __future__ import absolute_import +from __future__ import division + +from apache_beam.coders.coders import BooleanCoder +from apache_beam.transforms import ptransform +from apache_beam.transforms import userstate +from apache_beam.transforms.core import DoFn +from apache_beam.transforms.core import Map +from apache_beam.transforms.core import ParDo +from apache_beam.transforms.timeutil import TimeDomain +from apache_beam.typehints import trivial_inference +from apache_beam.utils.timestamp import Duration +from apache_beam.utils.timestamp import Timestamp + +__all__ = [ +'DeduplictaionWithinDuration', +] + + +class DeduplictaionWithinDuration(ptransform.PTransform): Review comment: PTransform names are generally verbs. Perhaps just call this `Deduplicate`. The duration argument would typically be part of the constructor, e.g. `Deduplicate(duration=...)`, so wouldn't need to be repeated in the name. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 404984) Time Spent: 2h 50m (was: 2h 40m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 2h 50m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404986&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404986
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 17/Mar/20 21:25
Start Date: 17/Mar/20 21:25
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r393978547
##
File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py
##
@@ -448,6 +449,42 @@ def is_buffered_correctly(actual):
assert_that(actual, is_buffered_correctly)
+ def test_deduplication_transform_with_processing_time(self):
+# Note that current FnApiRunner doesn't respect either real timestamp.
+with self.create_pipeline() as p:
+ inputs = [
+ window.TimestampedValue('value_1', 1),
+ window.TimestampedValue('value_1', 2),
+ window.TimestampedValue('value_1', 3)
+ ]
+ actual = (
+ p
+ | beam.Create(inputs)
+ | beam.WindowInto(window.FixedWindows(10))
+ | DeduplictaionWithinDuration(duration=timestamp.Duration(5)))
+ assert_that(actual, equal_to(['value_1']))
+
+ @unittest.skip('TestStream not yet supported')
+ def test_deduplication_transform_with_event_time(self):
+test_stream = (
+TestStream().advance_watermark_to(0).add_elements([
+window.TimestampedValue('value_1', 1),
+window.TimestampedValue('value_1', 2),
+window.TimestampedValue('value_1', 10)
Review comment:
It's be a stronger test to add elements one at a time.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 404986)
Time Spent: 3h 10m (was: 3h)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 3h 10m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404981&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404981 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 21:25 Start Date: 17/Mar/20 21:25 Worklog Time Spent: 10m Work Description: robertwb commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r393968846 ## File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py ## @@ -65,6 +65,7 @@ from apache_beam.transforms import environments from apache_beam.transforms import userstate from apache_beam.transforms import window +from apache_beam.transforms.deduplicate import DeduplictaionWithinDuration Review comment: Generally we've been following the convention of importing modules rather than objects from modules. http://google.github.io/styleguide/pyguide.html#22-imports 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 404981) Time Spent: 2h 40m (was: 2.5h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 2h 40m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404980&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404980 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 21:25 Start Date: 17/Mar/20 21:25 Worklog Time Spent: 10m Work Description: robertwb commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r393968339 ## File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py ## @@ -448,6 +449,42 @@ def is_buffered_correctly(actual): assert_that(actual, is_buffered_correctly) + def test_deduplication_transform_with_processing_time(self): Review comment: fn_api_runner_test primarily for testing primitives. I would put this in apache_beam.transforms.deduplicate_test (or util_test, if that's where we plan to move it). 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 404980) Time Spent: 2.5h (was: 2h 20m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 2.5h > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404982&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404982
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 17/Mar/20 21:25
Start Date: 17/Mar/20 21:25
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r393970454
##
File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py
##
@@ -448,6 +449,42 @@ def is_buffered_correctly(actual):
assert_that(actual, is_buffered_correctly)
+ def test_deduplication_transform_with_processing_time(self):
+# Note that current FnApiRunner doesn't respect either real timestamp.
+with self.create_pipeline() as p:
+ inputs = [
+ window.TimestampedValue('value_1', 1),
+ window.TimestampedValue('value_1', 2),
+ window.TimestampedValue('value_1', 3)
+ ]
+ actual = (
+ p
+ | beam.Create(inputs)
+ | beam.WindowInto(window.FixedWindows(10))
+ | DeduplictaionWithinDuration(duration=timestamp.Duration(5)))
+ assert_that(actual, equal_to(['value_1']))
+
+ @unittest.skip('TestStream not yet supported')
+ def test_deduplication_transform_with_event_time(self):
+test_stream = (
+TestStream().advance_watermark_to(0).add_elements([
+window.TimestampedValue('value_1', 1),
+window.TimestampedValue('value_1', 2),
+window.TimestampedValue('value_1', 10)
+]).advance_watermark_to(10).add_elements([
+window.TimestampedValue('value_1', 6)
+]).add_elements([window.TimestampedValue('value_2', 15)
+ ]).advance_watermark_to_infinity())
+
+with self.create_pipeline() as p:
+ actual = (
+ p
+ | test_stream
+ | DeduplictaionWithinDuration(
+ duration=timestamp.Duration(10),
+ time_domain=userstate.TimeDomain.WATERMARK))
+ assert_that(actual, equal_to(['value_1', 'value_1', 'value_2']))
Review comment:
The duplication of value_1 probably merits a comment.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 404982)
Time Spent: 2h 40m (was: 2.5h)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 2h 40m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404985&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404985
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 17/Mar/20 21:25
Start Date: 17/Mar/20 21:25
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r393976398
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,108 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.core import DoFn
+from apache_beam.transforms.core import Map
+from apache_beam.transforms.core import ParDo
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.typehints import trivial_inference
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import Timestamp
+
+__all__ = [
+'DeduplictaionWithinDuration',
+]
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicates input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self, time_domain=TimeDomain.REAL_TIME, duration=Duration(10 * 60)):
+if time_domain not in (TimeDomain.WATERMARK, TimeDomain.REAL_TIME):
+ raise ValueError(
+ 'Unsupported TimeDomain: %r. DeduplictaionWithinDuration'
+ 'expects TimeDomain.WATERMARK or TimeDomain.REAL_TIME' %
+ (time_domain, ))
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True in seen_state.read():
+ return
+
+if domain == TimeDomain.REAL_TIME:
+ expiry_timer.set(Timestamp.now() + duration)
+elif domain == TimeDomain.WATERMARK:
+ expiry_timer.set(ts + duration)
+else:
+ raise ValueError(
+ 'Unsupported TimeDomain: %r. DeduplictaionWithinDuration'
+ 'expects TimeDomain.WATERMARK or TimeDomain.REAL_TIME' %
+ (domain, ))
+seen_state.add(True)
+value, _ = element
+yield value
+
+ def infer_output_type(self, input_type):
+key_type, _ = trivial_inference.key_value_types(input_type)
+return key_type
+
+ @userstate.on_timer(timer_spec)
+ def process_timer(self, seen_state=DoFn.StateParam(state_spec)):
+seen_state.clear()
+
+return DeduplicationFn()
+
+ def expand(self, pcoll):
+return (
+pcoll
+| 'KeyByElement' >> Map(lambda x: (x, None))
+| 'DeduplicateFn' >> ParDo(self._create_deduplicate_fn()))
Review comment:
I would expose a DeduplicatePerKey transform which emits a
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404983&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404983 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 21:25 Start Date: 17/Mar/20 21:25 Worklog Time Spent: 10m Work Description: robertwb commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r393969469 ## File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py ## @@ -448,6 +449,42 @@ def is_buffered_correctly(actual): assert_that(actual, is_buffered_correctly) + def test_deduplication_transform_with_processing_time(self): +# Note that current FnApiRunner doesn't respect either real timestamp. Review comment: IIRC, the FnApiRunner does respect timestamps. (Not sure about the Create transform though.) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 404983) Time Spent: 2h 50m (was: 2h 40m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 2h 50m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404987&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404987
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 17/Mar/20 21:25
Start Date: 17/Mar/20 21:25
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r393978328
##
File path: sdks/python/apache_beam/transforms/deduplicate.py
##
@@ -0,0 +1,108 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+"""a collection of ptransforms for deduplicating elements."""
+
+from __future__ import absolute_import
+from __future__ import division
+
+from apache_beam.coders.coders import BooleanCoder
+from apache_beam.transforms import ptransform
+from apache_beam.transforms import userstate
+from apache_beam.transforms.core import DoFn
+from apache_beam.transforms.core import Map
+from apache_beam.transforms.core import ParDo
+from apache_beam.transforms.timeutil import TimeDomain
+from apache_beam.typehints import trivial_inference
+from apache_beam.utils.timestamp import Duration
+from apache_beam.utils.timestamp import Timestamp
+
+__all__ = [
+'DeduplictaionWithinDuration',
+]
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicates input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self, time_domain=TimeDomain.REAL_TIME, duration=Duration(10 * 60)):
+if time_domain not in (TimeDomain.WATERMARK, TimeDomain.REAL_TIME):
+ raise ValueError(
+ 'Unsupported TimeDomain: %r. DeduplictaionWithinDuration'
+ 'expects TimeDomain.WATERMARK or TimeDomain.REAL_TIME' %
+ (time_domain, ))
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True in seen_state.read():
+ return
+
+if domain == TimeDomain.REAL_TIME:
+ expiry_timer.set(Timestamp.now() + duration)
+elif domain == TimeDomain.WATERMARK:
+ expiry_timer.set(ts + duration)
Review comment:
Suppose one has values with timestamps 10, 1, 12, and duration 10. This will
not properly deduplicate because the timer will fire at 11 clearing state
before 12 gets seen.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 404987)
Time Spent: 3h 10m (was: 3h)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=404988&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-404988 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 17/Mar/20 21:25 Start Date: 17/Mar/20 21:25 Worklog Time Spent: 10m Work Description: robertwb commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r393974676 ## File path: sdks/python/apache_beam/transforms/deduplicate.py ## @@ -0,0 +1,108 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# pytype: skip-file + +"""a collection of ptransforms for deduplicating elements.""" + +from __future__ import absolute_import +from __future__ import division + +from apache_beam.coders.coders import BooleanCoder +from apache_beam.transforms import ptransform +from apache_beam.transforms import userstate +from apache_beam.transforms.core import DoFn +from apache_beam.transforms.core import Map +from apache_beam.transforms.core import ParDo +from apache_beam.transforms.timeutil import TimeDomain +from apache_beam.typehints import trivial_inference +from apache_beam.utils.timestamp import Duration +from apache_beam.utils.timestamp import Timestamp + +__all__ = [ +'DeduplictaionWithinDuration', +] + + +class DeduplictaionWithinDuration(ptransform.PTransform): + """ A PTransform which deduplicates input records over a time domain and + threshold. Values in different windows will NOT be considered duplicates of + each other. Deduplication is best effort. + + The durations specified may impose memory and/or storage requirements within + a runner and care might need to be used to ensure that the deduplication time + limit is long enough to remove duplicates but short enough to not cause + performance problems within a runner. Each runner may provide an optimized + implementation of their choice using the deduplication time domain and + threshold specified. + + Does not preserve any order the input PCollection might have had. + """ + def __init__( + self, time_domain=TimeDomain.REAL_TIME, duration=Duration(10 * 60)): Review comment: I would make duration a required parameter; I don't think there's a good default we can pick for all pipelines. (This also allows us to have a more intelligent default in the future, e.g. dynamically based on the actual watermark/propagation delay in the pipeline.) Also, is it not possible to have both a real and watermark delay? This API forces you to do one or the other. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 404988) Time Spent: 3h 10m (was: 3h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 3h 10m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=403211&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-403211 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 13/Mar/20 22:34 Start Date: 13/Mar/20 22:34 Worklog Time Spent: 10m Work Description: udim commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r392514885 ## File path: sdks/python/apache_beam/runners/sdf_utils.py ## @@ -244,3 +251,63 @@ def get_estimator_state(self): return None return _NoOpWatermarkEstimator() + + +class DeduplictaionWithinDuration(ptransform.PTransform): Review comment: No, no reason. Go ahead with using a separate file 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 403211) Time Spent: 2h 20m (was: 2h 10m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 2h 20m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=403162&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-403162 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 13/Mar/20 20:40 Start Date: 13/Mar/20 20:40 Worklog Time Spent: 10m Work Description: boyuanzz commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r392461155 ## File path: sdks/python/apache_beam/runners/sdf_utils.py ## @@ -244,3 +251,63 @@ def get_estimator_state(self): return None return _NoOpWatermarkEstimator() + + +class DeduplictaionWithinDuration(ptransform.PTransform): Review comment: Thanks Udi! Similar to `watermark_estimators.py`, I'm thinking about providing a collection of transforms for deduplicating purpose. Any reason that we don't put a group of similar transforms into one module? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 403162) Time Spent: 2h 10m (was: 2h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 2h 10m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=403030&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-403030 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 13/Mar/20 17:58 Start Date: 13/Mar/20 17:58 Worklog Time Spent: 10m Work Description: udim commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r392386221 ## File path: sdks/python/apache_beam/runners/sdf_utils.py ## @@ -244,3 +251,63 @@ def get_estimator_state(self): return None return _NoOpWatermarkEstimator() + + +class DeduplictaionWithinDuration(ptransform.PTransform): Review comment: Unlike Java, we typically don't put transforms in their own module in Python. I would recommend putting it under an existing file such as transforms/util.py. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 403030) Time Spent: 2h (was: 1h 50m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 2h > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402668&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402668
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 13/Mar/20 05:36
Start Date: 13/Mar/20 05:36
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r392038085
##
File path: sdks/python/apache_beam/runners/sdf_utils.py
##
@@ -244,3 +251,63 @@ def get_estimator_state(self):
return None
return _NoOpWatermarkEstimator()
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicate input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self,
+ time_domain=userstate.TimeDomain.REAL_TIME,
+ duration=Duration(10 * 60)):
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True not in seen_state.read():
+ if domain is userstate.TimeDomain.REAL_TIME:
Review comment:
When writing this part, I think we should update user-face timer API in
python to do similar thing as Java, which makes life easier a lot.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 402668)
Time Spent: 1h 50m (was: 1h 40m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 1h 50m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402667&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402667 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 13/Mar/20 05:34 Start Date: 13/Mar/20 05:34 Worklog Time Spent: 10m Work Description: boyuanzz commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r392037695 ## File path: sdks/python/apache_beam/runners/sdf_utils.py ## @@ -244,3 +251,63 @@ def get_estimator_state(self): return None return _NoOpWatermarkEstimator() + + +class DeduplictaionWithinDuration(ptransform.PTransform): Review comment: Moved transform to `sdks/python/apache_beam/transforms/deduplicate.py` 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 402667) Time Spent: 1h 40m (was: 1.5h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 1h 40m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402665&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402665
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 13/Mar/20 05:33
Start Date: 13/Mar/20 05:33
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r392037568
##
File path: sdks/python/apache_beam/runners/sdf_utils.py
##
@@ -244,3 +251,63 @@ def get_estimator_state(self):
return None
return _NoOpWatermarkEstimator()
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicate input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self,
+ time_domain=userstate.TimeDomain.REAL_TIME,
+ duration=Duration(10 * 60)):
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True not in seen_state.read():
+ if domain is userstate.TimeDomain.REAL_TIME:
Review comment:
Python timer only supports `REAL_TIME ` and `WATERMARK `. I changed it to
checking the type of `TimeDomain` when constructing
`DeduplictaionWithinDuration`.
The type is string so it's not switchable, and python doesn't have switch.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 402665)
Time Spent: 1h 20m (was: 1h 10m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 1h 20m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402666&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402666
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 13/Mar/20 05:33
Start Date: 13/Mar/20 05:33
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r392037597
##
File path: sdks/python/apache_beam/runners/sdf_utils.py
##
@@ -244,3 +251,63 @@ def get_estimator_state(self):
return None
return _NoOpWatermarkEstimator()
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicate input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self,
+ time_domain=userstate.TimeDomain.REAL_TIME,
+ duration=Duration(10 * 60)):
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True not in seen_state.read():
+ if domain is userstate.TimeDomain.REAL_TIME:
+expiry_timer.set(Timestamp.now() + duration)
+ else:
+expiry_timer.set(ts + duration)
+ seen_state.add(True)
+ value, _ = element
+ yield value
Review comment:
Done.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 402666)
Time Spent: 1.5h (was: 1h 20m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402564&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402564 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 12/Mar/20 23:49 Start Date: 12/Mar/20 23:49 Worklog Time Spent: 10m Work Description: boyuanzz commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r391962688 ## File path: sdks/python/apache_beam/runners/sdf_utils.py ## @@ -244,3 +251,63 @@ def get_estimator_state(self): return None return _NoOpWatermarkEstimator() + + +class DeduplictaionWithinDuration(ptransform.PTransform): Review comment: I would prefer the second one: `sdks/python/apache_beam/transforms/deduplicate.py` 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 402564) Time Spent: 1h (was: 50m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 1h > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402565&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402565
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 12/Mar/20 23:49
Start Date: 12/Mar/20 23:49
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r391962780
##
File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py
##
@@ -1250,6 +1313,9 @@ def test_sdf_with_sdf_initiated_checkpointing(self):
def test_sdf_with_watermark_tracking(self):
raise unittest.SkipTest("This test is for a single worker only.")
+ def test_deduplication_transform_with_processing_time(self):
+raise unittest.SkipTest("This test is for a single worker only.")
Review comment:
No, I don't think so.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 402565)
Time Spent: 1h 10m (was: 1h)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 1h 10m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402297&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402297 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 12/Mar/20 16:28 Start Date: 12/Mar/20 16:28 Worklog Time Spent: 10m Work Description: lukecwik commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r391740608 ## File path: sdks/python/apache_beam/runners/sdf_utils.py ## @@ -244,3 +251,63 @@ def get_estimator_state(self): return None return _NoOpWatermarkEstimator() + + +class DeduplictaionWithinDuration(ptransform.PTransform): + """ A PTransform which deduplicate input records over a time domain and Review comment: ```suggestion """ A PTransform which deduplicates input records over a time domain and ``` 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 402297) Time Spent: 40m (was: 0.5h) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 40m > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[ https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402296&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402296 ] ASF GitHub Bot logged work on BEAM-9454: Author: ASF GitHub Bot Created on: 12/Mar/20 16:28 Start Date: 12/Mar/20 16:28 Worklog Time Spent: 10m Work Description: lukecwik commented on pull request #11060: [BEAM-9454] Create Deduplication transform based on user timer/state URL: https://github.com/apache/beam/pull/11060#discussion_r391728602 ## File path: sdks/python/apache_beam/runners/sdf_utils.py ## @@ -244,3 +251,63 @@ def get_estimator_state(self): return None return _NoOpWatermarkEstimator() + + +class DeduplictaionWithinDuration(ptransform.PTransform): Review comment: It would be useful to expose a keyed deduplication transform as the common implementation that all use internally so in the future we can turn into a well known URN and then runners could provide optimized deduplication transform implementations. We want pipeline authors to use this transform and I think it should go into sdks/python/apache_beam/transforms/util.py or into a dedicated file such as sdks/python/apache_beam/transforms/ such as deduplicate.py. CC: @udim What do you think? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 402296) Time Spent: 0.5h (was: 20m) > Add Deduplication transform for SDF > --- > > Key: BEAM-9454 > URL: https://issues.apache.org/jira/browse/BEAM-9454 > Project: Beam > Issue Type: New Feature > Components: sdk-java-core, sdk-py-core >Reporter: Boyuan Zhang >Priority: Major > Time Spent: 0.5h > Remaining Estimate: 0h > > When SDF is used as a source-like operation, it's necessary to provide a > default Deduplication transform for the SDF user to deduplicate values by > certain unique id. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402295&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402295
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 12/Mar/20 16:28
Start Date: 12/Mar/20 16:28
Worklog Time Spent: 10m
Work Description: lukecwik commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r391727529
##
File path: sdks/python/apache_beam/runners/sdf_utils.py
##
@@ -244,3 +251,63 @@ def get_estimator_state(self):
return None
return _NoOpWatermarkEstimator()
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicate input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self,
+ time_domain=userstate.TimeDomain.REAL_TIME,
+ duration=Duration(10 * 60)):
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True not in seen_state.read():
+ if domain is userstate.TimeDomain.REAL_TIME:
+expiry_timer.set(Timestamp.now() + duration)
+ else:
+expiry_timer.set(ts + duration)
+ seen_state.add(True)
+ value, _ = element
+ yield value
Review comment:
I should have used a guard style statement in the Java implementation
instead of nesting.
```suggestion
if True in seen_state.read():
return
if domain is userstate.TimeDomain.REAL_TIME:
expiry_timer.set(Timestamp.now() + duration)
else:
expiry_timer.set(ts + duration)
seen_state.add(True)
value, _ = element
yield value
```
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 402295)
Time Spent: 20m (was: 10m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 20m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402294&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402294
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 12/Mar/20 16:28
Start Date: 12/Mar/20 16:28
Worklog Time Spent: 10m
Work Description: lukecwik commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r391726042
##
File path: sdks/python/apache_beam/runners/sdf_utils.py
##
@@ -244,3 +251,63 @@ def get_estimator_state(self):
return None
return _NoOpWatermarkEstimator()
+
+
+class DeduplictaionWithinDuration(ptransform.PTransform):
+ """ A PTransform which deduplicate input records over a time domain and
+ threshold. Values in different windows will NOT be considered duplicates of
+ each other. Deduplication is best effort.
+
+ The durations specified may impose memory and/or storage requirements within
+ a runner and care might need to be used to ensure that the deduplication time
+ limit is long enough to remove duplicates but short enough to not cause
+ performance problems within a runner. Each runner may provide an optimized
+ implementation of their choice using the deduplication time domain and
+ threshold specified.
+
+ Does not preserve any order the input PCollection might have had.
+ """
+ def __init__(
+ self,
+ time_domain=userstate.TimeDomain.REAL_TIME,
+ duration=Duration(10 * 60)):
+self.time_domain = time_domain
+self.duration = duration
+
+ def _create_deduplicate_fn(self):
+timer_spec = userstate.TimerSpec('expiry_timer', self.time_domain)
+state_spec = userstate.BagStateSpec('seen', BooleanCoder())
+duration = self.duration
+domain = self.time_domain
+
+class DeduplicationFn(DoFn):
+ def process(
+ self,
+ element,
+ ts=DoFn.TimestampParam,
+ seen_state=DoFn.StateParam(state_spec),
+ expiry_timer=DoFn.TimerParam(timer_spec)):
+if True not in seen_state.read():
+ if domain is userstate.TimeDomain.REAL_TIME:
Review comment:
The Python timer API seems primitive compared to the Java SDK
Use a switch case for the three types:
DEPENDENT_REAL_TIME
REAL_TIME
WATERMARK
and throw an error otherwise since you won't know what to do.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 402294)
Remaining Estimate: 0h
Time Spent: 10m
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 10m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-9454) Add Deduplication transform for SDF
[
https://issues.apache.org/jira/browse/BEAM-9454?focusedWorklogId=402298&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-402298
]
ASF GitHub Bot logged work on BEAM-9454:
Author: ASF GitHub Bot
Created on: 12/Mar/20 16:28
Start Date: 12/Mar/20 16:28
Worklog Time Spent: 10m
Work Description: lukecwik commented on pull request #11060: [BEAM-9454]
Create Deduplication transform based on user timer/state
URL: https://github.com/apache/beam/pull/11060#discussion_r391729852
##
File path: sdks/python/apache_beam/runners/portability/fn_api_runner_test.py
##
@@ -1250,6 +1313,9 @@ def test_sdf_with_sdf_initiated_checkpointing(self):
def test_sdf_with_watermark_tracking(self):
raise unittest.SkipTest("This test is for a single worker only.")
+ def test_deduplication_transform_with_processing_time(self):
+raise unittest.SkipTest("This test is for a single worker only.")
Review comment:
Is test stream unsupported in these other setups?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 402298)
Time Spent: 50m (was: 40m)
> Add Deduplication transform for SDF
> ---
>
> Key: BEAM-9454
> URL: https://issues.apache.org/jira/browse/BEAM-9454
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core, sdk-py-core
>Reporter: Boyuan Zhang
>Priority: Major
> Time Spent: 50m
> Remaining Estimate: 0h
>
> When SDF is used as a source-like operation, it's necessary to provide a
> default Deduplication transform for the SDF user to deduplicate values by
> certain unique id.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
