Add MatchContinuously PTransform by InigoSJ · Pull Request #15106 · apache/beam

InigoSJ · 2021-06-30T12:42:08Z

Adds MatchContinuously to Python SDK. This is has the same purpose as FileIO.MatchAll.Continuously but a different implementation, since Watch is not available.

The idea is to use PeriodicImpulse and MatchAll to check for new files, and use a StatefulDoFn to remove duplicates.

I added the experimental since in the future it may we worth changing to a Python version of Watch

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.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Samza	Twister2
Go	---	---	---
Java
Python	---	---	---
XLang		---	---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

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

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

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

InigoSJ · 2021-06-30T12:46:25Z

R: @pabloem

I had some problem running the tests on my env, so I would need to double check them here.

codecov · 2021-06-30T12:59:25Z

Codecov Report

Merging #15106 (d315d57) into master (b86fcf9) will increase coverage by 0.02%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##           master   #15106      +/-   ##
==========================================
+ Coverage   83.78%   83.81%   +0.02%     
==========================================
  Files         439      441       +2     
  Lines       59245    59530     +285     
==========================================
+ Hits        49638    49894     +256     
- Misses       9607     9636      +29

Impacted Files	Coverage Δ
sdks/python/apache_beam/io/fileio.py	`95.89% <100.00%> (+0.42%)`	⬆️
sdks/python/apache_beam/internal/metrics/metric.py	`86.17% <0.00%> (-1.07%)`	⬇️
...hon/apache_beam/runners/worker/bundle_processor.py	`93.48% <0.00%> (-0.26%)`	⬇️
setup.py	`0.00% <0.00%> (ø)`
...ks/python/apache_beam/ml/gcp/recommendations_ai.py	`87.26% <0.00%> (ø)`
...n/apache_beam/ml/gcp/recommendations_ai_test_it.py	`69.76% <0.00%> (ø)`
sdks/python/apache_beam/runners/common.py	`89.32% <0.00%> (+0.14%)`	⬆️
...eam/runners/interactive/interactive_environment.py	`90.70% <0.00%> (+0.37%)`	⬆️
.../apache_beam/runners/direct/transform_evaluator.py	`90.44% <0.00%> (+0.65%)`	⬆️
...ks/python/apache_beam/runners/worker/data_plane.py	`92.42% <0.00%> (+1.81%)`	⬆️
... and 1 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update b86fcf9...d315d57. Read the comment docs.

InigoSJ · 2021-06-30T14:02:41Z

Run PythonDocker PreCommit

pabloem

Thanks @InigoSJ - this looks great. I added a few comments.

sdks/python/apache_beam/io/fileio.py

pabloem · 2021-06-30T19:53:02Z

sdks/python/apache_beam/io/fileio_test.py

+    # Add file name that will be created mid-pipeline
+    files.append(FileSystems.join(tempdir, 'extra'))
+
+    interval = 1


maybe make a shorter interval so the test will run faster? I suppose you can't run this with a TestStream? Have you tried adding a TestStream that moves time immediately?

Yeah, I could make it shorter no problem. Regarding TestStream, I am not sure how this would be implemented here, since MatchContinuously needs a PBegin as "input" (since it uses PeriodicImpulse)

To add, PeriodicImpulse tests, use TestPipeline and not TestStream, I guess it should be the same here

sdks/python/apache_beam/io/fileio_test.py

pabloem · 2021-06-30T19:55:19Z

sdks/python/apache_beam/io/fileio.py

+  """Checks for new files for a given pattern every interval.
+
+  This ``PTransform`` returns a ``PCollection`` of matching files in the form
+  of ``FileMetadata`` objects.


Java has an EmptyMatchTreatment. Should we add that here too? https://beam.apache.org/releases/javadoc/2.30.0/org/apache/beam/sdk/io/fs/EmptyMatchTreatment.html

I thought about, but I am not sure this would have an actual use case. Not allowing EmptyMatches would produce an error, and since this is a streaming pipeline, that errors would be retried infinitely.

https://beam.apache.org/releases/pydoc/2.30.0/_modules/apache_beam/io/fileio.html#EmptyMatchTreatment

MatchAll has the option to add it and wouldn't be hard, but I think it would translate in the PTransform being less resilient. What do you think?

hmmm I think that's fair. Let's leave it as it is for now.

pabloem · 2021-06-30T20:44:57Z

Run Portable_Python PreCommit

pabloem · 2021-06-30T20:45:05Z

Run Python_PVR_Flink PreCommit

InigoSJ · 2021-06-30T21:12:29Z

Run Portable_Python PreCommit

InigoSJ · 2021-06-30T22:18:29Z

Run Portable_Python PreCommit

pabloem · 2021-06-30T23:07:45Z

Run Portable_Python PreCommit

pabloem · 2021-06-30T23:19:02Z

Run Portable_Python PreCommit

InigoSJ · 2021-07-01T07:53:05Z

Run Portable_Python PreCommit

pabloem · 2021-07-01T16:44:34Z

Run Portable_Python PreCommit

Add MatchContinuously PTransform

a28039d

InigoSJ added 2 commits June 30, 2021 14:13

Add MatchContinuously PTransform

52198a5

Add MatchContinuously PTransform

4e3db91

pabloem reviewed Jun 30, 2021

View reviewed changes

Add MatchContinuously PTransform

d315d57

pabloem merged commit 22ec4e6 into apache:master Jul 1, 2021

InigoSJ mentioned this pull request Jul 29, 2021

Refactor _RemoveDuplicates to use counter state #15242

Merged

4 tasks

Conversation

InigoSJ commented Jun 30, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

Uh oh!

InigoSJ commented Jun 30, 2021

Uh oh!

codecov bot commented Jun 30, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

InigoSJ commented Jun 30, 2021

Uh oh!

pabloem left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pabloem Jun 30, 2021

Choose a reason for hiding this comment

Uh oh!

InigoSJ Jun 30, 2021

Choose a reason for hiding this comment

Uh oh!

InigoSJ Jun 30, 2021

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pabloem Jun 30, 2021

Choose a reason for hiding this comment

Uh oh!

InigoSJ Jun 30, 2021

Choose a reason for hiding this comment

Uh oh!

pabloem Jun 30, 2021

Choose a reason for hiding this comment

Uh oh!

pabloem commented Jun 30, 2021

Uh oh!

pabloem commented Jun 30, 2021

Uh oh!

InigoSJ commented Jun 30, 2021

Uh oh!

InigoSJ commented Jun 30, 2021

Uh oh!

pabloem commented Jun 30, 2021

Uh oh!

pabloem commented Jun 30, 2021

Uh oh!

InigoSJ commented Jul 1, 2021

Uh oh!

pabloem commented Jul 1, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

InigoSJ commented Jun 30, 2021 •

edited

Loading

`ValidatesRunner` compliance status (on master branch)

codecov bot commented Jun 30, 2021 •

edited

Loading