From eb2ba9df12517d82f8c7a147efaaf17e2e0ab2c9 Mon Sep 17 00:00:00 2001 From: tnazarew Date: Wed, 21 Oct 2026 23:28:37 +0200 Subject: [PATCH] spark_dataproc: use n1-standard-4 for the cluster master Since one cluster runs all OpenLineage versions, up to 3 scenarios run at the same time and all their drivers run on the master. On n1-standard-2 (7.5 GB) the master ran out of memory now and then: in run 39672836849 the bigquery and bigquery_to_delta drivers were killed with "Driver received SIGTERM/SIGKILL signal and exited with 143 code, which potentially signifies a memory pressure". create-cluster gets a --master-machine-type option (default stays n1-standard-2) and the workflow uses n1-standard-4 (15 GB). Workers are unchanged: 8 instead of 6 vCPU per cluster. Co-Authored-By: Claude Opus 5.5 Signed-off-by: tnazarew --- .github/workflows/producer_spark_dataproc.yml | 4 +++- producer/spark_dataproc/runner/dataproc_workflow.py | 10 +++++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/.github/workflows/producer_spark_dataproc.yml b/.github/workflows/producer_spark_dataproc.yml index b4d1c4d6..1f466acf 100644 --- a/.github/workflows/producer_spark_dataproc.yml +++ b/.github/workflows/producer_spark_dataproc.yml @@ -165,13 +165,15 @@ jobs: --cluster-name "${{ steps.init.outputs.cluster_name }}" \ --credentials-file ${{ steps.gcp-auth.outputs.credentials_file_path }} \ --metadata "$metadata" \ + --master-machine-type n1-standard-4 \ --initialization-actions="${{ steps.upload-initialization-actions.outputs.init_actions }}" - name: Remove initialization actions from GCS if: ${{ always() && steps.upload-initialization-actions.outputs.scripts_dir }} run: gcloud storage rm -r "${{ steps.upload-initialization-actions.outputs.scripts_dir }}" || true - # openlineage versions run one after another, scenarios of a single version run in parallel; + # openlineage versions run one after another, scenarios of a single version run in parallel + # (their drivers all run on the master, which is why it's bigger than the workers); # failed scenarios don't fail the job, they show up in validation together with their driver logs - name: Run producer jobs and create OL events if: ${{ steps.init.outputs.any_scenarios }} diff --git a/producer/spark_dataproc/runner/dataproc_workflow.py b/producer/spark_dataproc/runner/dataproc_workflow.py index 0a1acb35..a8712ab6 100644 --- a/producer/spark_dataproc/runner/dataproc_workflow.py +++ b/producer/spark_dataproc/runner/dataproc_workflow.py @@ -211,6 +211,7 @@ async def create_cluster( metadata: typing.Dict[str, str], initialization_actions: typing.List[str], credentials: Credentials, + master_machine_type: str = "n1-standard-2", ) -> None: """ Asynchronously creates a Dataproc cluster. @@ -232,7 +233,7 @@ async def create_cluster( }, "master_config": { "num_instances": 1, - "machine_type_uri": "n1-standard-2", + "machine_type_uri": master_machine_type, "disk_config": { "boot_disk_type": "pd-standard", "boot_disk_size_gb": 300, @@ -607,6 +608,7 @@ async def create_cluster_command(args): metadata=spark_metadata, initialization_actions=initialization_actions, credentials=credentials, + master_machine_type=args.master_machine_type, ) @@ -766,6 +768,12 @@ def main(): default="", help="Cluster initialization action scripts gcs locatizations separated by commas", ) + parser_create.add_argument( + "--master-machine-type", + required=False, + default="n1-standard-2", + help="Machine type of the master node, which runs the drivers of all jobs (default: n1-standard-2)", + ) parser_create.set_defaults(func=create_cluster_command) # Run job command