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 2020/11/20 07:31:53 UTC
[GitHub] [beam] tszerszen opened a new pull request #13388: Go SDK Group by key load tests
tszerszen opened a new pull request #13388:
URL: https://github.com/apache/beam/pull/13388
This PR adds Group By Key load tests for Go 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)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
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] kamilwu merged pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
kamilwu merged pull request #13388:
URL: https://github.com/apache/beam/pull/13388
----------------------------------------------------------------
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] kamilwu commented on a change in pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13388:
URL: https://github.com/apache/beam/pull/13388#discussion_r528580215
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,106 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source. Besides of the standard
+// input options there are additional options:
+// * fanout (optional) - number of GBK operations to run in parallel
+// * iterations (optional) - number of reiterations over per-key-grouped
+// values to perform
+// input_options - options for Synthetic Sources.
+
+// Example test run:
+
+// go run sdks/go/test/load/group_by_key_test/group_by_key.go \
+// --fanout=1
+// --iterations=1
+// --input_options='{
+// \"num_records\": 300,
+// \"key_size\": 5,
+// \"value_size\": 15,
+// \"num_hot_keys\": 30,
+// \"hot_key_fraction\": 0.5
+// }'"
+
+package main
+
+import (
+ "context"
+ "flag"
+
+ "github.com/apache/beam/sdks/go/pkg/beam"
+ "github.com/apache/beam/sdks/go/pkg/beam/io/synthetic"
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/go/test/load"
+)
+
+var (
+ fanout = flag.Int(
+ "fanout",
+ 1,
+ "Fanout")
+ iterations = flag.Int(
+ "iterations",
+ 1,
+ "A number of subsequent ParDo transforms to be performed")
Review comment:
This is not true. Please change the description to:
`A number of reiterations over per-key-grouped values to perform.`
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,106 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source. Besides of the standard
+// input options there are additional options:
+// * fanout (optional) - number of GBK operations to run in parallel
+// * iterations (optional) - number of reiterations over per-key-grouped
+// values to perform
+// input_options - options for Synthetic Sources.
+
+// Example test run:
+
+// go run sdks/go/test/load/group_by_key_test/group_by_key.go \
+// --fanout=1
+// --iterations=1
+// --input_options='{
+// \"num_records\": 300,
+// \"key_size\": 5,
+// \"value_size\": 15,
+// \"num_hot_keys\": 30,
+// \"hot_key_fraction\": 0.5
+// }'"
+
+package main
+
+import (
+ "context"
+ "flag"
+
+ "github.com/apache/beam/sdks/go/pkg/beam"
+ "github.com/apache/beam/sdks/go/pkg/beam/io/synthetic"
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/go/test/load"
+)
+
+var (
+ fanout = flag.Int(
+ "fanout",
+ 1,
+ "Fanout")
+ iterations = flag.Int(
+ "iterations",
+ 1,
+ "A number of subsequent ParDo transforms to be performed")
+ syntheticConfig = flag.String(
+ "input_options",
+ "",
+ "A JSON object that describes the configuration for synthetic source")
+)
+
+func parseSyntheticConfig() synthetic.SourceConfig {
+ if *syntheticConfig == "" {
+ panic("--input_options not provided")
+ } else {
+ encoded := []byte(*syntheticConfig)
+ return synthetic.DefaultSourceConfig().BuildFromJSON(encoded)
+ }
+}
+
+func main() {
+ flag.Parse()
+ beam.Init()
+
+ ctx := context.Background()
+
+ p, s := beam.NewPipelineWithRoot()
+ src := synthetic.SourceSingle(s, parseSyntheticConfig())
+ pcoll := beam.ParDo(s, &load.RuntimeMonitor{}, src)
+ for i := 0; i < *fanout; i++ {
+ pcoll = beam.GroupByKey(s, src)
+ beam.ParDo(s, func(key []uint8, values func(*[]uint8) bool) ([]uint8, []uint8) {
Review comment:
Could you replace `uint8` with `byte`? Those are the same types, but the ParDo test uses `byte`, and we'd like to keep it consistent.
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,106 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source. Besides of the standard
+// input options there are additional options:
+// * fanout (optional) - number of GBK operations to run in parallel
+// * iterations (optional) - number of reiterations over per-key-grouped
+// values to perform
+// input_options - options for Synthetic Sources.
+
+// Example test run:
+
+// go run sdks/go/test/load/group_by_key_test/group_by_key.go \
+// --fanout=1
+// --iterations=1
+// --input_options='{
+// \"num_records\": 300,
+// \"key_size\": 5,
+// \"value_size\": 15,
+// \"num_hot_keys\": 30,
+// \"hot_key_fraction\": 0.5
+// }'"
+
+package main
+
+import (
+ "context"
+ "flag"
+
+ "github.com/apache/beam/sdks/go/pkg/beam"
+ "github.com/apache/beam/sdks/go/pkg/beam/io/synthetic"
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/go/test/load"
+)
+
+var (
+ fanout = flag.Int(
+ "fanout",
+ 1,
+ "Fanout")
Review comment:
There's no need to add `(optional)`, because, by default, every flag in Go is optional
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,106 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source. Besides of the standard
+// input options there are additional options:
+// * fanout (optional) - number of GBK operations to run in parallel
+// * iterations (optional) - number of reiterations over per-key-grouped
+// values to perform
+// input_options - options for Synthetic Sources.
+
+// Example test run:
+
+// go run sdks/go/test/load/group_by_key_test/group_by_key.go \
+// --fanout=1
+// --iterations=1
+// --input_options='{
+// \"num_records\": 300,
+// \"key_size\": 5,
+// \"value_size\": 15,
+// \"num_hot_keys\": 30,
+// \"hot_key_fraction\": 0.5
+// }'"
+
+package main
+
+import (
+ "context"
+ "flag"
+
+ "github.com/apache/beam/sdks/go/pkg/beam"
+ "github.com/apache/beam/sdks/go/pkg/beam/io/synthetic"
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/go/test/load"
+)
+
+var (
+ fanout = flag.Int(
+ "fanout",
+ 1,
+ "Fanout")
Review comment:
Please provide a description what this flag does. For example:
`A number of GroupByKey operations to perform in parallel.`
##########
File path: .test-infra/jenkins/job_LoadTests_GBK_Flink_Go.groovy
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as commonJobProperties
+import CommonTestProperties
+import LoadTestsBuilder as loadTestsBuilder
+import PhraseTriggeringPostCommitBuilder
+import Flink
+import InfluxDBCredentialsHelper
+
+import static LoadTestsBuilder.DOCKER_CONTAINER_REGISTRY
+
+String now = new Date().format('MMddHHmmss', TimeZone.getTimeZone('UTC'))
+
+def batchScenarios = {
+ [
+ [
+ title : 'Load test: 2GB of 10B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ input_options : """
+ {
+ "num_records": 200000000,
+ "key_size": 1,
+ "value_size": 9
+ }
+ """.trim().replaceAll("\\s", ""),
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Load test: 2GB of 100B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ input_options : """
+ {
+ "num_records": 20000000,
+ "key_size": 10,
+ "value_size": 90
+ }
+ """.trim().replaceAll("\\s", ""),
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Load test: 2GB of 100kB records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ input_options : """
+ {
+ "num_records": 20000,
+ "key_size": 10000,
+ "value_size": 90000
+ }
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Load test: fanout 4 times with 2GB 10-byte records total',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 1,
+ fanout : 4,
+ parallelism : 16,
+ input_options : """
+ {
+ "num_records": 5000000,
+ "key_size": 10,
+ "value_size": 90
+ }
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Load test: fanout 8 times with 2GB 10-byte records total',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 1,
+ fanout : 8,
+ parallelism : 16,
+ input_options : """
+ {
+ "num_records": 5000000,
+ "key_size": 10,
+ "value_size": 90
+ }
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Load test: reiterate 4 times 10kB values',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 4,
+ fanout : 1,
+ parallelism : 5,
+ input_options : """
+ {
+ "num_records": 5000000,
+ "key_size": 10,
+ "value_size": 90,
+ "num_hot_keys": 200,
+ "hot_key_fraction": 1
+ }
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Load test: reiterate 4 times 2MB values',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 4,
+ fanout : 1,
+ parallelism : 5,
+ input_options : """
+ {
+ "num_records": 20000000,
+ "key_size": 10,
+ "value_size": 90,
+ "num_hot_keys": 10,
+ "hot_key_fraction": 1
+ }
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ ].each { test -> test.pipelineOptions.putAll(additionalPipelineArgs) }
+}
+
+def loadTestJob = { scope, triggeringContext, mode ->
+ def numberOfWorkers = 5
Review comment:
Some of your tests need the parallelism of 16, but your Flink deployment is going to have only 5 task managers. You need to group your tests by parallelism, execute the first group, then rescale the cluster and execute the second group. Please take a look at `.test-infra/jenkins/job_LoadTests_Combine_Flink_Python.groovy` where such a pattern is being used.
##########
File path: .test-infra/jenkins/job_LoadTests_GBK_Flink_Go.groovy
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as commonJobProperties
+import CommonTestProperties
+import LoadTestsBuilder as loadTestsBuilder
+import PhraseTriggeringPostCommitBuilder
+import Flink
+import InfluxDBCredentialsHelper
+
+import static LoadTestsBuilder.DOCKER_CONTAINER_REGISTRY
+
+String now = new Date().format('MMddHHmmss', TimeZone.getTimeZone('UTC'))
+
+def batchScenarios = {
+ [
+ [
+ title : 'Load test: 2GB of 10B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
Review comment:
Please add `influx_namespace: 'flink'` for every scenario.
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,106 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source. Besides of the standard
+// input options there are additional options:
+// * fanout (optional) - number of GBK operations to run in parallel
+// * iterations (optional) - number of reiterations over per-key-grouped
+// values to perform
+// input_options - options for Synthetic Sources.
+
+// Example test run:
+
+// go run sdks/go/test/load/group_by_key_test/group_by_key.go \
Review comment:
It's good to copy-and-paste such examples to check if everything's alright. It seems something's wrong here:
`panic: Could not unmarshal SourceConfig: invalid character '\\' looking for beginning of object key string`
To be honest, I have mixed feelings about such examples. They can be useful in shell scripts, where there are plenty of positional arguments. But in this case, all arguments are named, and only one of them is required. What's more, those examples can easily become outdated, because it's easy to miss them while editing the file. That happened once to Python load tests.
If you decide to keep it, I guess you will have to remove backslashes to make it work.
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,106 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source. Besides of the standard
+// input options there are additional options:
+// * fanout (optional) - number of GBK operations to run in parallel
Review comment:
I think we don't need to write out every possible flag here. If someone invokes the test with `--help` flag or makes a mistake, they will see all available flags alongside their descriptions, which are defined below (starting at line 49)
##########
File path: .test-infra/jenkins/job_LoadTests_GBK_Flink_Go.groovy
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as commonJobProperties
+import CommonTestProperties
+import LoadTestsBuilder as loadTestsBuilder
+import PhraseTriggeringPostCommitBuilder
+import Flink
+import InfluxDBCredentialsHelper
+
+import static LoadTestsBuilder.DOCKER_CONTAINER_REGISTRY
+
+String now = new Date().format('MMddHHmmss', TimeZone.getTimeZone('UTC'))
+
+def batchScenarios = {
+ [
+ [
+ title : 'Load test: 2GB of 10B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ input_options : """
+ {
+ "num_records": 200000000,
+ "key_size": 1,
+ "value_size": 9
+ }
+ """.trim().replaceAll("\\s", ""),
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Load test: 2GB of 100B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ input_options : """
+ {
+ "num_records": 20000000,
+ "key_size": 10,
+ "value_size": 90
+ }
+ """.trim().replaceAll("\\s", ""),
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ endpoint : 'localhost:8099',
Review comment:
Please remove one extra space so that colons are in line. The same applies to other scenarios as well.
##########
File path: .test-infra/jenkins/job_LoadTests_GBK_Flink_Go.groovy
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as commonJobProperties
+import CommonTestProperties
+import LoadTestsBuilder as loadTestsBuilder
+import PhraseTriggeringPostCommitBuilder
+import Flink
+import InfluxDBCredentialsHelper
+
+import static LoadTestsBuilder.DOCKER_CONTAINER_REGISTRY
+
+String now = new Date().format('MMddHHmmss', TimeZone.getTimeZone('UTC'))
+
+def batchScenarios = {
+ [
+ [
+ title : 'Load test: 2GB of 10B records',
Review comment:
`Load test: 2GB of 10B records` -> `GroupByKey Go Load test: 2GB of 10B records`
----------------------------------------------------------------
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] tszerszen commented on pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13388:
URL: https://github.com/apache/beam/pull/13388#issuecomment-730961686
R: @kamilwu ?
----------------------------------------------------------------
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] kamilwu commented on a change in pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13388:
URL: https://github.com/apache/beam/pull/13388#discussion_r528684823
##########
File path: .test-infra/jenkins/job_LoadTests_GBK_Flink_Go.groovy
##########
@@ -30,178 +30,196 @@ String now = new Date().format('MMddHHmmss', TimeZone.getTimeZone('UTC'))
def batchScenarios = {
[
[
- title : 'Load test: 2GB of 10B records',
+ title : 'Group By Key Go Load test: 2GB of 10B records',
test : 'group_by_key',
runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
pipelineOptions: [
- job_name : "load-tests-go-flink-batch-gbk-1-${now}",
- influx_measurement : 'go_batch_gbk_1',
- input_options : """
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ input_options : """
{
"num_records": 200000000,
"key_size": 1,
"value_size": 9
}
- """.trim().replaceAll("\\s", ""),
- iterations : 1,
- fanout : 1,
- parallelism : 5,
- endpoint : 'localhost:8099',
- environment_type : 'DOCKER',
- environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ """.trim().replaceAll("\\s", ""),
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
]
],
[
- title : 'Load test: 2GB of 100B records',
+ title : 'Group By Key Go Load test: 2GB of 100B records',
test : 'group_by_key',
runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
pipelineOptions: [
- job_name : "load-tests-go-flink-batch-gbk-1-${now}",
- influx_measurement : 'go_batch_gbk_1',
- input_options : """
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ input_options : """
{
"num_records": 20000000,
"key_size": 10,
"value_size": 90
}
- """.trim().replaceAll("\\s", ""),
- iterations : 1,
- fanout : 1,
- parallelism : 5,
- endpoint : 'localhost:8099',
- environment_type : 'DOCKER',
- environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ """.trim().replaceAll("\\s", ""),
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
]
],
[
- title : 'Load test: 2GB of 100kB records',
+ title : 'Group By Key Go Load test: 2GB of 100kB records',
test : 'group_by_key',
runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
pipelineOptions: [
- job_name : "load-tests-go-flink-batch-gbk-1-${now}",
- influx_measurement : 'go_batch_gbk_1',
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
iterations : 1,
fanout : 1,
parallelism : 5,
- input_options : """
+ input_options : """
{
"num_records": 20000,
"key_size": 10000,
"value_size": 90000
}
- """.trim().replaceAll("\\s", ""),
- endpoint : 'localhost:8099',
- environment_type : 'DOCKER',
- environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
]
],
[
- title : 'Load test: fanout 4 times with 2GB 10-byte records total',
+ title : 'Group By Key Go Load test: fanout 4 times with 2GB 10-byte records total',
test : 'group_by_key',
runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
pipelineOptions: [
- job_name : "load-tests-go-flink-batch-gbk-1-${now}",
- influx_measurement : 'go_batch_gbk_1',
- iterations : 1,
- fanout : 4,
- parallelism : 16,
- input_options : """
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 1,
+ fanout : 4,
+ parallelism : 16,
+ input_options : """
{
"num_records": 5000000,
"key_size": 10,
"value_size": 90
}
- """.trim().replaceAll("\\s", ""),
- endpoint : 'localhost:8099',
- environment_type : 'DOCKER',
- environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
]
],
[
- title : 'Load test: fanout 8 times with 2GB 10-byte records total',
+ title : 'Group By Key Go Load test: fanout 8 times with 2GB 10-byte records total',
test : 'group_by_key',
runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
pipelineOptions: [
- job_name : "load-tests-go-flink-batch-gbk-1-${now}",
- influx_measurement : 'go_batch_gbk_1',
- iterations : 1,
- fanout : 8,
- parallelism : 16,
- input_options : """
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 1,
+ fanout : 8,
+ parallelism : 16,
+ input_options : """
{
"num_records": 5000000,
"key_size": 10,
"value_size": 90
}
- """.trim().replaceAll("\\s", ""),
- endpoint : 'localhost:8099',
- environment_type : 'DOCKER',
- environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
]
],
[
- title : 'Load test: reiterate 4 times 10kB values',
+ title : 'Group By Key Go Load test: reiterate 4 times 10kB values',
test : 'group_by_key',
runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
pipelineOptions: [
- job_name : "load-tests-go-flink-batch-gbk-1-${now}",
- influx_measurement : 'go_batch_gbk_1',
- iterations : 4,
- fanout : 1,
- parallelism : 5,
- input_options : """
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 4,
+ fanout : 1,
+ parallelism : 5,
+ input_options : """
{
"num_records": 5000000,
"key_size": 10,
"value_size": 90,
"num_hot_keys": 200,
"hot_key_fraction": 1
}
- """.trim().replaceAll("\\s", ""),
- endpoint : 'localhost:8099',
- environment_type : 'DOCKER',
- environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
]
],
[
- title : 'Load test: reiterate 4 times 2MB values',
+ title : 'Group By Key Go Load test: reiterate 4 times 2MB values',
test : 'group_by_key',
runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
pipelineOptions: [
- job_name : "load-tests-go-flink-batch-gbk-1-${now}",
- influx_measurement : 'go_batch_gbk_1',
- iterations : 4,
- fanout : 1,
- parallelism : 5,
- input_options : """
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ iterations : 4,
+ fanout : 1,
+ parallelism : 5,
+ input_options : """
{
"num_records": 20000000,
"key_size": 10,
"value_size": 90,
"num_hot_keys": 10,
"hot_key_fraction": 1
}
- """.trim().replaceAll("\\s", ""),
- endpoint : 'localhost:8099',
- environment_type : 'DOCKER',
- environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ """.trim().replaceAll("\\s", ""),
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
]
],
].each { test -> test.pipelineOptions.putAll(additionalPipelineArgs) }
}
def loadTestJob = { scope, triggeringContext, mode ->
- def numberOfWorkers = 5
+ Map<Integer, List> testScenariosByParallelism = batchScenarios().groupBy { test ->
+ test.pipelineOptions.parallelism
+ }
+ Integer initialParallelism = testScenariosByParallelism.keySet().iterator().next()
+ List initialScenarios = testScenariosByParallelism.remove(initialParallelism)
- Flink flink = new Flink(scope, "beam_LoadTests_Go_GBK_Flink_${mode.capitalize()}")
+ def flink = new Flink(scope, "beam_LoadTests_Python_Combine_Flink_${mode.capitalize()}")
Review comment:
`beam_LoadTests_Python_Combine_Flink_` -> `beam_LoadTests_Go_GBK_Flink_`
----------------------------------------------------------------
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] kamilwu commented on pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13388:
URL: https://github.com/apache/beam/pull/13388#issuecomment-733719284
Tests passed: https://github.com/apache/beam/pull/13407, we can merge this now. Thank you @tszerszen!
----------------------------------------------------------------
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] tszerszen commented on a change in pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13388:
URL: https://github.com/apache/beam/pull/13388#discussion_r530273194
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,88 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source.
+
+package main
+
+import (
+ "context"
+ "flag"
+
+ "github.com/apache/beam/sdks/go/pkg/beam"
+ "github.com/apache/beam/sdks/go/pkg/beam/io/synthetic"
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/go/test/load"
+)
+
+var (
+ fanout = flag.Int(
+ "fanout",
+ 1,
+ "A number of GroupByKey operations to perform in parallel.")
+ iterations = flag.Int(
+ "iterations",
+ 1,
+ "A number of reiterations over per-key-grouped values to perform.")
+ syntheticConfig = flag.String(
+ "input_options",
+ "",
+ "A JSON object that describes the configuration for synthetic source.")
+)
+
+func parseSyntheticConfig() synthetic.SourceConfig {
+ if *syntheticConfig == "" {
+ panic("--input_options not provided")
+ } else {
+ encoded := []byte(*syntheticConfig)
+ return synthetic.DefaultSourceConfig().BuildFromJSON(encoded)
+ }
+}
+
+func main() {
+ flag.Parse()
+ beam.Init()
+
+ ctx := context.Background()
+
+ p, s := beam.NewPipelineWithRoot()
+ src := synthetic.SourceSingle(s, parseSyntheticConfig())
+ pcoll := beam.ParDo(s, &load.RuntimeMonitor{}, src)
+ for i := 0; i < *fanout; i++ {
+ pcoll = beam.GroupByKey(s, src)
+ beam.ParDo(s, func(key []byte, values func(*[]byte) bool) ([]byte, []byte) {
+ for i := 0; i < *iterations; i++ {
+ var value []byte
+ for values(&value) {
+ if i == *iterations-1 {
+ return key, value
+ }
+ }
+ }
+ return key, []byte{0}
+ }, pcoll)
Review comment:
Thank you I fixed it in the last commit.
----------------------------------------------------------------
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] kamilwu commented on a change in pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13388:
URL: https://github.com/apache/beam/pull/13388#discussion_r528834920
##########
File path: .test-infra/jenkins/job_LoadTests_GBK_Flink_Go.groovy
##########
@@ -0,0 +1,220 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as commonJobProperties
+import CommonTestProperties
+import LoadTestsBuilder as loadTestsBuilder
+import PhraseTriggeringPostCommitBuilder
+import Flink
+import InfluxDBCredentialsHelper
+
+import static LoadTestsBuilder.DOCKER_CONTAINER_REGISTRY
+
+String now = new Date().format('MMddHHmmss', TimeZone.getTimeZone('UTC'))
+
+def batchScenarios = {
+ [
+ [
+ title : 'Group By Key Go Load test: 2GB of 10B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
+ influx_measurement : 'go_batch_gbk_1',
+ input_options : '\'{' +
+ '"num_records": 200000000,' +
+ '"key_size": 1,' +
+ '"value_size": 9}\'',
+ iterations : 1,
+ fanout : 1,
+ parallelism : 5,
+ endpoint : 'localhost:8099',
+ environment_type : 'DOCKER',
+ environment_config : "${DOCKER_CONTAINER_REGISTRY}/beam_go_sdk:latest",
+ ]
+ ],
+ [
+ title : 'Group By Key Go Load test: 2GB of 100B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
+ pipelineOptions: [
+ job_name : "load-tests-go-flink-batch-gbk-1-${now}",
Review comment:
It seems you forgot about changing the "1" to "2" and so on. It doesn't really matter in case of `job_name`, but it does matter in case of `influx_measurement`: all results would be put into the first chart, leaving other charts empty.
----------------------------------------------------------------
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] kamilwu commented on a change in pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13388:
URL: https://github.com/apache/beam/pull/13388#discussion_r528832366
##########
File path: .test-infra/jenkins/job_LoadTests_GBK_Flink_Go.groovy
##########
@@ -0,0 +1,220 @@
+/*
+ * 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.
+ */
+
+import CommonJobProperties as commonJobProperties
+import CommonTestProperties
+import LoadTestsBuilder as loadTestsBuilder
+import PhraseTriggeringPostCommitBuilder
+import Flink
+import InfluxDBCredentialsHelper
+
+import static LoadTestsBuilder.DOCKER_CONTAINER_REGISTRY
+
+String now = new Date().format('MMddHHmmss', TimeZone.getTimeZone('UTC'))
+
+def batchScenarios = {
+ [
+ [
+ title : 'Group By Key Go Load test: 2GB of 10B records',
+ test : 'group_by_key',
+ runner : CommonTestProperties.Runner.FLINK,
+ influx_namespace: 'flink',
Review comment:
`influx_namespace` should be a part of `pipelineOptions`. Please put it under the `influx_measurement`.
----------------------------------------------------------------
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] tszerszen commented on pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13388:
URL: https://github.com/apache/beam/pull/13388#issuecomment-732131067
Run Seed Job
----------------------------------------------------------------
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] kamilwu commented on a change in pull request #13388: Go SDK Group by key load tests
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13388:
URL: https://github.com/apache/beam/pull/13388#discussion_r529708301
##########
File path: sdks/go/test/load/group_by_key/group_by_key.go
##########
@@ -0,0 +1,88 @@
+// 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.
+
+// This is GroupByKey load test with Synthetic Source.
+
+package main
+
+import (
+ "context"
+ "flag"
+
+ "github.com/apache/beam/sdks/go/pkg/beam"
+ "github.com/apache/beam/sdks/go/pkg/beam/io/synthetic"
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/go/pkg/beam/x/beamx"
+ "github.com/apache/beam/sdks/go/test/load"
+)
+
+var (
+ fanout = flag.Int(
+ "fanout",
+ 1,
+ "A number of GroupByKey operations to perform in parallel.")
+ iterations = flag.Int(
+ "iterations",
+ 1,
+ "A number of reiterations over per-key-grouped values to perform.")
+ syntheticConfig = flag.String(
+ "input_options",
+ "",
+ "A JSON object that describes the configuration for synthetic source.")
+)
+
+func parseSyntheticConfig() synthetic.SourceConfig {
+ if *syntheticConfig == "" {
+ panic("--input_options not provided")
+ } else {
+ encoded := []byte(*syntheticConfig)
+ return synthetic.DefaultSourceConfig().BuildFromJSON(encoded)
+ }
+}
+
+func main() {
+ flag.Parse()
+ beam.Init()
+
+ ctx := context.Background()
+
+ p, s := beam.NewPipelineWithRoot()
+ src := synthetic.SourceSingle(s, parseSyntheticConfig())
+ pcoll := beam.ParDo(s, &load.RuntimeMonitor{}, src)
+ for i := 0; i < *fanout; i++ {
+ pcoll = beam.GroupByKey(s, src)
+ beam.ParDo(s, func(key []byte, values func(*[]byte) bool) ([]byte, []byte) {
+ for i := 0; i < *iterations; i++ {
+ var value []byte
+ for values(&value) {
+ if i == *iterations-1 {
+ return key, value
+ }
+ }
+ }
+ return key, []byte{0}
+ }, pcoll)
Review comment:
You're missing calls to ParDo with RuntimeMonitor here, at the end of the for loop. Those are important, because we want to measure the runtime of a whole pipeline.
----------------------------------------------------------------
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