Initial DaskRunner for Beam.

[alxmrs]

Here, I've created a minimum viable Apache Beam runner for Dask. My approach is to visit a Beam Pipeline an translate PCollections into Dask Bags, and PTransformations to Bag methods.

In this version, I have supported enough operations to make test pipeline asserts work. The test themselves are not comprehensive. Further, there are many Bag operations that could be translated for greater efficiency.

CC: @pabloem

Fixes: apache#18962

---

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).
• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• 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.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

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

[cisaacstern]

There's some ambiguity on how we can connect Dask to the runner. One of two options are:

• To create a full Bag graph up front, then call compute.

This conceptually aligned with what the DirectRunner does. AFAICT, it:

1. Instantiates a visitor which is passed into pipeline.visit
2. The visit_transform method of the visitor adds items to a few different iterable attributes of self
3. Those iterables are then passed to the executor in a few places here

• To create pseudo graph structures, and evaluate as we traverse the graph.

I don't fully grok what is meant by "pseudo graph structure" here, but the RayRunner evaluates during its visitation cycle:

1. The visitor and executor are collapsed into a single object
2. The hybrid visitor/executor's visit_transform method executes during visitation, using a translations dict to map beam operations onto their ray equivalents

It looks like 7cbf5fa lays a foundation for following the RayRunner approach, which makes sense to me!

cc @rabernat

[cisaacstern]

Also just thought I'd link the original meeting notes here, so we have all relevant documents in once place:

https://docs.google.com/document/d/1Awj_eNmH-WRSte3bKcCcUlQDiZ5mMKmCO_xV-mHWAak/edit#

[alxmrs]

Hey @cisaacstern! Check this out now -- I've made some good progress today. The beam pipeline doesn't return the proper thing, but after inspecting it in the debugger, it does seem to pass Beam/python operations to Dask Bag executors.

[cisaacstern]

Awesome, @alxmrs! I should have a moment to take a closer look at this later this week.