You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by vt...@apache.org on 2025/09/04 12:58:01 UTC
(beam) 01/01: Fix getting log level, fix runner for the workflow
This is an automated email from the ASF dual-hosted git repository.
vterentev pushed a commit to branch fix-prism-runner-log-level
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 79ef08cdae0436c8bbc7b79710b8e28a1b2b260a
Author: Vitaly Terentyev <vi...@akvelon.com>
AuthorDate: Mon Sep 1 16:08:42 2025 +0400
Fix getting log level, fix runner for the workflow
---
.../workflows/beam_LoadTests_Python_CoGBK_Flink_Batch.yml | 10 +++++-----
...ython_CoGBK_Dataflow_Flink_Batch_100b_Multiple_Keys.txt | 6 ++++--
.../python_CoGBK_Dataflow_Flink_Batch_100b_Single_Key.txt | 6 ++++--
.../python_CoGBK_Dataflow_Flink_Batch_10kB.txt | 6 ++++--
.../python/apache_beam/runners/portability/prism_runner.py | 14 +++++++++++++-
5 files changed, 30 insertions(+), 12 deletions(-)
diff --git a/.github/workflows/beam_LoadTests_Python_CoGBK_Flink_Batch.yml b/.github/workflows/beam_LoadTests_Python_CoGBK_Flink_Batch.yml
index 2c0c61007cd..fcbfbd0dce2 100644
--- a/.github/workflows/beam_LoadTests_Python_CoGBK_Flink_Batch.yml
+++ b/.github/workflows/beam_LoadTests_Python_CoGBK_Flink_Batch.yml
@@ -54,7 +54,7 @@ env:
HADOOP_DOWNLOAD_URL: https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
FLINK_TASKMANAGER_SLOTS: 1
DETACHED_MODE: true
- HARNESS_IMAGES_TO_PULL: gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
+ HARNESS_IMAGES_TO_PULL: gcr.io/apache-beam-testing/beam-sdk/beam_python3.9_sdk:latest
JOB_SERVER_IMAGE: gcr.io/apache-beam-testing/beam_portability/beam_flink1.17_job_server:latest
ARTIFACTS_DIR: gs://beam-flink-cluster/beam-loadtests-python-cogbk-flink-batch-${{ github.run_id }}
@@ -109,7 +109,7 @@ jobs:
arguments: |
--info \
-PloadTest.mainClass=apache_beam.testing.load_tests.co_group_by_key_test \
- -Prunner=FlinkRunner \
+ -Prunner=PortableRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Python_CoGBK_Flink_Batch_test_arguments_1 }} --job_name=load-tests-python-flink-batch-cogbk-1-${{ steps.datetime.outputs.datetime }}' \
- name: run CoGBK 2GB of 100B records with multiple keys
uses: ./.github/actions/gradle-command-self-hosted-action
@@ -118,7 +118,7 @@ jobs:
arguments: |
--info \
-PloadTest.mainClass=apache_beam.testing.load_tests.co_group_by_key_test \
- -Prunner=FlinkRunner \
+ -Prunner=PortableRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Python_CoGBK_Flink_Batch_test_arguments_2 }} --job_name=load-tests-python-flink-batch-cogbk-2-${{ steps.datetime.outputs.datetime }}' \
- name: run CoGBK reiterate 4 times 10kB values
uses: ./.github/actions/gradle-command-self-hosted-action
@@ -127,9 +127,9 @@ jobs:
arguments: |
--info \
-PloadTest.mainClass=apache_beam.testing.load_tests.co_group_by_key_test \
- -Prunner=FlinkRunner \
+ -Prunner=PortableRunner \
'-PloadTest.args=${{ env.beam_LoadTests_Python_CoGBK_Flink_Batch_test_arguments_2 }} --job_name=load-tests-python-flink-batch-cogbk-3-${{ steps.datetime.outputs.datetime }}' \
- name: Teardown Flink
if: always()
run: |
- ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete
\ No newline at end of file
+ ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete
diff --git a/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Multiple_Keys.txt b/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Multiple_Keys.txt
index 4b8a2f72010..163842d7e2b 100644
--- a/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Multiple_Keys.txt
+++ b/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Multiple_Keys.txt
@@ -23,6 +23,8 @@
--co_input_options=''{\\"num_records\\":2000000,\\"key_size\\":10,\\"value_size\\":90,\\"num_hot_keys\\":5,\\"hot_key_fraction\\":1}''
--iterations=1
--parallelism=5
---endpoint=localhost:8099
+--runner=PortableRunner
+--job_endpoint=localhost:8099
--environment_type=DOCKER
---environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
\ No newline at end of file
+--environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_python3.9_sdk:latest
+--timeout_ms=0
diff --git a/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Single_Key.txt b/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Single_Key.txt
index 3aeb927f04e..b704fa057be 100644
--- a/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Single_Key.txt
+++ b/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_100b_Single_Key.txt
@@ -23,6 +23,8 @@
--co_input_options=''{\\"num_records\\":2000000,\\"key_size\\":10,\\"value_size\\":90,\\"num_hot_keys\\":1000,\\"hot_key_fraction\\":1}''
--iterations=1
--parallelism=5
---endpoint=localhost:8099
+--runner=PortableRunner
+--job_endpoint=localhost:8099
--environment_type=DOCKER
---environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
\ No newline at end of file
+--environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_python3.9_sdk:latest
+--timeout_ms=0
diff --git a/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_10kB.txt b/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_10kB.txt
index e350e2d2994..d0a4c73b728 100644
--- a/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_10kB.txt
+++ b/.github/workflows/load-tests-pipeline-options/python_CoGBK_Dataflow_Flink_Batch_10kB.txt
@@ -23,6 +23,8 @@
--co_input_options=''{\\"num_records\\":2000000,\\"key_size\\":10,\\"value_size\\":90,\\"num_hot_keys\\":1000,\\"hot_key_fraction\\":1}''
--iterations=4
--parallelism=5
---endpoint=localhost:8099
+--runner=PortableRunner
+--job_endpoint=localhost:8099
--environment_type=DOCKER
---environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest
\ No newline at end of file
+--environment_config=gcr.io/apache-beam-testing/beam-sdk/beam_python3.9_sdk:latest
+--timeout_ms=0
diff --git a/sdks/python/apache_beam/runners/portability/prism_runner.py b/sdks/python/apache_beam/runners/portability/prism_runner.py
index 50ecdb1252e..cb17688c65e 100644
--- a/sdks/python/apache_beam/runners/portability/prism_runner.py
+++ b/sdks/python/apache_beam/runners/portability/prism_runner.py
@@ -28,6 +28,7 @@ import json
import logging
import os
import platform
+import re
import shutil
import stat
import subprocess
@@ -124,7 +125,18 @@ class PrismRunnerLogFilter(logging.Filter):
message = record.getMessage().replace(
'"!ERROR:time.Time year outside of range [0,9999]"', '')
json_record = json.loads(message)
- record.levelno = getattr(logging, json_record["level"])
+ level_str = json_record["level"]
+ # Example level with offset: 'ERROR+2'
+ if "+" in level_str or "-" in level_str:
+ match = re.match(r"([A-Z]+)([+-]\d+)", level_str)
+ if match:
+ base, offset = match.groups()
+ base_level = getattr(logging, base, logging.INFO)
+ record.levelno = base_level + int(offset)
+ else:
+ record.levelno = getattr(logging, level_str, logging.INFO)
+ else:
+ record.levelno = getattr(logging, level_str, logging.INFO)
record.levelname = logging.getLevelName(record.levelno)
if "source" in json_record:
record.funcName = json_record["source"]["function"]