[
https://issues.apache.org/jira/browse/BEAM-8335?focusedWorklogId=389630&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-389630
]
ASF GitHub Bot logged work on BEAM-8335:
----------------------------------------
Author: ASF GitHub Bot
Created on: 19/Feb/20 21:12
Start Date: 19/Feb/20 21:12
Worklog Time Spent: 10m
Work Description: aaltay commented on pull request #10899: [BEAM-8335]
Background Caching job
URL: https://github.com/apache/beam/pull/10899#discussion_r381545257
##########
File path: sdks/python/apache_beam/runners/interactive/interactive_beam.py
##########
@@ -34,6 +34,58 @@
from __future__ import absolute_import
from apache_beam.runners.interactive import interactive_environment as ie
+from apache_beam.runners.interactive.options import interactive_options
+
+
+class Options(interactive_options.InteractiveOptions):
+ """Options that guide how Interactive Beam works."""
+ @property
+ def enable_capture_replay(self):
+ """Whether replayable source data capture should be replayed for multiple
+ PCollection evaluations and pipeline runs as long as the data captured is
+ still valid."""
+ return self.capture_control._enable_capture_replay
+
+ @enable_capture_replay.setter
+ def enable_capture_replay(self, value):
+ """Sets whether source data capture should be replayed. True - Enables
+ capture of replayable source data so that following PCollection evaluations
+ and pipeline runs always use the same data captured; False - Disables
+ capture of replayable source data so that following PCollection evaluation
+ and pipeline runs always use new data from sources."""
+ self.capture_control._enable_capture_replay = value
+
+ @property
+ def capturable_sources(self):
+ """Interactive Beam automatically captures data from sources in this
set."""
+ return self.capture_control._capturable_sources
+
+ @property
+ def capture_duration(self):
+ """The data capture of sources ends as soon as the background caching job
+ has run for this long."""
+ return self.capture_control._capture_duration
+
+ @capture_duration.setter
+ def capture_duration(self, value):
+ """Sets the capture duration as a timedelta.
+
+ Example::
+
+ # Sets the capture duration limit to 10 seconds.
+ interactive_beam.options.capture_duration = timedelta(seconds=10)
+ # Evicts all captured data if there is any.
+ interactive_beam.evict_captured_data()
+ # The next PCollection evaluation will capture fresh data from sources,
+ # and the data captured will be replayed until another eviction.
+ """
+ self.capture_control._capture_duration = value
+
+ # TODO(BEAM-8335): add capture_size options when they are supported.
+
+
+# Users can set options to guide how Interactive Beam works.
+options = Options()
Review comment:
How do they set 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: 389630)
Time Spent: 62h 40m (was: 62.5h)
> Add streaming support to Interactive Beam
> -----------------------------------------
>
> Key: BEAM-8335
> URL: https://issues.apache.org/jira/browse/BEAM-8335
> Project: Beam
> Issue Type: Improvement
> Components: runner-py-interactive
> Reporter: Sam Rohde
> Assignee: Sam Rohde
> Priority: Major
> Time Spent: 62h 40m
> Remaining Estimate: 0h
>
> This issue tracks the work items to introduce streaming support to the
> Interactive Beam experience. This will allow users to:
> * Write and run a streaming job in IPython
> * Automatically cache records from unbounded sources
> * Add a replay experience that replays all cached records to simulate the
> original pipeline execution
> * Add controls to play/pause/stop/step individual elements from the cached
> records
> * Add ability to inspect/visualize unbounded PCollections
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
