[BEAM-8335] Modify the StreamingCache to subclass the CacheManager #11005

rohdesamuel · 2020-02-28T22:19:45Z

Modifies the StreamingCache to subclass the CacheManager. This allows for the overriding of the source and sink transforms.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) 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.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

rohdesamuel · 2020-03-03T00:30:49Z

R: @robertwb

robertwb

I just realized I started reviewing the full CL, not just that one commit.

sdks/python/apache_beam/runners/direct/direct_runner.py

sdks/python/apache_beam/options/pipeline_options.py

sdks/python/apache_beam/runners/direct/clock.py

sdks/python/apache_beam/runners/direct/evaluation_context.py

sdks/python/apache_beam/runners/direct/transform_evaluator.py

sdks/python/apache_beam/runners/interactive/cache_manager.py

sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py

robertwb · 2020-03-04T00:15:45Z

sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py

+    # Always read at least once to read the whole file.
+    while True:
+      pos = fh.tell()
+      line = fh.readline()


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.

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 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 was not fixed.

I think this is okay though. It's a bug if the file is being concurrently written to.

We've discussed this. Sam will add the fix by re-reading from the initial position.

Ah sorry, I misunderstood. I thought you were referring to multiple concurrent writers. I'll change the code to take this into account.

sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py

robertwb · 2020-03-05T17:53:21Z

Restest this please

sdks/python/apache_beam/runners/direct/test_stream_impl.py

robertwb · 2020-03-06T20:25:01Z

sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py

+    # Always read at least once to read the whole file.
+    while True:
+      pos = fh.tell()
+      line = fh.readline()


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.

sdks/python/apache_beam/runners/direct/transform_evaluator.py

sdks/python/apache_beam/runners/direct/direct_runner.py

sdks/python/apache_beam/runners/direct/transform_evaluator.py

robertwb

Thanks. FWIW, it's easier to review if you don't force-push so I can see the edits from last time. (You can rebase to squash individual commits into a fewer number of commits.)

sdks/python/apache_beam/runners/direct/test_stream_impl.py

sdks/python/apache_beam/runners/direct/transform_evaluator.py

sdks/python/apache_beam/runners/interactive/cache_manager.py

sdks/python/apache_beam/runners/direct/direct_runner.py

sdks/python/apache_beam/runners/direct/transform_evaluator.py

robertwb · 2020-03-07T01:07:27Z

sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py

+    # Always read at least once to read the whole file.
+    while True:
+      pos = fh.tell()
+      line = fh.readline()


This was not fixed.

pabloem · 2020-03-10T18:35:35Z

retest this please

pabloem · 2020-03-10T19:54:44Z

retest this please

pabloem · 2020-03-11T00:07:39Z

retest this please

pabloem · 2020-03-11T00:07:47Z

retest this please

pabloem · 2020-03-11T00:10:49Z

retest this please

sdks/python/apache_beam/runners/interactive/cache_manager.py

pabloem · 2020-03-11T17:52:35Z

sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py

+    # Always read at least once to read the whole file.
+    while True:
+      pos = fh.tell()
+      line = fh.readline()


We've discussed this. Sam will add the fix by re-reading from the initial position.

pabloem · 2020-03-11T18:55:30Z

retest this please

pabloem · 2020-03-11T18:56:15Z

retest this please

pabloem · 2020-03-11T19:00:56Z

retest this please

pabloem · 2020-03-11T19:10:16Z

retest this please

pabloem · 2020-03-11T20:37:47Z

Run Python PreCommit

pabloem · 2020-03-11T20:43:57Z

Run Python PreCommit

pabloem · 2020-03-11T20:49:47Z

Run Python PreCommit

pabloem · 2020-03-11T21:21:11Z

retest this please

pabloem · 2020-03-11T21:21:20Z

retest this please

pabloem · 2020-03-11T21:27:19Z

Run Python PreCommit

Change-Id: Ib61aa3fac53d9109178744e11eeebe5c5da0929c

pabloem · 2020-03-11T23:13:43Z

retest this please

pabloem · 2020-03-11T23:13:53Z

retest this please

pabloem · 2020-03-11T23:14:00Z

retest this please

pabloem · 2020-03-11T23:42:34Z

retest this please

pabloem · 2020-03-11T23:42:45Z

retest this please

pabloem · 2020-03-12T06:19:47Z

Merging as Robert's last comment was addressed by re-reading from the previous postion.

aaltay · 2020-03-19T16:47:12Z

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.

aaltay · 2020-03-19T18:44:37Z

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.

probot-autolabeler bot added the python label Feb 28, 2020

rohdesamuel changed the title ~~Streamingcache merge~~ [BEAM-8335] Modify the StreamingCache to subclass the CacheManager Feb 28, 2020

rohdesamuel force-pushed the streamingcache_merge branch 3 times, most recently from 514c276 to 62cb7a9 Compare March 3, 2020 00:29

rohdesamuel marked this pull request as ready for review March 3, 2020 00:30

robertwb reviewed Mar 3, 2020

View reviewed changes

robertwb requested changes Mar 4, 2020

View reviewed changes

rohdesamuel mentioned this pull request Mar 4, 2020

[BEAM-8335] TeststreamService integration with DirectRunner #10994

Merged

4 tasks

rohdesamuel force-pushed the streamingcache_merge branch 2 times, most recently from 3bc5d82 to f41c34f Compare March 5, 2020 00:42

rohdesamuel force-pushed the streamingcache_merge branch 3 times, most recently from 6d65928 to 67097b0 Compare March 6, 2020 19:16

robertwb reviewed Mar 6, 2020

View reviewed changes

robertwb reviewed Mar 7, 2020

View reviewed changes

rohdesamuel force-pushed the streamingcache_merge branch 2 times, most recently from 7b05651 to 8ff7323 Compare March 10, 2020 18:29

probot-autolabeler bot added extensions java sql website labels Mar 10, 2020

rohdesamuel force-pushed the streamingcache_merge branch from 8ff7323 to 3f520ba Compare March 10, 2020 18:32

rohdesamuel force-pushed the streamingcache_merge branch from 3f520ba to 7a6a4c6 Compare March 10, 2020 20:16

pabloem reviewed Mar 11, 2020

View reviewed changes

rohdesamuel force-pushed the streamingcache_merge branch from 7a6a4c6 to af68392 Compare March 11, 2020 18:48

KevinGG mentioned this pull request Mar 11, 2020

[BEAM-8335] Implemented Capture Size limitation #11050

Merged

4 tasks

rohdesamuel force-pushed the streamingcache_merge branch from af68392 to 3b38f88 Compare March 11, 2020 21:19

[BEAM-8335] Modify the StreamingCache to subclass the CacheManager

c69d409

Change-Id: Ib61aa3fac53d9109178744e11eeebe5c5da0929c

rohdesamuel force-pushed the streamingcache_merge branch from 3b38f88 to c69d409 Compare March 11, 2020 23:11

pabloem merged commit bb9826c into apache:master Mar 12, 2020

[BEAM-8335] Modify the StreamingCache to subclass the CacheManager #11005

[BEAM-8335] Modify the StreamingCache to subclass the CacheManager #11005

Conversation

rohdesamuel commented Feb 28, 2020 • edited

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

rohdesamuel commented Mar 3, 2020

robertwb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robertwb commented Mar 5, 2020

Choose a reason for hiding this comment

robertwb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pabloem commented Mar 10, 2020

pabloem commented Mar 10, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

Choose a reason for hiding this comment

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 11, 2020

pabloem commented Mar 12, 2020

aaltay commented Mar 19, 2020

aaltay commented Mar 19, 2020

rohdesamuel commented Feb 28, 2020 •

edited