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 <[email protected]> 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() {
