spark_dataproc: use n1-standard-4 for the cluster master

.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