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/04/01 21:52:11 UTC
[GitHub] [beam] amaliujia opened a new pull request #14404: [BEAM-12083] Nexmark Query 13.
amaliujia opened a new pull request #14404:
URL: https://github.com/apache/beam/pull/14404
In portability world, we expect on workers there are runners and SDKs. Current Nexmark queries, e.g. Nexmark Query 0, does not stress the boundary between runner and SDK.
So I propose to have Nexmark query 13, which contains a source + GBK + ParDo, to stress the boundary between runner and SDK.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>SDK</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>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<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>---</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/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_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">
</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">
</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">
</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">
</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">
</a>
<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">
</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">
</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">
</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">
</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">
</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">
</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">
</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>
</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">
</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">
</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>
<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">
</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">
</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">
</a>
</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">
</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">
</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">
</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">
</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>---</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>---</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>---</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>
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">
</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">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img src="https://camo.githubusercontent.com/4565d7b7e907114e6c1d12323408bd903aa252fefed5eeab93701b05c9628a84/68747470733a2f2f63692d6265616d2e6170616368652e6f72672f6a6f622f6265616d5f507265436f6d6d69745f507974686f6e446f636b65725f43726f6e2f62616467652f69636f6e" alt="Build Status" data-canonical-src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img src="https://camo.githubusercontent.com/21afb097a5745440598bee5c59a027b140585eec871c3f3b883200610fabf722/68747470733a2f2f63692d6265616d2e6170616368652e6f72672f6a6f622f6265616d5f507265436f6d6d69745f507974686f6e446f63735f43726f6e2f62616467652f69636f6e" alt="Build Status" data-canonical-src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon">
</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>---</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-814487183
I guess https://ci-beam.apache.org/job/beam_PostCommit_Java_Nexmark_Spark/ is not running the same stuff.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812921606
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-814486981
Breaks all of them I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r605979222
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "bytes");
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ coder.encode(c.element(), outStream, Coder.Context.OUTER);
Review comment:
I doubt that a benchmark will be useful if this ParDo do nothing. CPU might be far under full utilization in that case.
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "bytes");
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ coder.encode(c.element(), outStream, Coder.Context.OUTER);
Review comment:
I doubt that a benchmark will be useful if this ParDo does nothing. CPU might be far under full utilization in that case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] udim commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
udim commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-813706652
This may have broken beam_PostCommit_Java_Nexmark_Direct:
```
11:13:46 Caused by: java.lang.IllegalStateException: GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without a trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812201862
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812692941
@y1chi I added a parameter to control how much CPU intensive work that the ParDo will do.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] y1chi commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r606422856
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "serde-bytes");
+ private long elementCount = 0;
+ private long opFrequency =
+ (configuration.pardoCPUFactor > 0.0 && configuration.pardoCPUFactor < 1.0)
+ ? (long) (1.0 / configuration.pardoCPUFactor)
+ : 1L;
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ elementCount++;
+ Event event;
+ if (configuration.pardoCPUFactor != 0
+ && (opFrequency == 1L || elementCount % opFrequency == 0L)) {
Review comment:
how about using Random.nextDouble() compared with pardoCPUFactor?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812274687
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia closed pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia closed pull request #14404:
URL: https://github.com/apache/beam/pull/14404
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812216127
Run Java_Examples_Dataflow PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812337459
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] y1chi commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r606542827
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Random;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "serde-bytes");
+ private final Random random = new Random();
+ private double pardoCPUFactor =
+ (configuration.pardoCPUFactor >= 0.0 && configuration.pardoCPUFactor <= 1.0)
Review comment:
I think the condition check is probably redundant
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812643604
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r605977936
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "bytes");
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ coder.encode(c.element(), outStream, Coder.Context.OUTER);
Review comment:
This models a CPU intensive workload and the execution on workers have both runner and SDK harness running.
I think having CPU intensive ParDo is more closer to worst case workload. Real world workload could just between the ParDo does nothing and the ParDo uses lots of CPU.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia edited a comment on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia edited a comment on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812191393
R: @y1chi @chamikaramj
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia merged pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia merged pull request #14404:
URL: https://github.com/apache/beam/pull/14404
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r606536862
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "serde-bytes");
+ private long elementCount = 0;
+ private long opFrequency =
+ (configuration.pardoCPUFactor > 0.0 && configuration.pardoCPUFactor < 1.0)
+ ? (long) (1.0 / configuration.pardoCPUFactor)
+ : 1L;
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ elementCount++;
+ Event event;
+ if (configuration.pardoCPUFactor != 0
+ && (opFrequency == 1L || elementCount % opFrequency == 0L)) {
Review comment:
Thanks for the idea. I changed to use Randome.nextDouble() now
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "serde-bytes");
+ private long elementCount = 0;
+ private long opFrequency =
+ (configuration.pardoCPUFactor > 0.0 && configuration.pardoCPUFactor < 1.0)
+ ? (long) (1.0 / configuration.pardoCPUFactor)
+ : 1L;
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ elementCount++;
+ Event event;
+ if (configuration.pardoCPUFactor != 0
+ && (opFrequency == 1L || elementCount % opFrequency == 0L)) {
Review comment:
Thanks for the idea. I changed to use Random.nextDouble() now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812808270
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] y1chi commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r606489007
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "serde-bytes");
+ private long elementCount = 0;
+ private long opFrequency =
+ (configuration.pardoCPUFactor > 0.0 && configuration.pardoCPUFactor < 1.0)
+ ? (long) (1.0 / configuration.pardoCPUFactor)
+ : 1L;
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ elementCount++;
+ Event event;
+ if (configuration.pardoCPUFactor != 0
+ && (opFrequency == 1L || elementCount % opFrequency == 0L)) {
Review comment:
I mean instead of using mod you can just achieve same effect with `if (Random.nextDouble() < configuration.pardoCPUFactor)` right? Plus this is more accurate because you don't have to round up the opFrequency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r606468028
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "serde-bytes");
+ private long elementCount = 0;
+ private long opFrequency =
+ (configuration.pardoCPUFactor > 0.0 && configuration.pardoCPUFactor < 1.0)
+ ? (long) (1.0 / configuration.pardoCPUFactor)
+ : 1L;
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ elementCount++;
+ Event event;
+ if (configuration.pardoCPUFactor != 0
+ && (opFrequency == 1L || elementCount % opFrequency == 0L)) {
Review comment:
Do you mean always use randomization, instead of using a fixed pardoCPUFactor?
If so, I think have a way to control pardoCPUFactor is better at this moment For example, in near term usage, I would expect to always set `pardoCPUFactor` as 1.0 to make the ParDo do CPU intensive work
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r605977936
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "bytes");
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ coder.encode(c.element(), outStream, Coder.Context.OUTER);
Review comment:
This models a CPU intensive workload and the execution on workers have both runner and SDK harness running.
I think having CPU intensive ParDo is more closer to real world workload.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r606614778
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Random;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "serde-bytes");
+ private final Random random = new Random();
+ private double pardoCPUFactor =
+ (configuration.pardoCPUFactor >= 0.0 && configuration.pardoCPUFactor <= 1.0)
Review comment:
I prefer to treat a value outside of 0.0~1.0 to default value because value outside this range seems less meaningful. Though I guess without this check the code will still work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812191393
R: @y1chi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] y1chi commented on a change in pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
y1chi commented on a change in pull request #14404:
URL: https://github.com/apache/beam/pull/14404#discussion_r605974661
##########
File path: sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/Query13.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.nexmark.queries;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.nexmark.NexmarkConfiguration;
+import org.apache.beam.sdk.nexmark.model.Event;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Reshuffle.AssignShardFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Query "13" PORTABILITY_BATCH (not in original suite).
+ *
+ * <p>This benchmark is created to stress the boundary of runner and SDK in portability world. The
+ * basic shape of this benchmark is source + GBK + ParDo, in which the GBK read + ParDo will require
+ * that runner reads from shuffle and connects with SDK to do CPU intensive computation.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
+})
+public class Query13 extends NexmarkQueryTransform<Event> {
+ private final NexmarkConfiguration configuration;
+
+ public Query13(NexmarkConfiguration configuration) {
+ super("Query13");
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollection<Event> expand(PCollection<Event> events) {
+ final Coder<Event> coder = events.getCoder();
+ return events
+ .apply("Pair with random key", ParDo.of(new AssignShardFn<>(configuration.numKeyBuckets)))
+ .apply(GroupByKey.create())
+ .apply(
+ "ExpandIterable",
+ ParDo.of(
+ new DoFn<KV<Integer, Iterable<Event>>, Event>() {
+ @ProcessElement
+ public void processElement(
+ @Element KV<Integer, Iterable<Event>> element, OutputReceiver<Event> r) {
+ for (Event value : element.getValue()) {
+ r.output(value);
+ }
+ }
+ }))
+ // Force round trip through coder.
+ .apply(
+ name + ".Serialize",
+ ParDo.of(
+ new DoFn<Event, Event>() {
+ private final Counter bytesMetric = Metrics.counter(name, "bytes");
+
+ @ProcessElement
+ public void processElement(ProcessContext c) throws CoderException, IOException {
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ coder.encode(c.element(), outStream, Coder.Context.OUTER);
Review comment:
Do we need this encoding/decoding round trip if the query aims to benchmark against the SDK/runner boundary?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-813093681
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-813101273
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] amaliujia commented on pull request #14404: [BEAM-12083] Nexmark Query 13.
Posted by GitBox <gi...@apache.org>.
amaliujia commented on pull request #14404:
URL: https://github.com/apache/beam/pull/14404#issuecomment-812201145
Tested by running
```
./gradlew :sdks:java:testing:nexmark:run -Pnexmark.runner=":runners:direct-java" -Pnexmark.args="
--runner=DirectRunner
--streaming=false
--suite=SMOKE
--manageResources=false
--monitorJobs=true
--enforceEncodability=true
--enforceImmutability=true --query=15"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org