42 support spark resource configuration in notebook (#318)

xuwenyihust · web-flow · commit 8161b4bfc7bc · 2024-08-13T16:32:49.000+08:00
* Update SparkSession configuration and add API request in startup.py

* Add logging to create_spark_dev function in startup.py

* Refactor create_spark_dev function in startup.py

* Refactor create_spark_dev function and add logging in startup.py

* Refactor create_spark_dev function and add logging in startup.py

* Refactor create_spark_dev function and update API request in startup.py

* Refactor create_spark_dev function and handle exception in startup.py

* Refactor create_spark_dev function and update API request in startup.py

* Refactor create_spark_dev function and update API request in startup.py

* Refactor create_spark_dev function and update API request in startup.py

* Refactor create_spark_dev function and update API request in startup.py

* Refactor create_spark_dev function and update API request in startup.py

* Refactor create_spark_dev function and handle exception in startup.py

* Refactor create_spark_dev function and update API request in startup.py

* Update config variable name in startup.py
diff --git a/docker/notebook/startup.py b/docker/notebook/startup.py
@@ -5,9 +5,14 @@
 from IPython import get_ipython
 from IPython.display import *
 from kubernetes import client, config
+import requests
+import logging
 
 environment = os.getenv('ENVIRONMENT', 'development')  # Default to 'development' if not set
 
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
 # Set the environment variables
 def set_env():
     # kubernetes_host = os.environ.get('KUBERNETES_SERVICE_HOST')
@@ -27,56 +32,57 @@ def set_env():
 
 # Create a Spark session
 # def create_spark(app_name, master_url):
-    spark = SparkSession.builder \
-        .appName(app_name) \
-        .master(kubernetes_url) \
-        .config("spark.submit.deployMode", "client") \
-        .config("spark.driver.host", driver_host) \
-        .config("spark.driver.cores", "1") \
-        .config("spark.driver.memory", "1g") \
-        .config("spark.executor.instances", "1") \
-        .config("spark.executor.cores", "1") \
-        .config("spark.executor.memory", "1g") \
-        .config("spark.kubernetes.namespace", namespace) \
-        .config("spark.kubernetes.container.image", executor_image) \
-        .config("spark.kubernetes.authenticate.driver.serviceAccountName", service_account) \
-        .config("spark.kubernetes.authenticate.executor.serviceAccountName", service_account) \
-        .config("spark.eventLog.enabled", "true") \
-        .config("spark.eventLog.dir", f"gs://{bucket_name}/event-logs/") \
-        .config("spark.history.fs.logDirectory", f"gs://{bucket_name}/event-logs/") \
-        .config("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem") \
-        .config("spark.hadoop.fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS") \
-        .config("spark.hadoop.fs.gs.auth.service.account.enable", "true") \
-        .getOrCreate()
+    # spark = SparkSession.builder \
+    #     .appName(app_name) \
+    #     .master(kubernetes_url) \
+    #     .config("spark.submit.deployMode", "client") \
+    #     .config("spark.driver.host", driver_host) \
+    #     .config("spark.driver.cores", "1") \
+    #     .config("spark.driver.memory", "1g") \
+    #     .config("spark.executor.instances", "1") \
+    #     .config("spark.executor.cores", "1") \
+    #     .config("spark.executor.memory", "1g") \
+    #     .config("spark.kubernetes.namespace", namespace) \
+    #     .config("spark.kubernetes.container.image", executor_image) \
+    #     .config("spark.kubernetes.authenticate.driver.serviceAccountName", service_account) \
+    #     .config("spark.kubernetes.authenticate.executor.serviceAccountName", service_account) \
+    #     .config("spark.eventLog.enabled", "true") \
+    #     .config("spark.eventLog.dir", f"gs://{bucket_name}/event-logs/") \
+    #     .config("spark.history.fs.logDirectory", f"gs://{bucket_name}/event-logs/") \
+    #     .config("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem") \
+    #     .config("spark.hadoop.fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS") \
+    #     .config("spark.hadoop.fs.gs.auth.service.account.enable", "true") \
+    #     .getOrCreate()
     
-    return spark
+    # return spark
 
 # def start():
-    # Configuring the API client
-    config.load_incluster_config()
+#     # Configuring the API client
+#     config.load_incluster_config()
 
-    # Creating an API instance to interact with the K8s service
-    v1 = client.CoreV1Api()
+#     # Creating an API instance to interact with the K8s service
+#     v1 = client.CoreV1Api()
 
-    # Fetching the service details
-    service_name = os.environ.get("WEBUI_SERVICE_NAME", "notebook-spark-ui")
-    service = v1.read_namespaced_service(service_name, namespace)
+#     # Fetching the service details
+#     service_name = os.environ.get("WEBUI_SERVICE_NAME", "notebook-spark-ui")
+#     service = v1.read_namespaced_service(service_name, namespace)
 
-    webui_host = service.status.load_balancer.ingress[0].ip
-    webui_port = spark.sparkContext.uiWebUrl.split(":")[-1]
-    webui_url = f"http://{webui_host}:{webui_port}"
+#     webui_host = service.status.load_balancer.ingress[0].ip
+#     webui_port = spark.sparkContext.uiWebUrl.split(":")[-1]
+#     webui_url = f"http://{webui_host}:{webui_port}"
 
-    msg = f"**App name**: {app_name}\n\n" + \
-        f"**Master**: {kubernetes_url}\n\n" + \
-        f"**Driver host**: {driver_host}\n\n" + \
-        f"**Spark UI**: {webui_url}"
+#     msg = f"**App name**: {app_name}\n\n" + \
+#         f"**Master**: {kubernetes_url}\n\n" + \
+#         f"**Driver host**: {driver_host}\n\n" + \
+#         f"**Spark UI**: {webui_url}"
 
-    display(Markdown(msg))
+#     display(Markdown(msg))
 
 class PawMarkSparkSession:
 
-    def __init__(self, spark_session):
+    def __init__(self, config_json, spark_session):
         self._spark_session = spark_session
+        self._config_json = config_json
         self.history_server_base_url = "http://localhost:18080"
     
     def __getattr__(self, name):
@@ -94,26 +100,36 @@ def _repr_html_(self):
         return f"""
         <div style="border: 1px solid #e8e8e8; padding: 10px;">
             <h3>Spark Session Information</h3>
+            <p><strong>Config:</strong> {self._config_json}</p>
             <p><strong>Application ID:</strong> {application_id}</p>
             <p><strong>Spark UI:</strong> <a href="{spark_ui_link}">{spark_ui_link}</a></p>
         </div>
         """
 
 def create_spark_dev():
-    spark = PawMarkSparkSession(SparkSession.builder \
-        .appName("PySpark Example") \
-        .master("spark://spark-master:7077") \
-        .config("spark.jars.packages", "io.delta:delta-spark_2.12:3.0.0") \
-        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
-        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
-        .config("spark.eventLog.enabled", "true") \
-        .config("spark.eventLog.dir", "/opt/data/spark-events") \
-        .config("spark.history.fs.logDirectory", "/opt/data/spark-events") \
-        .config("spark.sql.warehouse.dir", "/opt/data/spark-warehouse") \
-        .config("executor.memory", "1g") \
-        .config("executor.cores", "1") \
-        .config("spark.executor.instances", "1") \
-        .getOrCreate())
+    logger.info("Creating Spark session")
+    try:
+        config_json = requests.get("http://server:5002/spark_app/config").json()
+    except Exception as e:
+        config_json = 'Error loading config: ' + str(e)
+
+    spark = PawMarkSparkSession(
+        config_json,
+        SparkSession.builder \
+            .appName("PySpark Example") \
+            .master("spark://spark-master:7077") \
+            .config("spark.jars.packages", "io.delta:delta-spark_2.12:3.0.0") \
+            .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
+            .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
+            .config("spark.eventLog.enabled", "true") \
+            .config("spark.eventLog.dir", "/opt/data/spark-events") \
+            .config("spark.history.fs.logDirectory", "/opt/data/spark-events") \
+            .config("spark.sql.warehouse.dir", "/opt/data/spark-warehouse") \
+            .config("executor.memory", config_json['executor.memory']) \
+            .config("executor.cores", config_json['executor.cores']) \
+            .config("spark.executor.instances", config_json['spark.executor.instances']) \
+            .getOrCreate()
+        )
     
     return spark
     
diff --git a/examples/user_0@gmail.com/quickstart.ipynb b/examples/user_0@gmail.com/quickstart.ipynb
@@ -2,6 +2,7 @@
  "cells": [
   {
    "cell_type": "markdown",
+   "isExecuted": true,
    "lastExecutionResult": null,
    "lastExecutionTime": null,
    "metadata": {},
@@ -21,6 +22,7 @@
   },
   {
    "cell_type": "markdown",
+   "isExecuted": true,
    "lastExecutionResult": null,
    "lastExecutionTime": null,
    "metadata": {},
@@ -31,6 +33,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
+   "isExecuted": false,
    "lastExecutionResult": null,
    "lastExecutionTime": null,
    "metadata": {},
@@ -40,28 +43,10 @@
     "import pandas as pd"
    ]
   },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "lastExecutionResult": null,
-   "lastExecutionTime": null,
-   "metadata": {},
-   "outputs": [
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "666\n"
-     ]
-    }
-   ],
-   "source": [
-    "print(666)"
-   ]
-  },
   {
    "cell_type": "code",
    "execution_count": 4,
+   "isExecuted": false,
    "lastExecutionResult": null,
    "lastExecutionTime": null,
    "metadata": {},
@@ -93,6 +78,7 @@
   {
    "cell_type": "code",
    "execution_count": 5,
+   "isExecuted": false,
    "lastExecutionResult": null,
    "lastExecutionTime": null,
    "metadata": {},
@@ -105,6 +91,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
+   "isExecuted": false,
    "lastExecutionResult": null,
    "lastExecutionTime": null,
    "metadata": {},
@@ -129,6 +116,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
+   "isExecuted": false,
    "lastExecutionResult": null,
    "lastExecutionTime": null,
    "metadata": {},
@@ -153,6 +141,15 @@
    "source": [
     "spark.sql(\"select * from test\").show()"
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "spark.stop()"
+   ]
   }
  ],
  "metadata": {
@@ -172,7 +169,8 @@
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
    "version": "3.11.6"
-  }
+  },
+  "uuid": "f193bc0d-d108-4ceb-babc-9fc341dfa83d"
  },
  "nbformat": 4,
  "nbformat_minor": 4
diff --git a/server/app/routes/spark_app.py b/server/app/routes/spark_app.py
@@ -10,4 +10,8 @@
 def create_spark_app(spark_app_id):
     data = request.get_json()
     notebook_path = data.get('notebookPath', None)
-    return SparkApp.create_spark_app(spark_app_id=spark_app_id, notebook_path=notebook_path)
+    return SparkApp.create_spark_app(spark_app_id=spark_app_id, notebook_path=notebook_path)
+
+@spark_app_blueprint.route('/spark_app/config', methods=['GET'])
+def get_spark_app_config():
+    return SparkApp.get_spark_app_config()
diff --git a/server/app/services/spark_app.py b/server/app/services/spark_app.py
@@ -40,6 +40,19 @@ def get_spark_app_by_id(spark_app_id: str = None):
       status=200
     )
   
+  @staticmethod
+  def get_spark_app_config():
+    spark_app_config = {
+      'executor.memory': '512m',
+      'executor.cores': '1',
+      'spark.executor.instances': '1',
+    }
+
+    return Response(
+      response=json.dumps(spark_app_config),
+      status=200
+    )
+  
   @staticmethod
   def create_spark_app(spark_app_id: str = None, notebook_path: str = None):
     logger.info(f"Creating spark app with id: {spark_app_id} for notebook path: {notebook_path}")
diff --git a/webapp/src/components/auth/LoginForm.js b/webapp/src/components/auth/LoginForm.js
@@ -33,7 +33,7 @@ function LoginForm({ onLogin }) {
         } else {
             // The login failed
             // You might want to show an error message here
-            console.error('Failed to log in');
+            console.error('Failed to log in: ', response.text, response.status);
             setError(true);
         }
     };
diff --git a/webapp/src/models/SparkModel.js b/webapp/src/models/SparkModel.js
@@ -8,9 +8,6 @@ class SparkModel {
   static isSparkInfo(html) {
     const parser = new DOMParser();
     const doc = parser.parseFromString(html, 'text/html');
-    console.log('html:', html);
-    console.log('doc:', doc);
-
     // Check if the HTML includes Spark info
     const sparkInfo = doc.querySelector('h3');
     console.log('sparkInfo:', sparkInfo);

Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ function LoginForm({ onLogin }) {`
`33`	`33`	`} else {`
`34`	`34`	`// The login failed`
`35`	`35`	`// You might want to show an error message here`
`36`		`- console.error('Failed to log in');`
	`36`	`+ console.error('Failed to log in: ', response.text, response.status);`
`37`	`37`	`setError(true);`
`38`	`38`	`}`
`39`	`39`	`};`