You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by GitBox <gi...@apache.org> on 2020/02/28 22:19:47 UTC
[GitHub] [beam] rohdesamuel opened a new pull request #11005: Streamingcache
merge
rohdesamuel opened a new pull request #11005: Streamingcache merge
URL: https://github.com/apache/beam/pull/11005
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r391156960
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
Review comment:
We've discussed this. Sam will add the fix by re-reading from the initial position.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem merged pull request #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem merged pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387887557
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -212,6 +218,19 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+ if self._evaluation_context._test_stream_event_stub:
+ stub = self._evaluation_context._test_stream_event_stub
+
+ if list(test_stream.output_tags) == [None]:
+ event_request = beam_runner_api_pb2.EventsRequest()
+ else:
+ event_request = beam_runner_api_pb2.EventsRequest(
+ keys=list(test_stream.output_tags))
+
+ test_stream_event_channel = stub.Events(event_request)
Review comment:
Moved to the test_stream_impl 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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388010028
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
+
+ # Check if we are at EOF.
+ if not line:
+ # Complete reading only when the cache is complete.
+ if self._is_cache_complete():
+ break
+
+ if not tail:
+ break
+
+ # Otherwise wait for new data in the file to be written.
+ time.sleep(0.5)
+ fh.seek(pos)
+ else:
+ # The first line at pos = 0 is always the header. Read the line without
+ # the new line.
+ if pos == 0:
+ header = TestStreamFileHeader()
+ header.ParseFromString(self._coder.decode(line[:-1]))
+ yield header
+ else:
+ record = TestStreamFileRecord()
+ record.ParseFromString(self._coder.decode(line[:-1]))
+ yield record
+
+ def read(self, tail):
+ """Reads all TestStreamFile(Header|TestStreamFileRecord)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ with self._wait_until_file_exists() as f:
+ for e in self._emit_from_file(f, tail):
+ yield e
+
+
+class StreamingCache(CacheManager):
"""Abstraction that holds the logic for reading and writing to cache.
"""
- def __init__(self, readers):
- self._readers = readers
+ def __init__(
+ self, cache_dir, is_cache_complete=None, sample_resolution_sec=0.1):
+ self._sample_resolution_sec = sample_resolution_sec
+ self._is_cache_complete = is_cache_complete
+
+ if cache_dir:
+ self._cache_dir = cache_dir
+ else:
+ self._cache_dir = tempfile.mkdtemp(
+ prefix='interactive-temp-', dir=os.environ.get('TEST_TMPDIR', None))
+
+ # List of saved pcoders keyed by PCollection path. It is OK to keep this
+ # list in memory because once FileBasedCacheManager object is
+ # destroyed/re-created it loses the access to previously written cache
+ # objects anyways even if cache_dir already exists. In other words,
+ # it is not possible to resume execution of Beam pipeline from the
+ # saved cache if FileBasedCacheManager has been reset.
+ #
+ # However, if we are to implement better cache persistence, one needs
+ # to take care of keeping consistency between the cached PCollection
+ # and its PCoder type.
+ self._saved_pcoders = {}
+ self._default_pcoder = SafeFastPrimitivesCoder()
+
+ def exists(self, *labels):
+ path = os.path.join(self._cache_dir, *labels)
+ return os.path.exists(path)
+
+ # TODO(srohde): Modify this to return the correct version.
+ def read(self, *labels):
+ """Returns a generator to read all records from file.
+
+ Does not tail.
+ """
+ if not self.exists(*labels):
+ return [].__iter__(), -1
+
+ reader = StreamingCacheSource(
+ self._cache_dir, labels,
+ is_cache_complete=self._is_cache_complete).read(tail=False)
+ header = next(reader)
+ return StreamingCache.Reader([header], [reader]).read(), 1
+
+ def read_multiple(self, labels):
+ """Returns a generator to read all records from file.
+
+ Does tail until the cache is complete. This is because it is used in the
+ TestStreamServiceController to read from file which is only used during
+ pipeline runtime which needs to block.
+ """
+ readers = [
+ StreamingCacheSource(
+ self._cache_dir, l,
+ is_cache_complete=self._is_cache_complete).read(tail=True)
+ for l in labels
+ ]
+ headers = [next(r) for r in readers]
+ return StreamingCache.Reader(headers, readers).read()
+
+ def write(self, values, *labels):
+ """Writes the given values to cache.
+ """
+ to_write = [v.SerializeToString() for v in values]
Review comment:
Moved to loop
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597379444
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597934081
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-598025782
Merging as Robert's last comment was addressed by re-reading from the previous postion.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388009579
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +177,35 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return [].__iter__(), -1
Review comment:
Gotcha, didn't know about this. Changed to `iter([])`
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387888483
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -471,7 +537,44 @@ def process_element(self, element):
# We can either have the _TestStream or the _WatermarkController to emit
# the elements. We chose to emit in the _WatermarkController so that the
# element is emitted at the correct watermark value.
- for event in self.test_stream.events(self.current_index):
+ events = []
+ if self.watermark == MIN_TIMESTAMP:
+ for event in self.test_stream._set_up(self.test_stream.output_tags):
+ events.append(event)
+
+ if self.test_stream_event_channel:
+ try:
+ event = next(self.test_stream_event_channel)
Review comment:
Ack, unified it with the two methods in the test_stream_impl to return iterators. These will be stored in the global EvaluationContext 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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389168571
##########
File path: sdks/python/apache_beam/runners/direct/test_stream_impl.py
##########
@@ -226,17 +237,53 @@ def expand(self, pcoll):
def _infer_output_coder(self, input_type=None, input_coder=None):
return self.coder
- def _events_from_script(self, index):
- yield self._events[index]
-
- def events(self, index):
- return self._events_from_script(index)
-
- def begin(self):
- return 0
-
- def end(self, index):
- return index >= len(self._events)
+ @staticmethod
+ def events_from_script(events):
+ """Yields the in-memory events.
+ """
+ return itertools.chain(events)
- def next(self, index):
- return index + 1
+ @staticmethod
+ def events_from_rpc(endpoint, output_tags, coder):
+ """Yields the events received from the given endpoint.
+ """
+ stub_channel = grpc.insecure_channel(endpoint)
+ stub = beam_runner_api_pb2_grpc.TestStreamServiceStub(stub_channel)
+
+ # Request the PCollections that we are looking for from the service.
+ if list(output_tags) == [None]:
Review comment:
Yes, the proto expects a string
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388018492
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
Review comment:
Done, made [BEAM-9447](https://issues.apache.org/jira/browse/BEAM-9447)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r391157062
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
Review comment:
Ah sorry, I misunderstood. I thought you were referring to multiple concurrent writers. I'll change the code to take this into account.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597925663
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387366883
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
Review comment:
Why is it best-effort?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389122984
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -421,8 +440,12 @@ def process_element(self, element):
main_output = list(self._outputs)[0]
bundle = self._evaluation_context.create_bundle(main_output)
for tv in event.timestamped_values:
- bundle.output(
- GlobalWindows.windowed_value(tv.value, timestamp=tv.timestamp))
+ # Unreify the value into the correct window.
+ try:
+ bundle.output(WindowedValue(**tv.value))
Review comment:
Alternatively, it might be cleaner to just pass the test stream event proto here rather than decoding it and then interpreting it. (Maybe this'd be a larger change.)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389129387
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
Review comment:
No, there should only be one TestStream per pipeline. That's the reason we introduced test streams with multiple outputs. (I also fail to see how multiple TestStreams would work, as they would all seem to share the same global context and so it seems all test stream evaluators would emit all events).
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387270211
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -212,6 +218,19 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+ if self._evaluation_context._test_stream_event_stub:
+ stub = self._evaluation_context._test_stream_event_stub
+
+ if list(test_stream.output_tags) == [None]:
+ event_request = beam_runner_api_pb2.EventsRequest()
+ else:
+ event_request = beam_runner_api_pb2.EventsRequest(
+ keys=list(test_stream.output_tags))
Review comment:
Why does the set of keys even need to be sent here?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597816526
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r390011895
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
Review comment:
I'm happy to see this.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389134329
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
+ # pipeline (with only one controlling the clock). Here, we use an array in
+ # the global EvaluationContext state to keep track of the iterator for each
+ # event stream.
+ idx = len(self._evaluation_context._test_stream_events)
+
+ # If there was an endpoint defined then get the events from the
+ # TestStreamService.
+ if self._evaluation_context._test_stream_endpoint:
Review comment:
Likewise, why are we plumbing _test_stream_events through the context instead of creating it in _TestStreamEvaluator?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387884186
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -212,6 +218,19 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+ if self._evaluation_context._test_stream_event_stub:
+ stub = self._evaluation_context._test_stream_event_stub
+
+ if list(test_stream.output_tags) == [None]:
Review comment:
Added 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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387341699
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +177,35 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return [].__iter__(), -1
Review comment:
Thanks .`iter([])` is slightly more idiomatic, but this is fine.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387258635
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -399,6 +395,15 @@ def visit_transform(self, applied_ptransform):
self.consumer_tracking_visitor = ConsumerTrackingPipelineVisitor()
pipeline.visit(self.consumer_tracking_visitor)
+ test_stream_service_endpoint = \
Review comment:
Avoid backslashes for continuation. If needed, use ()'s (or just let yapf do it). (Creating the stub here seems a violation of encapsulation though, best to do it where it's used.)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r390008544
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
Review comment:
I think this is okay though. It's a bug if the file is being concurrently written to.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r391189510
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +196,38 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return iter([]), -1
- source = self.source(*labels)
+ # Otherwise, return a generator to the cached PCollection.
+ source = self._source(*labels)
range_tracker = source.get_range_tracker(None, None)
- result = list(source.read(range_tracker))
+ reader = source.read(range_tracker)
version = self._latest_version(*labels)
- return result, version
+ return reader, version
+
+ def write(self, values, *labels):
+ sink = self._sink(*labels)
+ path = self._path(*labels)
+
+ init_result = sink.initialize_write()
+ writer = sink.open_writer(init_result, path)
+ for v in values:
+ writer.write(v)
+ writer.close()
def source(self, *labels):
+ return beam.io.Read(self._source(*labels))
+
+ def sink(self, labels):
+ return beam.io.Write(self._sink(*labels))
+
+ def _source(self, *labels):
return self._reader_class(
self._glob_path(*labels), coder=self.load_pcoder(*labels))._source
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389209967
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -408,7 +410,8 @@ def visit_transform(self, applied_ptransform):
self.consumer_tracking_visitor.value_to_consumers,
self.consumer_tracking_visitor.step_names,
self.consumer_tracking_visitor.views,
- clock)
+ clock,
+ test_stream_visitor.endpoint)
Review comment:
Not changed?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387267059
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -421,8 +440,12 @@ def process_element(self, element):
main_output = list(self._outputs)[0]
bundle = self._evaluation_context.create_bundle(main_output)
for tv in event.timestamped_values:
- bundle.output(
- GlobalWindows.windowed_value(tv.value, timestamp=tv.timestamp))
+ # Unreify the value into the correct window.
+ try:
+ bundle.output(WindowedValue(**tv.value))
Review comment:
This is fragile--we should not be (attempting to) interpret values as dicts of windowed value parameters and creating globally windowed values if anything goes wrong. Instead, store the WindowedValue object in tv.value. (Alternatively, one could let `events` be full WindowedValues rather than TimestampedValues, but that might be a bigger change.)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389208621
##########
File path: sdks/python/apache_beam/runners/direct/test_stream_impl.py
##########
@@ -226,17 +237,53 @@ def expand(self, pcoll):
def _infer_output_coder(self, input_type=None, input_coder=None):
return self.coder
- def _events_from_script(self, index):
- yield self._events[index]
-
- def events(self, index):
- return self._events_from_script(index)
-
- def begin(self):
- return 0
-
- def end(self, index):
- return index >= len(self._events)
+ @staticmethod
+ def events_from_script(events):
+ """Yields the in-memory events.
+ """
+ return itertools.chain(events)
- def next(self, index):
- return index + 1
+ @staticmethod
+ def events_from_rpc(endpoint, output_tags, coder):
+ """Yields the events received from the given endpoint.
+ """
+ stub_channel = grpc.insecure_channel(endpoint)
+ stub = beam_runner_api_pb2_grpc.TestStreamServiceStub(stub_channel)
+
+ # Request the PCollections that we are looking for from the service.
+ if list(output_tags) == [None]:
Review comment:
What's the convention for this list being empty then? (And if it can be empty, in what cases do we have to provide it? Or should we explicitly be turning None to 'None'?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389205670
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -421,8 +440,12 @@ def process_element(self, element):
main_output = list(self._outputs)[0]
bundle = self._evaluation_context.create_bundle(main_output)
for tv in event.timestamped_values:
- bundle.output(
- GlobalWindows.windowed_value(tv.value, timestamp=tv.timestamp))
+ # Unreify the value into the correct window.
+ try:
+ bundle.output(WindowedValue(**tv.value))
Review comment:
Ack, for now I changed it to a WindowedValueHolder.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597925574
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on issue #11005: [BEAM-8335] Modify
the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-593703837
R: @robertwb
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r391139425
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +196,38 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return iter([]), -1
- source = self.source(*labels)
+ # Otherwise, return a generator to the cached PCollection.
+ source = self._source(*labels)
range_tracker = source.get_range_tracker(None, None)
- result = list(source.read(range_tracker))
+ reader = source.read(range_tracker)
version = self._latest_version(*labels)
- return result, version
+ return reader, version
+
+ def write(self, values, *labels):
+ sink = self._sink(*labels)
+ path = self._path(*labels)
+
+ init_result = sink.initialize_write()
+ writer = sink.open_writer(init_result, path)
+ for v in values:
+ writer.write(v)
+ writer.close()
def source(self, *labels):
+ return beam.io.Read(self._source(*labels))
+
+ def sink(self, labels):
+ return beam.io.Write(self._sink(*labels))
+
+ def _source(self, *labels):
return self._reader_class(
self._glob_path(*labels), coder=self.load_pcoder(*labels))._source
- def sink(self, *labels):
+ def _sink(self, *labels):
return self._writer_class(
Review comment:
it seems like you can just return the result of `self._writer_class(...)` all the way. This is a PTransform, so it is consistent with the itnerface (instead of re-wrapping the source / sink)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387269353
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -212,6 +218,19 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+ if self._evaluation_context._test_stream_event_stub:
+ stub = self._evaluation_context._test_stream_event_stub
+
+ if list(test_stream.output_tags) == [None]:
+ event_request = beam_runner_api_pb2.EventsRequest()
+ else:
+ event_request = beam_runner_api_pb2.EventsRequest(
+ keys=list(test_stream.output_tags))
+
+ test_stream_event_channel = stub.Events(event_request)
Review comment:
Why is this being created here?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597282906
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389876668
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -69,8 +69,8 @@ def read(self, *labels):
*labels: List of labels for PCollection instance.
Returns:
- Tuple[List[Any], int]: A tuple containing a list of items in the
- PCollection and the version number.
+ A tuple containing an iterator for the items in the PCollection and the
Review comment:
I was getting some pydoc errors and removed it. I put them back in as type hints.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387259605
##########
File path: sdks/python/apache_beam/options/pipeline_options.py
##########
@@ -1127,6 +1127,12 @@ def _add_argparse_args(cls, parser):
help='The time to wait (in milliseconds) for test pipeline to finish. '
'If it is set to None, it will wait indefinitely until the job '
'is finished.')
+ parser.add_argument(
+ '--test_stream_service_endpoint',
Review comment:
This shouldn't be a pipeline option, it should be a parameter in the TestStream proto itself.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389877508
##########
File path: sdks/python/apache_beam/runners/direct/test_stream_impl.py
##########
@@ -226,17 +237,53 @@ def expand(self, pcoll):
def _infer_output_coder(self, input_type=None, input_coder=None):
return self.coder
- def _events_from_script(self, index):
- yield self._events[index]
-
- def events(self, index):
- return self._events_from_script(index)
-
- def begin(self):
- return 0
-
- def end(self, index):
- return index >= len(self._events)
+ @staticmethod
+ def events_from_script(events):
+ """Yields the in-memory events.
+ """
+ return itertools.chain(events)
- def next(self, index):
- return index + 1
+ @staticmethod
+ def events_from_rpc(endpoint, output_tags, coder):
+ """Yields the events received from the given endpoint.
+ """
+ stub_channel = grpc.insecure_channel(endpoint)
+ stub = beam_runner_api_pb2_grpc.TestStreamServiceStub(stub_channel)
+
+ # Request the PCollections that we are looking for from the service.
+ if list(output_tags) == [None]:
Review comment:
Good point, I changed it to stringify the list then convert 'None' to None on the service side.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387260307
##########
File path: sdks/python/apache_beam/runners/direct/clock.py
##########
@@ -44,11 +46,11 @@ def time(self):
class TestClock(object):
"""Clock used for Testing"""
- def __init__(self, current_time=0):
- self._current_time = current_time
+ def __init__(self, current_time=None):
+ self._current_time = current_time if current_time else Timestamp()
def time(self):
- return self._current_time
+ return float(self._current_time)
Review comment:
We shouldn't be changing the API of TestClock. If the caller needs a float, convert it there.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389126452
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
Review comment:
This only holds if the file is not being written to concurrently (just tried it out). Otherwise it may get to EOF half way through a line.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388009916
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389210658
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
Review comment:
This was not fixed.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387882929
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -114,13 +117,6 @@ def visit_transform(self, applied_ptransform):
# FnApiRunner, and the pipeline was not meant to be run as streaming.
use_fnapi_runner = (_FnApiRunnerSupportVisitor().accept(pipeline))
- # Also ensure grpc is available.
Review comment:
Added back
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597379482
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597809289
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597871799
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] aaltay commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
aaltay commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-601288550
I believe you need to add pathlib / pathlib2 to the setup.py. At least for py2. For python3 it might be part of the standard library.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387371561
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
Review comment:
Is it possible for this to read part of a line? Perhaps the check should be whether it ends in a newline, otherwise you have to go back and try to read more.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597925638
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389205758
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -500,23 +523,21 @@ def __init__(
input_committed_bundle,
side_inputs)
self.test_stream = applied_ptransform.transform
+ self.event_index = 0
Review comment:
Removed this. Also, the is_done is needed because of the small cleanup. There used to be a weird C++ iterator style on the TestStream object to get the event. So there needs to be a small state variable from the process_elements call to the finish_bundle call informing itself if there are any more elements to process.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387259167
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -114,13 +117,6 @@ def visit_transform(self, applied_ptransform):
# FnApiRunner, and the pipeline was not meant to be run as streaming.
use_fnapi_runner = (_FnApiRunnerSupportVisitor().accept(pipeline))
- # Also ensure grpc is available.
Review comment:
Why was this removed?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387888120
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -432,6 +455,45 @@ def finish_bundle(self):
self, self.bundles, [], None, {None: self._watermark})
+class PairWithTimingEvaluator(_TransformEvaluator):
+ """TransformEvaluator for the PairWithTiming transform.
+
+ This transform takes an element as an input and outputs
+ KV(element, `TimingInfo`). Where the `TimingInfo` contains both the
+ processing time timestamp and watermark.
+ """
+ def __init__(
+ self,
+ evaluation_context,
+ applied_ptransform,
+ input_committed_bundle,
+ side_inputs):
+ assert not side_inputs
+ super(PairWithTimingEvaluator, self).__init__(
+ evaluation_context,
+ applied_ptransform,
+ input_committed_bundle,
+ side_inputs)
+
+ def start_bundle(self):
+ main_output = list(self._outputs)[0]
+ self.bundle = self._evaluation_context.create_bundle(main_output)
+
+ def process_element(self, element):
+ watermark_manager = self._evaluation_context._watermark_manager
+ watermarks = watermark_manager.get_watermarks(self._applied_ptransform)
+
+ output_watermark = watermarks.output_watermark
+ now = Timestamp(seconds=watermark_manager._clock.time())
+ timing_info = TimingInfo(now, output_watermark)
+
+ element.value = (element.value, timing_info)
Review comment:
Ack returning a new WindowedValue
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389950042
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
+ # pipeline (with only one controlling the clock). Here, we use an array in
+ # the global EvaluationContext state to keep track of the iterator for each
+ # event stream.
+ idx = len(self._evaluation_context._test_stream_events)
+
+ # If there was an endpoint defined then get the events from the
+ # TestStreamService.
+ if self._evaluation_context._test_stream_endpoint:
Review comment:
Gotcha, I'll set the global context once and move the state to a global on the _TestStreamEvaluator. That way the owner is strictly the _TestStreamEvaluator.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597380237
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387367517
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
Review comment:
Can you please file a JIRA to make this more general? (I don't see anything yet that would preclude writing something that works for all runners.)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597934026
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-595361932
Restest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387887355
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -421,8 +440,12 @@ def process_element(self, element):
main_output = list(self._outputs)[0]
bundle = self._evaluation_context.create_bundle(main_output)
for tv in event.timestamped_values:
- bundle.output(
- GlobalWindows.windowed_value(tv.value, timestamp=tv.timestamp))
+ # Unreify the value into the correct window.
+ try:
+ bundle.output(WindowedValue(**tv.value))
Review comment:
Unfortunately the Python SDK interprets an element that's a WindowedValue as an element that should have its windowing information changed. This is used directly by the ReverseTestStream to produce WindowedValue args that keep the windowing information. But if these elements were WindowedValues themselves, then the Python SDK would reinterpret and unnest them at every step.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387280719
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -471,7 +537,44 @@ def process_element(self, element):
# We can either have the _TestStream or the _WatermarkController to emit
# the elements. We chose to emit in the _WatermarkController so that the
# element is emitted at the correct watermark value.
- for event in self.test_stream.events(self.current_index):
+ events = []
+ if self.watermark == MIN_TIMESTAMP:
+ for event in self.test_stream._set_up(self.test_stream.output_tags):
+ events.append(event)
+
+ if self.test_stream_event_channel:
+ try:
+ event = next(self.test_stream_event_channel)
Review comment:
Don't we already have this translation logic elsewhere. Also, it seems we could unify events-from-memory vs. events-from-grpc more.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388009717
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
Review comment:
Clarified in the comment that it is best-effort replay not best-effort writing. The original comment was confusing.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389877173
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -408,7 +410,8 @@ def visit_transform(self, applied_ptransform):
self.consumer_tracking_visitor.value_to_consumers,
self.consumer_tracking_visitor.step_names,
self.consumer_tracking_visitor.views,
- clock)
+ clock,
+ test_stream_visitor.endpoint)
Review comment:
I pushed this particular change to https://github.com/apache/beam/pull/10994
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389881311
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
+ # pipeline (with only one controlling the clock). Here, we use an array in
+ # the global EvaluationContext state to keep track of the iterator for each
+ # event stream.
+ idx = len(self._evaluation_context._test_stream_events)
Review comment:
In https://github.com/apache/beam/pull/10994 I changed this to a singleton
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389209836
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -69,8 +69,8 @@ def read(self, *labels):
*labels: List of labels for PCollection instance.
Returns:
- Tuple[List[Any], int]: A tuple containing a list of items in the
- PCollection and the version number.
+ A tuple containing an iterator for the items in the PCollection and the
Review comment:
Please leave the types (it makes it easier to follow).
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389209005
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
+ # pipeline (with only one controlling the clock). Here, we use an array in
+ # the global EvaluationContext state to keep track of the iterator for each
+ # event stream.
+ idx = len(self._evaluation_context._test_stream_events)
+
+ # If there was an endpoint defined then get the events from the
+ # TestStreamService.
+ if self._evaluation_context._test_stream_endpoint:
Review comment:
I'm not convinced that storing it on the global context is an improvement on keeping it on the bundle that gets passed back to the (new) _TestStreamEvaluator instance. It seems harder to track who owns/accesses it (and also that it only gets written to once). I agree the original way of doing things may not have been ideal either.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388009673
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +177,35 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return [].__iter__(), -1
- source = self.source(*labels)
+ # Otherwise, return a generator to the cached PCollection.
+ source = self._source(*labels)
range_tracker = source.get_range_tracker(None, None)
- result = list(source.read(range_tracker))
+ reader = source.read(range_tracker)
version = self._latest_version(*labels)
- return result, version
+ return reader, version
+
+ def write(self, values, *labels):
+ sink = self._sink(*labels)
+ path = self._path(*labels)
+ with open(path, 'wb') as f:
+ for v in values:
+ sink.write_record(f, v)
Review comment:
Cool, thank you. Changed to the above code.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387373091
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
+
+ # Check if we are at EOF.
+ if not line:
+ # Complete reading only when the cache is complete.
+ if self._is_cache_complete():
+ break
+
+ if not tail:
+ break
+
+ # Otherwise wait for new data in the file to be written.
+ time.sleep(0.5)
+ fh.seek(pos)
+ else:
+ # The first line at pos = 0 is always the header. Read the line without
+ # the new line.
+ if pos == 0:
+ header = TestStreamFileHeader()
+ header.ParseFromString(self._coder.decode(line[:-1]))
+ yield header
+ else:
+ record = TestStreamFileRecord()
+ record.ParseFromString(self._coder.decode(line[:-1]))
+ yield record
+
+ def read(self, tail):
+ """Reads all TestStreamFile(Header|TestStreamFileRecord)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ with self._wait_until_file_exists() as f:
+ for e in self._emit_from_file(f, tail):
+ yield e
+
+
+class StreamingCache(CacheManager):
"""Abstraction that holds the logic for reading and writing to cache.
"""
- def __init__(self, readers):
- self._readers = readers
+ def __init__(
+ self, cache_dir, is_cache_complete=None, sample_resolution_sec=0.1):
+ self._sample_resolution_sec = sample_resolution_sec
+ self._is_cache_complete = is_cache_complete
+
+ if cache_dir:
+ self._cache_dir = cache_dir
+ else:
+ self._cache_dir = tempfile.mkdtemp(
+ prefix='interactive-temp-', dir=os.environ.get('TEST_TMPDIR', None))
+
+ # List of saved pcoders keyed by PCollection path. It is OK to keep this
+ # list in memory because once FileBasedCacheManager object is
+ # destroyed/re-created it loses the access to previously written cache
+ # objects anyways even if cache_dir already exists. In other words,
+ # it is not possible to resume execution of Beam pipeline from the
+ # saved cache if FileBasedCacheManager has been reset.
+ #
+ # However, if we are to implement better cache persistence, one needs
+ # to take care of keeping consistency between the cached PCollection
+ # and its PCoder type.
+ self._saved_pcoders = {}
+ self._default_pcoder = SafeFastPrimitivesCoder()
+
+ def exists(self, *labels):
+ path = os.path.join(self._cache_dir, *labels)
+ return os.path.exists(path)
+
+ # TODO(srohde): Modify this to return the correct version.
+ def read(self, *labels):
+ """Returns a generator to read all records from file.
+
+ Does not tail.
+ """
+ if not self.exists(*labels):
+ return [].__iter__(), -1
+
+ reader = StreamingCacheSource(
+ self._cache_dir, labels,
+ is_cache_complete=self._is_cache_complete).read(tail=False)
+ header = next(reader)
+ return StreamingCache.Reader([header], [reader]).read(), 1
+
+ def read_multiple(self, labels):
+ """Returns a generator to read all records from file.
+
+ Does tail until the cache is complete. This is because it is used in the
+ TestStreamServiceController to read from file which is only used during
+ pipeline runtime which needs to block.
+ """
+ readers = [
+ StreamingCacheSource(
+ self._cache_dir, l,
+ is_cache_complete=self._is_cache_complete).read(tail=True)
+ for l in labels
+ ]
+ headers = [next(r) for r in readers]
+ return StreamingCache.Reader(headers, readers).read()
+
+ def write(self, values, *labels):
+ """Writes the given values to cache.
+ """
+ to_write = [v.SerializeToString() for v in values]
Review comment:
Why is this not done in the loop below?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389205690
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
Review comment:
Ack I updated the TestStreamTests to use the multi-output teststream instead of two teststreams. And changed this back to a singleton.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389117616
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -421,8 +440,12 @@ def process_element(self, element):
main_output = list(self._outputs)[0]
bundle = self._evaluation_context.create_bundle(main_output)
for tv in event.timestamped_values:
- bundle.output(
- GlobalWindows.windowed_value(tv.value, timestamp=tv.timestamp))
+ # Unreify the value into the correct window.
+ try:
+ bundle.output(WindowedValue(**tv.value))
Review comment:
In that case, it'd be cleaner to make a new class WindowedValueHolder, or use a 1-tuple of WindowedValue, or something explicit like that rather than trying to use every element as kwargs to WindowedValue constructor and falling back otherwise.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597869429
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r391189581
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +196,38 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return iter([]), -1
- source = self.source(*labels)
+ # Otherwise, return a generator to the cached PCollection.
+ source = self._source(*labels)
range_tracker = source.get_range_tracker(None, None)
- result = list(source.read(range_tracker))
+ reader = source.read(range_tracker)
version = self._latest_version(*labels)
- return result, version
+ return reader, version
+
+ def write(self, values, *labels):
+ sink = self._sink(*labels)
+ path = self._path(*labels)
+
+ init_result = sink.initialize_write()
+ writer = sink.open_writer(init_result, path)
+ for v in values:
+ writer.write(v)
+ writer.close()
def source(self, *labels):
+ return beam.io.Read(self._source(*labels))
+
+ def sink(self, labels):
+ return beam.io.Write(self._sink(*labels))
+
+ def _source(self, *labels):
return self._reader_class(
self._glob_path(*labels), coder=self.load_pcoder(*labels))._source
- def sink(self, *labels):
+ def _sink(self, *labels):
return self._writer_class(
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597886665
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387345481
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +177,35 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return [].__iter__(), -1
- source = self.source(*labels)
+ # Otherwise, return a generator to the cached PCollection.
+ source = self._source(*labels)
range_tracker = source.get_range_tracker(None, None)
- result = list(source.read(range_tracker))
+ reader = source.read(range_tracker)
version = self._latest_version(*labels)
- return result, version
+ return reader, version
+
+ def write(self, values, *labels):
+ sink = self._sink(*labels)
+ path = self._path(*labels)
+ with open(path, 'wb') as f:
+ for v in values:
+ sink.write_record(f, v)
Review comment:
This is not part of the public API for sinks. Instead, do
```
writer = sink.open_writer(init_result, path)
for v in values:
writer.write(v)
write_results.append(writer.close())
```
https://github.com/apache/beam/blob/release-2.18.0/sdks/python/apache_beam/io/iobase.py#L668
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r391139271
##########
File path: sdks/python/apache_beam/runners/interactive/cache_manager.py
##########
@@ -167,20 +196,38 @@ def load_pcoder(self, *labels):
self._saved_pcoders[self._path(*labels)])
def read(self, *labels):
+ # Return an iterator to an empty list if it doesn't exist.
if not self.exists(*labels):
- return [], -1
+ return iter([]), -1
- source = self.source(*labels)
+ # Otherwise, return a generator to the cached PCollection.
+ source = self._source(*labels)
range_tracker = source.get_range_tracker(None, None)
- result = list(source.read(range_tracker))
+ reader = source.read(range_tracker)
version = self._latest_version(*labels)
- return result, version
+ return reader, version
+
+ def write(self, values, *labels):
+ sink = self._sink(*labels)
+ path = self._path(*labels)
+
+ init_result = sink.initialize_write()
+ writer = sink.open_writer(init_result, path)
+ for v in values:
+ writer.write(v)
+ writer.close()
def source(self, *labels):
+ return beam.io.Read(self._source(*labels))
+
+ def sink(self, labels):
+ return beam.io.Write(self._sink(*labels))
+
+ def _source(self, *labels):
return self._reader_class(
self._glob_path(*labels), coder=self.load_pcoder(*labels))._source
Review comment:
it seems like you can just return the result of `self._reader_class(...)` all the way. This is a PTransform, so it is consistent with the itnerface (instead of re-wrapping the source / sink)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389131463
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -408,7 +410,8 @@ def visit_transform(self, applied_ptransform):
self.consumer_tracking_visitor.value_to_consumers,
self.consumer_tracking_visitor.step_names,
self.consumer_tracking_visitor.views,
- clock)
+ clock,
+ test_stream_visitor.endpoint)
Review comment:
Why is this passed via the context, instead of accessed directly on the test stream object when it is being processed?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388018812
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
+ # Raise the exception if it is not a FileNotFoundError.
+ raise
+ time.sleep(1)
+ if now_secs >= timeout_timestamp_secs:
+ raise RuntimeError(
+ "Timed out waiting for file '{}' to be available".format(path))
+ return f
+
+ def _emit_from_file(self, fh, tail):
+ """Emits the TestStreamFile(Header|Record)s from file.
+
+ This returns a generator to be able to read all lines from the given file.
+ If `tail` is True, then it will wait until the cache is complete to exit.
+ Otherwise, it will read the file only once.
+ """
+ # Always read at least once to read the whole file.
+ while True:
+ pos = fh.tell()
+ line = fh.readline()
Review comment:
The Python API for the `readline()` only allows for part of a line to be read if the optional parameter `size` is given and is non-negative. So this is okay and will read the whole line.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387262647
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -212,6 +218,19 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+ if self._evaluation_context._test_stream_event_stub:
+ stub = self._evaluation_context._test_stream_event_stub
+
+ if list(test_stream.output_tags) == [None]:
Review comment:
This logic feels odd, needs an explanation.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389205704
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -408,7 +410,8 @@ def visit_transform(self, applied_ptransform):
self.consumer_tracking_visitor.value_to_consumers,
self.consumer_tracking_visitor.step_names,
self.consumer_tracking_visitor.views,
- clock)
+ clock,
+ test_stream_visitor.endpoint)
Review comment:
Ack, changed to get endpoint from TestStream object.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387368473
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ filename,
+ sample_resolution_sec,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._filename = filename
+ self._sample_resolution_sec = sample_resolution_sec
+ self._coder = coder
+ self._path = os.path.join(self._cache_dir, self._filename)
+
+ @property
+ def path(self):
+ """Returns the path the sink leads to."""
+ return self._path
+
+ def expand(self, pcoll):
+ class StreamingWriteToText(beam.DoFn):
+ """DoFn that performs the writing.
+
+ Note that the other file writing methods cannot be used in streaming
+ contexts.
+ """
+ def __init__(self, full_path, coder=SafeFastPrimitivesCoder()):
+ self._full_path = full_path
+ self._coder = coder
+
+ # Try and make the given path.
+ os.makedirs(os.path.dirname(full_path), exist_ok=True)
+
+ def start_bundle(self):
+ # Open the file for 'append-mode' and writing 'bytes'.
+ self._fh = open(self._full_path, 'ab')
+
+ def finish_bundle(self):
+ self._fh.close()
+
+ def process(self, e):
+ """Appends the given element to the file.
+ """
+ self._fh.write(self._coder.encode(e))
+ self._fh.write(b'\n')
+
+ return (
+ pcoll
+ | ReverseTestStream(
+ output_tag=self._filename,
+ sample_resolution_sec=self._sample_resolution_sec,
+ output_format=ReverseTestStream.Format.
+ SERIALIZED_TEST_STREAM_FILE_RECORDS,
+ coder=self._coder)
+ | beam.ParDo(
+ StreamingWriteToText(full_path=self._path, coder=self._coder)))
+
+
+class StreamingCacheSource:
+ """A class that reads and parses TestStreamFile(Header|Reader)s.
+
+ This source operates in the following way:
+
+ 1. Wait for up to `timeout_secs` for the file to be available.
+ 2. Read, parse, and emit the entire contents of the file
+ 3. Wait for more events to come or until `is_cache_complete` returns True
+ 4. If there are more events, then go to 2
+ 5. Otherwise, stop emitting.
+
+ This class is used to read from file and send its to the TestStream via the
+ StreamingCacheManager.Reader.
+ """
+ def __init__(
+ self,
+ cache_dir,
+ labels,
+ is_cache_complete=None,
+ coder=SafeFastPrimitivesCoder()):
+ self._cache_dir = cache_dir
+ self._coder = coder
+ self._labels = labels
+ self._is_cache_complete = (
+ is_cache_complete if is_cache_complete else lambda: True)
+
+ def _wait_until_file_exists(self, timeout_secs=30):
+ """Blocks until the file exists for a maximum of timeout_secs.
+ """
+ f = None
+ now_secs = time.time()
+ timeout_timestamp_secs = now_secs + timeout_secs
+
+ # Wait for up to `timeout_secs` for the file to be available.
+ while f is None and now_secs < timeout_timestamp_secs:
+ now_secs = time.time()
+ try:
+ path = os.path.join(self._cache_dir, *self._labels)
+ f = open(path, mode='r')
+ except EnvironmentError as e:
+ # For Python2 and Python3 compatibility, this checks the
+ # EnvironmentError to see if the file exists.
+ # TODO: Change this to a FileNotFoundError when Python3 migration is
+ # complete.
+ import errno
+ if e.errno != errno.ENOENT:
Review comment:
Just using os.path.exists would be cleaner.
```
start = time.time()
while not os.path.exists(path):
time.sleep(1)
if time.time() - start > timeout_timestamp_secs:
raise RuntimeError(...)
return open(path)
```
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597884370
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597884434
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387883702
##########
File path: sdks/python/apache_beam/runners/direct/clock.py
##########
@@ -44,11 +46,11 @@ def time(self):
class TestClock(object):
"""Clock used for Testing"""
- def __init__(self, current_time=0):
- self._current_time = current_time
+ def __init__(self, current_time=None):
+ self._current_time = current_time if current_time else Timestamp()
def time(self):
- return self._current_time
+ return float(self._current_time)
Review comment:
Not changing it, fixing it. The TestClock has the same interface as the RealClock and should return the same thing. See the comment on the Clock interface.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389205722
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
+ # pipeline (with only one controlling the clock). Here, we use an array in
+ # the global EvaluationContext state to keep track of the iterator for each
+ # event stream.
+ idx = len(self._evaluation_context._test_stream_events)
+
+ # If there was an endpoint defined then get the events from the
+ # TestStreamService.
+ if self._evaluation_context._test_stream_endpoint:
Review comment:
This is a way to keep track of where in the event stream the TestStream is. Because the _TestStreamEvaluator can be created multiple times, the iterator will become invalid once we try to make a new one. This is actually a cleanup of the old implementation where the state was kept in an unprocessed bundle that was passed to itself.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] aaltay commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
aaltay commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-601352587
> I believe you need to add pathlib / pathlib2 to the setup.py. At least for py2. For python3 it might be part of the standard library.
This turns out to be a non-issue, because ipython depends on pathlib2 and brings that to the environment.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387262044
##########
File path: sdks/python/apache_beam/runners/direct/evaluation_context.py
##########
@@ -273,7 +274,8 @@ def __init__(self,
] # type: List[Tuple[TransformExecutor, Timestamp]]
self._counter_factory = counters.CounterFactory()
self._metrics = DirectMetrics()
-
+ self._test_stream_event_stub = test_stream_event_stub
+ self._test_stream_event_channel = None
Review comment:
Similarly, the test_stream_event_channel doesn't seem a property of the evaluation context, rather a property of the test stream (evaluator) itself.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387883019
##########
File path: sdks/python/apache_beam/options/pipeline_options.py
##########
@@ -1127,6 +1127,12 @@ def _add_argparse_args(cls, parser):
help='The time to wait (in milliseconds) for test pipeline to finish. '
'If it is set to None, it will wait indefinitely until the job '
'is finished.')
+ parser.add_argument(
+ '--test_stream_service_endpoint',
Review comment:
Ack, taken from proto
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597866914
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387887996
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -212,6 +218,19 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+ if self._evaluation_context._test_stream_event_stub:
+ stub = self._evaluation_context._test_stream_event_stub
+
+ if list(test_stream.output_tags) == [None]:
+ event_request = beam_runner_api_pb2.EventsRequest()
+ else:
+ event_request = beam_runner_api_pb2.EventsRequest(
+ keys=list(test_stream.output_tags))
Review comment:
To know which PCollections to read from. Added 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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389210201
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
+ # pipeline (with only one controlling the clock). Here, we use an array in
+ # the global EvaluationContext state to keep track of the iterator for each
+ # event stream.
+ idx = len(self._evaluation_context._test_stream_events)
Review comment:
Is this ever non-empty? It seems repeated calls would append everything. (Or is it populated elsewhere?)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597246280
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r388018492
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -19,15 +19,298 @@
from __future__ import absolute_import
+import itertools
+import os
+import shutil
+import tempfile
+import time
+
+import apache_beam as beam
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileHeader
+from apache_beam.portability.api.beam_interactive_api_pb2 import TestStreamFileRecord
from apache_beam.portability.api.beam_runner_api_pb2 import TestStreamPayload
+from apache_beam.runners.interactive.cache_manager import CacheManager
+from apache_beam.runners.interactive.cache_manager import SafeFastPrimitivesCoder
+from apache_beam.testing.test_stream import ReverseTestStream
from apache_beam.utils import timestamp
-class StreamingCache(object):
+class StreamingCacheSink(beam.PTransform):
+ """A PTransform that writes TestStreamFile(Header|Records)s to file.
+
+ This transform takes in an arbitrary element stream and writes the best-effort
+ list of TestStream events (as TestStreamFileRecords) to file.
+
+ Note that this PTransform is assumed to be only run on a single machine where
+ the following assumptions are correct: elements come in ordered, no two
+ transforms are writing to the same file. This PTransform is assumed to only
+ run correctly with the DirectRunner.
Review comment:
Done, made BEAM-9447
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387271259
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -432,6 +455,45 @@ def finish_bundle(self):
self, self.bundles, [], None, {None: self._watermark})
+class PairWithTimingEvaluator(_TransformEvaluator):
+ """TransformEvaluator for the PairWithTiming transform.
+
+ This transform takes an element as an input and outputs
+ KV(element, `TimingInfo`). Where the `TimingInfo` contains both the
+ processing time timestamp and watermark.
+ """
+ def __init__(
+ self,
+ evaluation_context,
+ applied_ptransform,
+ input_committed_bundle,
+ side_inputs):
+ assert not side_inputs
+ super(PairWithTimingEvaluator, self).__init__(
+ evaluation_context,
+ applied_ptransform,
+ input_committed_bundle,
+ side_inputs)
+
+ def start_bundle(self):
+ main_output = list(self._outputs)[0]
+ self.bundle = self._evaluation_context.create_bundle(main_output)
+
+ def process_element(self, element):
+ watermark_manager = self._evaluation_context._watermark_manager
+ watermarks = watermark_manager.get_watermarks(self._applied_ptransform)
+
+ output_watermark = watermarks.output_watermark
+ now = Timestamp(seconds=watermark_manager._clock.time())
+ timing_info = TimingInfo(now, output_watermark)
+
+ element.value = (element.value, timing_info)
Review comment:
Don't mutate the input.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387268742
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -421,8 +440,12 @@ def process_element(self, element):
main_output = list(self._outputs)[0]
bundle = self._evaluation_context.create_bundle(main_output)
for tv in event.timestamped_values:
- bundle.output(
- GlobalWindows.windowed_value(tv.value, timestamp=tv.timestamp))
+ # Unreify the value into the correct window.
+ try:
+ bundle.output(WindowedValue(**tv.value))
Review comment:
(For that matter, it seems odd to be mutating ElementEvents in an operator called _WatermarkControllerEvaluator.)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389131842
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -214,11 +214,31 @@ class _TestStreamRootBundleProvider(RootBundleProvider):
"""
def get_root_bundles(self):
test_stream = self._applied_ptransform.transform
+
+ # The TestStream specification allows for multiple TestStreams in the same
+ # pipeline (with only one controlling the clock). Here, we use an array in
+ # the global EvaluationContext state to keep track of the iterator for each
+ # event stream.
+ idx = len(self._evaluation_context._test_stream_events)
+
+ # If there was an endpoint defined then get the events from the
+ # TestStreamService.
+ if self._evaluation_context._test_stream_endpoint:
Review comment:
(Regarding the comment above) we should get this from test_stream, rather than plumb it through the context.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389123573
##########
File path: sdks/python/apache_beam/runners/direct/test_stream_impl.py
##########
@@ -226,17 +237,53 @@ def expand(self, pcoll):
def _infer_output_coder(self, input_type=None, input_coder=None):
return self.coder
- def _events_from_script(self, index):
- yield self._events[index]
-
- def events(self, index):
- return self._events_from_script(index)
-
- def begin(self):
- return 0
-
- def end(self, index):
- return index >= len(self._events)
+ @staticmethod
+ def events_from_script(events):
+ """Yields the in-memory events.
+ """
+ return itertools.chain(events)
- def next(self, index):
- return index + 1
+ @staticmethod
+ def events_from_rpc(endpoint, output_tags, coder):
+ """Yields the events received from the given endpoint.
+ """
+ stub_channel = grpc.insecure_channel(endpoint)
+ stub = beam_runner_api_pb2_grpc.TestStreamServiceStub(stub_channel)
+
+ # Request the PCollections that we are looking for from the service.
+ if list(output_tags) == [None]:
Review comment:
Why can't we just pass the `[None]`? Is it because None is not a string?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597811938
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r389136914
##########
File path: sdks/python/apache_beam/runners/direct/transform_evaluator.py
##########
@@ -500,23 +523,21 @@ def __init__(
input_committed_bundle,
side_inputs)
self.test_stream = applied_ptransform.transform
+ self.event_index = 0
Review comment:
Why did this have to move out of start_bundle? (Similarly, why do we need is_done now and didn't before?)
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387888793
##########
File path: sdks/python/apache_beam/runners/direct/direct_runner.py
##########
@@ -399,6 +395,15 @@ def visit_transform(self, applied_ptransform):
self.consumer_tracking_visitor = ConsumerTrackingPipelineVisitor()
pipeline.visit(self.consumer_tracking_visitor)
+ test_stream_service_endpoint = \
Review comment:
Ack, using parentheses.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11005:
[BEAM-8335] Modify the StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#discussion_r387883987
##########
File path: sdks/python/apache_beam/runners/direct/evaluation_context.py
##########
@@ -273,7 +274,8 @@ def __init__(self,
] # type: List[Tuple[TransformExecutor, Timestamp]]
self._counter_factory = counters.CounterFactory()
self._metrics = DirectMetrics()
-
+ self._test_stream_event_stub = test_stream_event_stub
+ self._test_stream_event_channel = None
Review comment:
Ack, I moved the event generation into the test_stream_impl 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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11005: [BEAM-8335] Modify the
StreamingCache to subclass the CacheManager
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11005: [BEAM-8335] Modify the StreamingCache to subclass the CacheManager
URL: https://github.com/apache/beam/pull/11005#issuecomment-597809672
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services