spark_dataproc: use n1-standard-4 for the cluster master (#347)

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.

Signed-off-by: tnazarew <tomasz.nazarewicz@getindata.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: tnazarew

.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 }}

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