You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2026/08/14 15:53:04 UTC
(beam) branch restore-load updated (5c9c5133584 -> 8612100c81f)
This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a change to branch restore-load
in repository https://gitbox.apache.org/repos/asf/beam.git
discard 5c9c5133584 Fix go GBK load test
new 8612100c81f Fix go GBK load test
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (5c9c5133584)
\
N -- N -- N refs/heads/restore-load (8612100c81f)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100b.txt | 2 +-
.../workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_10b.txt | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
(beam) 01/01: Fix go GBK load test
Posted by yh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch restore-load
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 8612100c81fc8c0fbf3972941c21dccbc8585668
Author: Yi Hu <ya...@google.com>
AuthorDate: Fri Aug 14 10:12:07 2026 -0400
Fix go GBK load test
* Restore go CoGBK load test parameter
* Set larger initialSplit size to avoid all elements processed in
single bundle
* Fix gbk load test to use emit, make sure gbk results are consumed
This is in alignment with CoGBK test and Java GBK test
---
.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml | 14 ++++++++++----
.../go_CoGBK_Flink_Batch_MultipleKey.txt | 4 ++--
.../go_CoGBK_Flink_Batch_Reiteration_10KB.txt | 4 ++--
.../go_CoGBK_Flink_Batch_Reiteration_2MB.txt | 4 ++--
.../go_GBK_Flink_Batch_100kb.txt | 2 +-
.../go_GBK_Flink_Batch_Fanout_4.txt | 2 +-
.../go_GBK_Flink_Batch_Fanout_8.txt | 2 +-
.../go_GBK_Flink_Batch_Reiteration_10KB.txt | 2 +-
.test-infra/dataproc/flink_cluster.sh | 6 ++++++
sdks/go/test/load/group_by_key/group_by_key.go | 14 +++++++++-----
10 files changed, 35 insertions(+), 19 deletions(-)
diff --git a/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml b/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml
index 731c5b21274..d29b4fd60a1 100644
--- a/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml
+++ b/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml
@@ -113,6 +113,7 @@ jobs:
-Prunner=FlinkRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_1 }}' \
- name: run GBK Flink Batch Go Load Test 2 (100 b records)
+ if: always()
timeout-minutes: 120
uses: ./.github/actions/gradle-command-self-hosted-action
with:
@@ -122,6 +123,7 @@ jobs:
-Prunner=FlinkRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_2 }}' \
- name: run GBK Flink Batch Go Load Test 3 (100 kb records)
+ if: always()
timeout-minutes: 120
uses: ./.github/actions/gradle-command-self-hosted-action
with:
@@ -131,6 +133,7 @@ jobs:
-Prunner=FlinkRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_3 }}' \
- name: run GBK Flink Batch Go Load Test 6 (reiterate 4 times 10 kb)
+ if: always()
timeout-minutes: 120
uses: ./.github/actions/gradle-command-self-hosted-action
with:
@@ -140,11 +143,13 @@ jobs:
-Prunner=FlinkRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_6 }}' \
- name: Resize Flink to 10 workers
+ if: always()
env:
FLINK_NUM_WORKERS: 10
run: |
cd ${{ github.workspace }}/.test-infra/dataproc; ./flink_cluster.sh resize
- name: run GBK Flink Batch Go Load Test 4 (fanout 4)
+ if: always()
timeout-minutes: 120
uses: ./.github/actions/gradle-command-self-hosted-action
with:
@@ -154,6 +159,7 @@ jobs:
-Prunner=FlinkRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_4 }}' \
- name: run GBK Flink Batch Go Load Test 5 (fanout 8)
+ if: always()
timeout-minutes: 120
uses: ./.github/actions/gradle-command-self-hosted-action
with:
@@ -162,7 +168,7 @@ jobs:
-PloadTest.mainClass=group_by_key \
-Prunner=FlinkRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_5 }}' \
- - name: Teardown Flink
- if: always()
- run: |
- ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete
+ # - name: Teardown Flink
+ # if: always()
+ # run: |
+ # ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete
diff --git a/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_MultipleKey.txt b/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_MultipleKey.txt
index 504b464eb1f..ec2c5c3e040 100644
--- a/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_MultipleKey.txt
+++ b/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_MultipleKey.txt
@@ -16,8 +16,8 @@
--influx_measurement=go_batch_cogbk_1
--influx_namespace=flink
---input_options=''{\"num_records\":10000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":5,\"hot_key_fraction\":1}''
---co_input_options=''{\"num_records\":1000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":1000,\"hot_key_fraction\":1}''
+--input_options=''{\"num_records\":1000000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":5,\"hot_key_fraction\":1}''
+--co_input_options=''{\"num_records\":100000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":1000,\"hot_key_fraction\":1}''
--iterations=1
--parallelism=5
--endpoint=localhost:8099
diff --git a/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_10KB.txt b/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_10KB.txt
index 7eb9a3c8053..7698c7aa7c7 100644
--- a/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_10KB.txt
+++ b/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_10KB.txt
@@ -16,8 +16,8 @@
--influx_measurement=go_batch_cogbk_2
--influx_namespace=flink
---input_options=''{\"num_records\":10000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":1000,\"hot_key_fraction\":1}''
---co_input_options=''{\"num_records\":1000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":10,\"hot_key_fraction\":1}''
+--input_options=''{\"num_records\":1000000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":100000,\"hot_key_fraction\":1}''
+--co_input_options=''{\"num_records\":100000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":1000,\"hot_key_fraction\":1}''
--iterations=4
--parallelism=5
--endpoint=localhost:8099
diff --git a/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_2MB.txt b/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_2MB.txt
index ccf5ae7cbf2..c6b1f5fcc33 100644
--- a/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_2MB.txt
+++ b/.github/workflows/load-tests-pipeline-options/go_CoGBK_Flink_Batch_Reiteration_2MB.txt
@@ -16,8 +16,8 @@
--influx_measurement=go_batch_cogbk_3
--influx_namespace=flink
---input_options=''{\"num_records\":100,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":100,\"hot_key_fraction\":1}''
---co_input_options=''{\"num_records\":10,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":10,\"hot_key_fraction\":1}''
+--input_options=''{\"num_records\":1000000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":1000,\"hot_key_fraction\":1}''
+--co_input_options=''{\"num_records\":100000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":1000,\"hot_key_fraction\":1}''
--iterations=4
--parallelism=5
--endpoint=localhost:8099
diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt
index f02e6984c81..a188f8c0978 100644
--- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt
+++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_100kb.txt
@@ -19,7 +19,7 @@
--iterations=1
--fanout=1
--parallelism=5
---input_options=''{\"num_records\":50,\"key_size\":10000,\"value_size\":90000}''
+--input_options=''{\"num_records\":1000,\"key_size\":10000,\"value_size\":90000}''
--endpoint=localhost:8099
--environment_type=DOCKER
--environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt
index 0042a9b80f3..4378d56a8f8 100644
--- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt
+++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_4.txt
@@ -19,7 +19,7 @@
--iterations=1
--fanout=4
--parallelism=16
---input_options=''{\"num_records\":10000,\"key_size\":10,\"value_size\":90}''
+--input_options=''{\"num_records\":100000,\"key_size\":10,\"value_size\":90}''
--endpoint=localhost:8099
--environment_type=DOCKER
--environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt
index fb14c2da58d..43292d57717 100644
--- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt
+++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Fanout_8.txt
@@ -19,7 +19,7 @@
--iterations=1
--fanout=8
--parallelism=16
---input_options=''{\"num_records\":10000,\"key_size\":10,\"value_size\":90}''
+--input_options=''{\"num_records\":100000,\"key_size\":10,\"value_size\":90}''
--endpoint=localhost:8099
--environment_type=DOCKER
--environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
diff --git a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt
index ee220853c60..c4d33c21482 100644
--- a/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt
+++ b/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt
@@ -19,7 +19,7 @@
--iterations=4
--fanout=1
--parallelism=5
---input_options=''{\"num_records\":10000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":100,\"hot_key_fraction\":1}''
+--input_options=''{\"num_records\":1000000,\"key_size\":10,\"value_size\":90,\"num_hot_keys\":200,\"hot_key_fraction\":1}''
--endpoint=localhost:8099
--environment_type=DOCKER
--environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
diff --git a/.test-infra/dataproc/flink_cluster.sh b/.test-infra/dataproc/flink_cluster.sh
index e146aa4b6ce..17180ced89d 100755
--- a/.test-infra/dataproc/flink_cluster.sh
+++ b/.test-infra/dataproc/flink_cluster.sh
@@ -183,6 +183,12 @@ function create() {
start_tunnel
}
+# Connects to an existing Flink cluster and opens an SSH tunnel.
+function tunnel() {
+ get_leader
+ start_tunnel
+}
+
# Resizes an active Flink cluster.
function resize() {
if [[ -z "$FLINK_NUM_WORKERS" ]]; then
diff --git a/sdks/go/test/load/group_by_key/group_by_key.go b/sdks/go/test/load/group_by_key/group_by_key.go
index 645afabedef..4b2bc738796 100644
--- a/sdks/go/test/load/group_by_key/group_by_key.go
+++ b/sdks/go/test/load/group_by_key/group_by_key.go
@@ -49,12 +49,17 @@ func parseSyntheticConfig() synthetic.SourceConfig {
panic("--input_options not provided")
} else {
encoded := []byte(*syntheticConfig)
- return synthetic.DefaultSourceConfig().BuildFromJSON(encoded)
+ cfg := synthetic.DefaultSourceConfig().BuildFromJSON(encoded)
+ if cfg.NumElements >= 100 && cfg.InitialSplits <= 1 {
+ cfg.InitialSplits = 100
+ }
+ return cfg
}
}
func init() {
- register.DoFn2x2[[]byte, func(*[]byte) bool, []byte, []byte]((*ungroupAndReiterateFn)(nil))
+ register.DoFn3x0[[]byte, func(*[]byte) bool, func([]byte, []byte)]((*ungroupAndReiterateFn)(nil))
+ register.Emitter2[[]byte, []byte]()
register.Iter1[[]byte]()
}
@@ -65,16 +70,15 @@ type ungroupAndReiterateFn struct {
// TODO use re-iterators once supported.
-func (fn *ungroupAndReiterateFn) ProcessElement(key []byte, values func(*[]byte) bool) ([]byte, []byte) {
+func (fn *ungroupAndReiterateFn) ProcessElement(key []byte, values func(*[]byte) bool, emit func([]byte, []byte)) {
var value []byte
for i := 0; i < fn.Iterations; i++ {
for values(&value) {
if i == fn.Iterations-1 {
- return key, value
+ emit(key, value)
}
}
}
- return key, []byte{0}
}
func main() {