Params for Flink parallelism

Author: jakubadamek

Diff for: performance-tests/scaling/config.sh.tmpl

@@ -7,6 +7,7 @@ export FHIR_UPLOADER_CORES=8
 export ENABLE_UPLOAD=false
 export ENABLE_DOWNLOAD=true
 export JDBC_MODE=false
+# DirectRunner, FlinkRunner or DataflowRunner
 export FHIR_ETL_RUNNER=DataflowRunner
 
 # pipeline-scaling-1, pipeline-scaling-2, pipeline-scaling-belgium

Diff for: performance-tests/scaling/setup_google3.sh

@@ -10,13 +10,15 @@ set -o nounset
 case "$DB_TYPE" in
   "alloy")
     ALLOY_INSTANCE="projects/fhir-analytics-test/locations/us-central1/clusters/pipeline-scaling-alloydb-1/instances/pipeline-scaling-alloydb-largest"
+    sudo killall alloydb-auth-proxy || true
     nohup ~/Downloads/alloydb-auth-proxy $ALLOY_INSTANCE &
     if [[ "$ENABLE_UPLOAD" = true ]]; then
       for cmd in "DROP DATABASE IF EXISTS" "CREATE DATABASE"; do
         PGPASSWORD="$DB_PASSWORD" psql -h 127.0.0.1 -p 5432 -U "$DB_USERNAME" -c "$cmd $DB_PATIENTS"
       done
     else
       # Check DB connection.
+      sleep 1
       PGPASSWORD="$DB_PASSWORD" psql -h 127.0.0.1 -p 5432 -U "$DB_USERNAME" -c "SELECT 1"
     fi
     DB_CONNECTION="jdbc:postgresql:///${DB_PATIENTS}?127.0.0.1:5432"

Diff for: performance-tests/scaling/upload_download.py

@@ -45,15 +45,22 @@ def download():
 
 def run_fhir_etl(parquet_dir):
     common_etl_args = [
-        "--fasterCopy=true",
         f"--runner={FHIR_ETL_RUNNER}",
         f"--resourceList=Patient,Encounter,Observation",
-        f"--outputParquetPath={parquet_dir}",
-        # Dataflow runner:
-        f"--region={SQL_ZONE}",
-        f"--numWorkers={NUM_WORKERS}",
-        "--gcpTempLocation=gs://fhir-analytics-test/dataflow_temp"
+        f"--outputParquetPath={parquet_dir}"
     ]
+    if FHIR_ETL_RUNNER == "DataflowRunner":
+        common_etl_args.extend([
+            f"--region={SQL_ZONE}",
+            f"--numWorkers={NUM_WORKERS}",
+            "--gcpTempLocation=gs://fhir-analytics-test/dataflow_temp",
+        ])
+    if FHIR_ETL_RUNNER == "FlinkRunner":
+        common_etl_args.extend([
+            "--parallelism=100",
+            "--maxParallelism=100",
+            "--fasterCopy=true",
+        ])
 
     if not JDBC_MODE:
         # Test HAPI server readiness.

Diff for: performance-tests/scaling/variables.sh

@@ -7,25 +7,34 @@ export DB_PATIENTS="patients_$PATIENTS"
 export DIR_WITH_THIS_SCRIPT
 DIR_WITH_THIS_SCRIPT="$( cd -- "$(dirname "$0")" >/dev/null 2>&1 && pwd -P )"
 
-case "$POSTGRES_DB_INSTANCE" in
-  "pipeline-scaling-belgium")
-    export DB_USERNAME="postgres"
-    export DB_PASSWORD="C%_/\Rn-=fI5f$}7"
-    export ZONE=belgium
-    ;;
-  "pipeline-scaling-1")
-    export DB_USERNAME="pipeline-scaling-user"
-    export DB_PASSWORD="C%_/\Rn-=fI5f$}7"
+export DB_USERNAME="postgres"
+export DB_PASSWORD="C%_/\Rn-=fI5f$}7"
+
+case "$DB_TYPE" in
+  "alloy")
     export ZONE=us-central
     ;;
-  "pipeline-scaling-2")
-    export DB_USERNAME="postgres"
-    export DB_PASSWORD="C%_/\Rn-=fI5f$}7"
-    export ZONE=us-central
+  "postgres")
+    case "$POSTGRES_DB_INSTANCE" in
+      "pipeline-scaling-belgium")
+        export ZONE=belgium
+        ;;
+      "pipeline-scaling-1")
+        export DB_USERNAME="pipeline-scaling-user"
+        export ZONE=us-central
+        ;;
+      "pipeline-scaling-2")
+        export ZONE=us-central
+        ;;
+      *)
+        echo "Invalid POSTGRES_DB_INSTANCE $POSTGRES_DB_INSTANCE"
+        exit 2
+        ;;
+    esac
     ;;
   *)
-    echo "Invalid POSTGRES_DB_INSTANCE $POSTGRES_DB_INSTANCE"
-    exit 2
+    echo "Invalid DB_TYPE DB_TYPE"
+    exit 3
     ;;
 esac