You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/08/24 18:25:41 UTC
[GitHub] [beam] pabloem opened a new pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
pabloem opened a new pull request #15378:
URL: https://github.com/apache/beam/pull/15378
**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).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon?subject=V1+Streaming">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon?subject=V1+Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon?subject=V2+Streaming">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon?subject=Java+8">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon?subject=Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon?subject=Portable+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon?subject=Structured+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon?subject=ValCont">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon?subject=V1+Java11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon?subject=3.7">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon?subject=3.8">
</a>
</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon?subject=Tests">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
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.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hughack commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
hughack commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r726716161
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
Additionally, should it refer to stages or elements? When building a pipeline you don't necessarily know what stages will result from any fusion that occurs, so would you specify a requirement for key-ordered delivery between elements rather than stages?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem edited a comment on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem edited a comment on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906690959
- Flink streaming VR tests: [Build Scan](https://scans.gradle.com/s/x4whlt2gchcum/tests#:runners:flink:1.13:job-server-validatesportablerunnerstreaming-org.apache.beam.sdk.transforms-perkeyorderingtest)
- Flink Bartch VR tests: [Jenkins output](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch_PR/261/console) - not running any of the new tests, as it does not support key-ordered delivery.
- Dataflow legacy VR tests: [Build Scan](https://scans.gradle.com/s/pcww3rqmcmu24/tests#:runners:google-cloud-dataflow-java-validatesrunnerlegacyworkertest-org.apache.beam.sdk.transforms-perkeyorderingtest)
- Direct VR tests: [Build Scan](https://scans.gradle.com/s/45pjufvncnoq6/tests#:runners:direct-java-validatesrunner-org.apache.beam.sdk.transforms-perkeyorderingtest)
- Samza tests:[Build Scan]( https://scans.gradle.com/s/npxbjepkaliqk/tests#:runners:samza-validatesrunner-org.apache.beam.sdk.transforms-perkeyorderingtest)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906690959
- Flink streaming VR tests: [Build Scan](https://scans.gradle.com/s/x4whlt2gchcum/tests#:runners:flink:1.13:job-server-validatesportablerunnerstreaming-org.apache.beam.sdk.transforms-perkeyorderingtest)
- Dataflow legacy VR tests: [Build Scan](https://scans.gradle.com/s/pcww3rqmcmu24/tests#:runners:google-cloud-dataflow-java-validatesrunnerlegacyworkertest-org.apache.beam.sdk.transforms-perkeyorderingtest)
- Direct VR tests: [Build Scan](https://scans.gradle.com/s/45pjufvncnoq6/tests#:runners:direct-java-validatesrunner-org.apache.beam.sdk.transforms-perkeyorderingtest)
- Samza tests:[Build Scan]( https://scans.gradle.com/s/npxbjepkaliqk/tests#:runners:samza-validatesrunner-org.apache.beam.sdk.transforms-perkeyorderingtest)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905164726
Run Dataflow ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906091108
Run Java 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hughack commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
hughack commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r726715529
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -50,6 +50,11 @@ ways, such as:
This may allow the runner to avoid serializing elements; instead, the runner
can just pass the elements in memory.
+Passing elements between transforms that are running on the same worker is
Review comment:
These definitions are slightly different to the glossary ones for [fusion](https://beam.apache.org/documentation/glossary/#fusion) and [stage](https://beam.apache.org/documentation/glossary/#stage). Maybe just have a link in the last bullet point above saying runners might use a strategy called fusion which combines elements into stages?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem removed a comment on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem removed a comment on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906103802
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem merged pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem merged pull request #15378:
URL: https://github.com/apache/beam/pull/15378
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1013586830
> > > We've converted portable Flink VR to a precommit already, I'll follow up with other jobs later.
> >
> >
> > @ibzib, did the Spark VR also become a precommit?
>
> See my comments on #16345
Ack. Thank you.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-921109352
Run Website_Stage_GCS 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r718806316
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
I was thinking about this comment and it occurs to me that you have to have key-limited parallelism in the producer of course. I don't think this is required anywhere in the model (state/timers require key+window limited parallelism) and also not mentioned 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hughack commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
hughack commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-967766895
@pabloem - Yeah looks good, the only comment would be around @kennknowles point. I don't think "key-limited parallelism" is mentioned anywhere else in the Beam model. My understanding is it does happen, despite not explicitly being documented anywhere i can see. I'd be interested how Dataflow handles scaling the number of workers and if the guarantees still hold for any key that starts getting handled by a different worker.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905164694
Run Java Flink PortableValidatesRunner Streaming
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906098204
Run Dataflow ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-904896801
Run Java Flink PortableValidatesRunner Streaming
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698745582
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
+ Collections.shuffle(perKeyElements);
+ List<String> allKeys =
+ Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).stream()
+ .map(elm -> String.format("k%s", elm))
+ .collect(Collectors.toList());
+ PCollection<KV<String, Integer>> kvSeeds =
Review comment:
I don't think you'll be able to parameterize this, you'll need different test methods for each type.
I believe runners see these as byte[] today, so the behavior here actually depends on coders, so changes in the coder will change the ordering. (Future relational work will likely make some runners aware of the coder contents as well.) The byte coder is the trivial passthrough type, so `Byte` might actually be the best for a basic "does the runner produce deterministic ordering" test.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1005379795
@pabloem @ibzib - Should this be reverted? (For reference https://issues.apache.org/jira/browse/BEAM-13522 is the tracker.)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r718806316
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
I was thinking about this comment and it occurs to me that you have to have key-limited parallelism in the producer of course. I don't think this is required anywhere in the model (state/timers require key+window limited parallelism) and also not mentioned 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905929710
Run Dataflow ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698829954
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesPerKeyOrderInStage.java
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing;
+
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Category tag for validation tests which rely on a runner providing per-key ordering in between
+ * transforms in the same stage. Tests tagged with {@link UsesPerKeyOrderInStage} should be run for
Review comment:
"Stage" is not a concept in the Beam model, unless you also define it elsewhere in this PR. (commenting as I go)
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesPerKeyOrderedDelivery.java
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing;
+
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Category tag for validation tests which rely on a runner providing per-key ordering. Tests tagged
+ * with {@link UsesPerKeyOrderedDelivery} should be run for runners which support key-to-key
+ * ordering of elements across shuffle / stage boundaries.
Review comment:
"shuffle" is also not a model concept. For example runners can choose to move shuffles around so then the ordered delivery might change. Unless you can demonstrate/require that allowable optimizations preserve the ordered delivery you are looking for.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -626,7 +627,7 @@ public void process(OutputReceiver<String> r, PipelineOptions options) {
}
@Test
- @Category(ValidatesRunner.class)
+ @Category({ValidatesRunner.class, UsesParDoLifecycle.class})
Review comment:
Seems unrelated.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
Java `assert` is, I think, less preferable than a library that generates more elaborate error messages. Also, failures within a user `DoFn` may be swallowed by a runner. Can the tests be described in terms of a dead-letter output and `PAssert`?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1006992749
I agree with @lukecwik, and @ibzib's proposal sounds like a good idea to enforce that.
@ibzib - How do we proceed with your proposal? Are you planning to do that? Are you waiting for input on the proposal?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hughack commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
hughack commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r726712021
##########
File path: website/www/site/data/capability_matrix.yaml
##########
@@ -1704,3 +1704,42 @@ capability-matrix:
l1:
l2:
l3:
+ - name: Key-ordered delivery
+ description: The runner offers guarantees for the order in which elements are passed in between operations. See <a href=''>per-key ordering semantics.</a>
+ values:
+ - class: dataflow
+ l1: "Partially"
+ l2:
+ l3: Dataflow performs different shuffling algorithms for batch and streaming. Dataflow guarantees key-ordered delivery in streaming, though not in batch.
Review comment:
Is there any reference that this is true? It is something i am trying to figure out at the moment.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r724797638
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
What I meant was that if we we have a chain of transforms as follows:
```
A -> B -> C
```
if all these transforms are stateful transforms, then there is no guarantee for ordering of elements emitted from A arriving at C. The only exception would be when there is no change in key, because then we can prove that the ordering will be preserved due to transitivity. If key between A and B changes, then there is no guarantee for ordering at C (even if the key changes back to the same as emitted from A).
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905201535
Run Java Flink PortableValidatesRunner Streaming
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905193956
Run Dataflow PortabilityApi ValidatesRunner
--
<br class="Apple-interchange-newline">
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem removed a comment on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem removed a comment on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905926642
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905198140
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905951592
Run Direct ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905193043
Run Java Spark PortableValidatesRunner Streaming
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698714753
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
Sorry if my comment wasn't clear, here is what I'm thinking: the case above is `mached < this.perKeyElements.size() && !...` the assert triggers on `!...`, so the assert actually triggers on `matched >= this.perKeyElements.size()`.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
After the latest update you should never be able to hit this assert, you can leave it as is.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
+ Collections.shuffle(perKeyElements);
+ List<String> allKeys =
+ Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).stream()
+ .map(elm -> String.format("k%s", elm))
+ .collect(Collectors.toList());
+ PCollection<KV<String, Integer>> kvSeeds =
Review comment:
I don't think you'll be able to parameterize this, you'll need different test methods for each type.
I believe runners see these as byte[] today, so the behavior here actually depends on coders, so changes in the coder will change the ordering. (Future relational work will likely make some runners aware of the coder contents as well.) The byte coder is the trivial passthrough type, so `Byte` might actually be the best for a basic "does the runner produce deterministic ordering" test.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,290 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ return;
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else if (matched >= this.perKeyElements.size()) {
+ throw new RuntimeException("Got more elements than expected!");
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
Review comment:
Sorry for the late addition: You might want to test some more extreme values. VarIntCoder is variable sized, so around those size boundaries might be good to test. https://github.com/apache/beam/blob/50fcf55d60e8fa7a8399d63e030c930a2d45402a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/VarIntCoder.java#L29
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-938138864
@pabloem - Could you respond to the open comments?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906572854
Run Java Flink PortableValidatesRunner Batch
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906103891
Run Direct ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698703596
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
the case above is denied (`!...`?)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698776278
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,290 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ return;
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else if (matched >= this.perKeyElements.size()) {
+ throw new RuntimeException("Got more elements than expected!");
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
Review comment:
I tried that. All good.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905946469
Run Java Flink PortableValidatesRunner Batch
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r697686751
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
nit: I don't think this assert is going to work, it is the case above?
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
Review comment:
It appears `matched >= this.perKeyElements.size()` should be its own "Got more elements than expected" error case and throw an exception.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
+ Collections.shuffle(perKeyElements);
+ List<String> allKeys =
+ Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).stream()
+ .map(elm -> String.format("k%s", elm))
+ .collect(Collectors.toList());
+ PCollection<KV<String, Integer>> kvSeeds =
Review comment:
Should we test types other than `Integer`?
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
Review comment:
nit: Possibly make this explicit with a `return` as well?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698726328
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
After the latest update you should never be able to hit this assert, you can leave it as is.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698829954
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesPerKeyOrderInStage.java
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing;
+
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Category tag for validation tests which rely on a runner providing per-key ordering in between
+ * transforms in the same stage. Tests tagged with {@link UsesPerKeyOrderInStage} should be run for
Review comment:
"Stage" is not a concept in the Beam model, unless you also define it elsewhere in this PR. (commenting as I go)
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesPerKeyOrderedDelivery.java
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing;
+
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Category tag for validation tests which rely on a runner providing per-key ordering. Tests tagged
+ * with {@link UsesPerKeyOrderedDelivery} should be run for runners which support key-to-key
+ * ordering of elements across shuffle / stage boundaries.
Review comment:
"shuffle" is also not a model concept. For example runners can choose to move shuffles around so then the ordered delivery might change. Unless you can demonstrate/require that allowable optimizations preserve the ordered delivery you are looking for.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -626,7 +627,7 @@ public void process(OutputReceiver<String> r, PipelineOptions options) {
}
@Test
- @Category(ValidatesRunner.class)
+ @Category({ValidatesRunner.class, UsesParDoLifecycle.class})
Review comment:
Seems unrelated.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
Java `assert` is, I think, less preferable than a library that generates more elaborate error messages. Also, failures within a user `DoFn` may be swallowed by a runner. Can the tests be described in terms of a dead-letter output and `PAssert`?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1006996158
We've converted portable Flink VR to a precommit already, I'll follow up with other jobs later.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1013338703
> We've converted portable Flink VR to a precommit already, I'll follow up with other jobs later.
@ibzib, did the Spark VR also become a 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698714753
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
Sorry if my comment wasn't clear, here is what I'm thinking: the case above is `mached < this.perKeyElements.size() && !...` the assert triggers on `!...`, so the assert actually triggers on `matched >= this.perKeyElements.size()`.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905193142
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1006879983
I don't think it should be reverted but better care around adding new test categories without verifying them should be done in the future.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906756656
Run Java Spark PortableValidatesRunner Batch
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905926589
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905194058
Run Dataflow PortabilityApi ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905946253
Run Dataflow ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem removed a comment on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem removed a comment on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905193956
Run Dataflow PortabilityApi ValidatesRunner
--
<br class="Apple-interchange-newline">
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906768168
Run Java Flink PortableValidatesRunner Batch
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-904892616
Run Dataflow ValidatesRunner
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r747900044
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
Hi all! Sorry about the delay, but I've finally tried to address these comments. Can you please take a look and LMK what you think>
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesPerKeyOrderedDelivery.java
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing;
+
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Category tag for validation tests which rely on a runner providing per-key ordering. Tests tagged
+ * with {@link UsesPerKeyOrderedDelivery} should be run for runners which support key-to-key
+ * ordering of elements across shuffle / stage boundaries.
Review comment:
renamed to bundle and across bundle boundaries. WDYT?
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -50,6 +50,11 @@ ways, such as:
This may allow the runner to avoid serializing elements; instead, the runner
can just pass the elements in memory.
+Passing elements between transforms that are running on the same worker is
Review comment:
done
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesPerKeyOrderInStage.java
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing;
+
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Category tag for validation tests which rely on a runner providing per-key ordering in between
+ * transforms in the same stage. Tests tagged with {@link UsesPerKeyOrderInStage} should be run for
Review comment:
renamed this to bundle. WDYT?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-938138864
@pabloem - Could you respond to the open comments?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
apilloud commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-907386800
Generated website changes are here: https://apache-beam-website-pull-requests.storage.googleapis.com/15378/documentation/runners/capability-matrix/index.html
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698723637
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
Review comment:
done
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906104079
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905946288
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hughack commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
hughack commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r748669254
##########
File path: website/www/site/data/capability_matrix.yaml
##########
@@ -1704,3 +1704,42 @@ capability-matrix:
l1:
l2:
l3:
+ - name: Key-ordered delivery
+ description: The runner offers guarantees for the order in which elements are passed in between operations. See <a href=''>per-key ordering semantics.</a>
+ values:
+ - class: dataflow
+ l1: "Partially"
+ l2:
+ l3: Dataflow performs different shuffling algorithms for batch and streaming. Dataflow guarantees key-ordered delivery in streaming, though not in batch.
Review comment:
Ok, i guess it makes sense to define the concept here first.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-970700349
@kennknowles PTAL : )
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1006885097
I've proposed making Flink/Spark validates runner tests precommits for `runners/flink` and `runners/spark` file changes respectively, as well as `sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/` since often new VR tests are added there without checking if they pass on these runners. https://issues.apache.org/jira/browse/BEAM-13521
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r747900422
##########
File path: website/www/site/data/capability_matrix.yaml
##########
@@ -1704,3 +1704,42 @@ capability-matrix:
l1:
l2:
l3:
+ - name: Key-ordered delivery
+ description: The runner offers guarantees for the order in which elements are passed in between operations. See <a href=''>per-key ordering semantics.</a>
+ values:
+ - class: dataflow
+ l1: "Partially"
+ l2:
+ l3: Dataflow performs different shuffling algorithms for batch and streaming. Dataflow guarantees key-ordered delivery in streaming, though not in batch.
Review comment:
hm we have not stated this in Dataflow documentation (I'm working on that), but it's true for streaming.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-981966843
@kennknowles PTAL - if no other comments by the end of the week, I'll merge form lazy consensus.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1013545165
> > We've converted portable Flink VR to a precommit already, I'll follow up with other jobs later.
>
> @ibzib, did the Spark VR also become a precommit?
See my comments on #16345
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-1003473470
https://github.com/apache/beam/pull/15378 broke spark PVR as well: https://ci-beam.apache.org/job/beam_PreCommit_Java_PVR_Spark_Batch_Cron/5140/ (build 5139 passes)
This Jenkins Job was renamed and is still [failing]( https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/32/console).
PVR tests sickbayed in https://github.com/apache/beam/pull/16411
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698745771
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,290 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ return;
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else if (matched >= this.perKeyElements.size()) {
+ throw new RuntimeException("Got more elements than expected!");
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
Review comment:
Sorry for the late addition: You might want to test some more extreme values. VarIntCoder is variable sized, so around those size boundaries might be good to test. https://github.com/apache/beam/blob/50fcf55d60e8fa7a8399d63e030c930a2d45402a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/VarIntCoder.java#L29
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698726049
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
+ Collections.shuffle(perKeyElements);
+ List<String> allKeys =
+ Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).stream()
+ .map(elm -> String.format("k%s", elm))
+ .collect(Collectors.toList());
+ PCollection<KV<String, Integer>> kvSeeds =
Review comment:
hmmm ... I could try to parameterize the test with different types -though IDK if the `Parameterized` JUnit runner integrates well with the ValidatesRunner framework. Do you have a suggestion? Should I try adding a more complex type instead of just ints?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-906103802
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r718268697
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
Although it seems to be mentioned I think it would be good to restate here explicitly, that this holds if and only if the two PTransforms are directly connected - or at least there is no intermediate grouping transform in between. "A downstream PCollection" in the above paragraph might be interpreted as *any* downstream PCollection, which does not hold.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem removed a comment on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem removed a comment on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905193184
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [WIP] Starting to define per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-905946611
Run Java Spark PortableValidatesRunner Batch
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r698703596
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
Review comment:
the case above is denied (`!...`?)
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
Review comment:
done
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
Review comment:
done.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
+ Collections.shuffle(perKeyElements);
+ List<String> allKeys =
+ Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).stream()
+ .map(elm -> String.format("k%s", elm))
+ .collect(Collectors.toList());
+ PCollection<KV<String, Integer>> kvSeeds =
Review comment:
hmmm ... I could try to parameterize the test with different types -though IDK if the `Parameterized` JUnit runner integrates well with the ValidatesRunner framework. Do you have a suggestion? Should I try adding a more complex type instead of just ints?
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java
##########
@@ -0,0 +1,290 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderInStage;
+import org.apache.beam.sdk.testing.UsesPerKeyOrderedDelivery;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(JUnit4.class)
+public class PerKeyOrderingTest implements Serializable {
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static final Logger LOG = LoggerFactory.getLogger(PerKeyOrderingTest.class);
+
+ private static class VerifyDoFn<T> extends DoFn<KV<String, T>, KV<String, Boolean>> {
+
+ private final List<T> perKeyElements;
+
+ VerifyDoFn(List<T> perKeyElements) {
+ this.perKeyElements = perKeyElements;
+ }
+
+ @StateId("matchedElements")
+ private final StateSpec<ValueState<Integer>> elementsMatchedState = StateSpecs.value();
+
+ @ProcessElement
+ public void verifyOrder(
+ @Element KV<String, T> elm,
+ OutputReceiver<KV<String, Boolean>> receiver,
+ @StateId("matchedElements") ValueState<Integer> matchedElements) {
+ Integer matched = matchedElements.read();
+ matched = matched == null ? 0 : matched;
+ if (matched == -1) {
+ // When matched is set to -1, it means that we have met an error, and elements on this
+ // key are not matched anymore - thus we ignore all inputs.
+ return;
+ } else if (matched < this.perKeyElements.size()
+ && !this.perKeyElements.get(matched).equals(elm.getValue())) {
+ // If we meet this condition, then the order of elements is not what we're expecting.
+ // We mark `matched` as -1, and output a failed ordering.
+ matchedElements.write(-1);
+ receiver.output(KV.of(elm.getKey(), false));
+ } else if (matched >= this.perKeyElements.size()) {
+ throw new RuntimeException("Got more elements than expected!");
+ } else {
+ assert this.perKeyElements.get(matched).equals(elm.getValue())
+ : String.format("Element %s is not expected %s", elm, this.perKeyElements.get(matched));
+ matchedElements.write(matched + 1);
+ // If we reached the end of perKeyElements, it means that all elements have been emitted in
+ // the expected order, and thus we mark this Key as successful.
+ if (matched + 1 == perKeyElements.size()) {
+ receiver.output(KV.of(elm.getKey(), true));
+ }
+ }
+ }
+ }
+
+ @Test
+ @Category({ValidatesRunner.class, UsesPerKeyOrderedDelivery.class})
+ public void testSingleCallOrderingWithShuffle() {
+ // Here we test that the output of a single process call in a DoFn will be output in order
+ List<Integer> perKeyElements =
+ Lists.newArrayList(-8, -7, -6, -5, -4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8);
Review comment:
I tried that. All good.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r718268697
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
Although it seems to be mentioned I think it would be good to restate here explicitly, that this holds if and only if the two PTransforms are directly connected - or at least there is no intermediate grouping transform in between. "A downstream PCollection" in the above paragraph might be interpreted as *any* downstream PCollection, which does not hold.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15378:
URL: https://github.com/apache/beam/pull/15378#issuecomment-985714964
moving forward as-is, but this is an area of active discussion, so please feel free to engage on this, and we cna work to keep this up to date.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hughack commented on a change in pull request #15378: [RFC] Define and document per-key ordering semantics for runners
Posted by GitBox <gi...@apache.org>.
hughack commented on a change in pull request #15378:
URL: https://github.com/apache/beam/pull/15378#discussion_r726711691
##########
File path: website/www/site/content/en/documentation/runtime/model.md
##########
@@ -77,6 +82,48 @@ element, and having to retry everything if there is a failure. For example, a
streaming runner may prefer to process and commit small bundles, and a batch
runner may prefer to process larger bundles.
+### Data partitioning and inter-stage execution
+
+Partitioning and parallelization of element processing within a Beam pipeline is
+dependent on two things:
+
+- Data source implementation
+- Inter-stage key parallelism
+
+Beam pipelines read data from a source (e.g. `KafkaIO`, `BigQueryIO`, `JdbcIO`,
+or your own source implementation). To implement a Source in Beam one must
+implement it as a Splittable `DoFn`. A Splittable `DoFn` provides the runner
+with interfaces to facilitate the splitting of work.
+
+When running key-based operations in Beam (e.g. `GroupByKey`, `Combine`,
+`Reshuffle.perKey`, and stateful `DoFn`s), Beam runners perform serialization
+and transfer of data known as *shuffle*<sup>1</sup>. Shuffle allows data
+elements of the same key to be processed together.
+
+The way in which runners *shuffle* data may be slightly different for Batch and
+Streaming execution modes.
+
+<sup>1</sup>Not to be confused with the `shuffle` operation in some runners.
+
+#### Data ordering in a pipeline execution
+The Beam model does not define strict guidelines regarding the order in which
+runners process elements or transport them across `PTransforms`. Runners are
+free to implement shuffling semantics in different forms.
+
+Some use cases exist where user pipelines may need to rely on specific ordering
+semantics in pipeline execution. The [capability matrix documents](/documentation/runners/capability-matrix/additional-common-features-not-yet-part-of-the-beam-model/index.html)
+runner behavior for **key-ordered delivery**.
+
+Consider a single Beam worker processing a series of bundles from the same Beam
+stage, and consider a `PTransform` that outputs data from this Stage into a
+downstream `PCollection`. Finally, consider two events *with the same key*
+emitted in a certain order by this worker (within the same bundle or as part of
+different bundles).
+
+We say that the Beam runner supports **key-ordered delivery** if it guarantees
+that these two events will be observed downstream in the same order,
+independently of the kind of transmission.
Review comment:
Agree with both the above points. Should it explicitly say the following?
- Key-ordered deliver is only guaranteed between immediately connected elements.
- To enforce key ordering both producer and consumer require key limited parallelism.
Or are these statements not accurate?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org