add delta spark packages to support delta lake

.gitignore

@@ -1,4 +1,5 @@
 .DS_Store
 .idea
 .coverage
-*_pycache__
+*_pycache__
+cdm_shared_workspace/

Dockerfile

@@ -6,9 +6,23 @@ USER root
 
 RUN apt-get update && apt-get install -y \
     # GCC required to resolve error during JupyterLab installation: psutil could not be installed from sources because gcc is not installed.
-    gcc \
+    gcc curl \
     && rm -rf /var/lib/apt/lists/*
 
+# Install jars to support delta lake spark operations
+ENV HADOOP_AWS_VER=3.3.4
+RUN curl -O https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/${HADOOP_AWS_VER}/hadoop-aws-${HADOOP_AWS_VER}.jar \
+    && mv hadoop-aws-${HADOOP_AWS_VER}.jar /opt/bitnami/spark/jars
+
+# NOTE: ensure Delta Spark jars matche python pip delta-spark version specified in the Pipfile
+ENV DELTA_SPARK_VER=3.2.0
+ENV SCALA_VER=2.12
+RUN curl -O https://repo1.maven.org/maven2/io/delta/delta-spark_${SCALA_VER}/${DELTA_SPARK_VER}/delta-spark_${SCALA_VER}-${DELTA_SPARK_VER}.jar \
+    && mv delta-spark_${SCALA_VER}-${DELTA_SPARK_VER}.jar /opt/bitnami/spark/jars
+
+Run curl -O https://repo1.maven.org/maven2/io/delta/delta-storage/${DELTA_SPARK_VER}/delta-storage-${DELTA_SPARK_VER}.jar \
+    && mv delta-storage-${DELTA_SPARK_VER}.jar /opt/bitnami/spark/jars
+
 # install pipenv
 RUN pip3 install pipenv
 
@@ -19,10 +33,8 @@ RUN pipenv sync --system
 COPY ./src/ /src
 ENV PYTHONPATH "${PYTHONPATH}:/src"
 
-COPY scripts/entrypoint.sh /opt/
-RUN chmod a+x /opt/entrypoint.sh
+COPY ./scripts/ /opt/scripts/
+RUN chmod a+x /opt/scripts/*.sh
 
 # Switch back to the original user
 USER ${ORI_USER}
-
-ENTRYPOINT ["/opt/entrypoint.sh"]

Pipfile

@@ -7,6 +7,9 @@ name = "pypi"
 jupyterlab= "==4.2.0"
 pyspark= "==3.5.1"
 boto3 = "==1.34.109"
+minio = "==7.2.7"
+delta-spark = "==3.2.0"  # should match JAR version (DELTA_SPARK_VER) specified in the Dockerfile
+pandas = "==2.2.2"
 
 [dev-packages]
 pytest = "==8.2.1"

README.md

@@ -79,7 +79,10 @@ In addition, the environment variable `SPARK_MASTER_URL` should also be configur
 ```python
 from spark.utils import get_spark_session
 
-spark = get_spark_session('TestApp')
+spark = get_spark_session(app_name)
+
+# To build spark session for Delta Lake operations, set the delta_lake parameter to True
+spark = get_spark_session(app_name, delta_lake=True)
 ```
 
 #### Manually Configuring SparkSession/SparkContext

docker-compose.yaml

@@ -5,7 +5,9 @@ version: '3'
 services:
 
   spark-master:
-    image: bitnami/spark:3.5.1
+    build:
+      context: .
+      dockerfile: Dockerfile
     container_name: spark-master
     ports:
       - "8080:8080"
@@ -51,10 +53,48 @@ services:
     environment:
       - SPARK_MASTER_URL=spark://spark-master:7077
 
+  minio:
+    image: minio/minio
+    container_name: spark-minio
+    expose:
+      - "9000"
+    ports:
+      - "9000:9000"
+      # MinIO Console is available at http://localhost:9001
+      - "9001:9001"
+    environment:
+      MINIO_ROOT_USER: minio
+      MINIO_ROOT_PASSWORD: minio123
+      MINIO_ACCESS_KEY: minio
+      MINIO_SECRET_KEY: minio123
+    healthcheck:
+      test: timeout 5s bash -c ':> /dev/tcp/127.0.0.1/9000' || exit 1
+      interval: 1s
+      timeout: 10s
+      retries: 5
+    # Note there is no bucket by default
+    command: server /data --console-address ":9001"
+
+  minio-create-bucket:
+    image: minio/mc
+    depends_on:
+      minio:
+        condition: service_healthy
+    entrypoint: >
+      bash -c "
+      mc alias set minio http://minio:9000 minio minio123 &&
+      if ! mc ls minio/delta-lake 2>/dev/null; then
+        mc mb minio/delta-lake && echo 'Bucket delta-lake created'
+      else
+        echo 'bucket delta-lake already exists'
+      fi
+      "
+
   notebook:
     build:
       context: .
       dockerfile: Dockerfile
+    entrypoint: /opt/scripts/notebook_entrypoint.sh
     container_name: spark-notebook
     ports:
       - "4041:4041"
@@ -63,4 +103,9 @@ services:
     environment:
       - NOTEBOOK_PORT=4041
       - SPARK_MASTER_URL=spark://spark-master:7077
-      - SPARK_DRIVER_HOST=spark-notebook
+      - SPARK_DRIVER_HOST=spark-notebook
+      - MINIO_URL=http://minio:9000
+      - MINIO_ACCESS_KEY=minio
+      - MINIO_SECRET_KEY=minio123
+    volumes:
+      - ./cdm_shared_workspace:/cdm_shared_workspace

src/spark/utils.py

@@ -4,12 +4,15 @@
 from pyspark.sql import SparkSession
 
 
-def get_spark_session(app_name: str, local: bool = False) -> SparkSession:
+def get_spark_session(app_name: str,
+                      local: bool = False,
+                      delta_lake: bool = False) -> SparkSession:
     """
     Helper to get and manage the `SparkSession` and keep all of our spark configuration params in one place.
 
     :param app_name: The name of the application
-    :param local: Whether to run the spark session locally or not
+    :param local: Whether to run the spark session locally or doesn't
+    :param delta_lake: build the spark session with delta lake support
 
     :return: A `SparkSession` object
     """
@@ -19,14 +22,39 @@
 
     spark_conf = SparkConf()
 
-    spark_conf.setAll(
-        [
-            (
-                "spark.master",
-                os.environ.get("SPARK_MASTER_URL", "spark://spark-master:7077"),
-            ),
-            ("spark.app.name", app_name),
-        ]
-    )
+    if delta_lake:
+
+        jars_dir = "/opt/bitnami/spark/jars/"
+        jar_files = [os.path.join(jars_dir, f) for f in os.listdir(jars_dir) if f.endswith(".jar")]
+        jars = ",".join(jar_files)
+
+        spark_conf.setAll(
+            [
+                (
+                    "spark.master",
+                    os.environ.get("SPARK_MASTER_URL", "spark://spark-master:7077"),
+                ),
+                ("spark.app.name", app_name),
+                ("spark.hadoop.fs.s3a.endpoint", os.environ.get("MINIO_URL")),
+                ("spark.hadoop.fs.s3a.access.key", os.environ.get("MINIO_ACCESS_KEY")),
+                ("spark.hadoop.fs.s3a.secret.key", os.environ.get("MINIO_SECRET_KEY")),
+                ("spark.jars", jars),
+                ("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension"),
+                ("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog"),
+                ("spark.hadoop.fs.s3a.path.style.access", "true"),
+                ("spark.hadoop.fs.s3a.connection.ssl.enabled", "false"),
+                ("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem"),
+            ]
+        )
+    else:
+        spark_conf.setAll(
+            [
+                (
+                    "spark.master",
+                    os.environ.get("SPARK_MASTER_URL", "spark://spark-master:7077"),
+                ),
+                ("spark.app.name", app_name),
+            ]
+        )
 
     return SparkSession.builder.config(conf=spark_conf).getOrCreate()