Merge pull request #12 from thalesmg/spark-sql

feat(spark-query): add endpoint to run arbitrary spark sql, make image more malleable

Author: thalesmg

spark-query/Dockerfile

@@ -1,25 +1,15 @@
-ARG BUILD_FROM=clojure:temurin-17-tools-deps-1.12.0.1530-bookworm-slim
-ARG RUN_FROM=eclipse-temurin:11-jdk-ubi9-minimal
+## Using an image with clojure so we may use `add-lib` in nREPL
+ARG RUN_FROM=clojure:temurin-17-tools-deps-1.12.0.1530-bookworm-slim
 
-FROM ${BUILD_FROM} AS builder
+FROM ${RUN_FROM}
 
 RUN mkdir -p /usr/src/app
 WORKDIR /usr/src/app
 
 # Cache layer with dependencies
 COPY ./deps.edn /usr/src/app/deps.edn
-RUN clojure -X:deps prep && \
-    clojure -P -T:build
+RUN clojure -X:deps prep
 
-COPY ./build.clj /usr/src/app/build.clj
 COPY ./src /usr/src/app/src
-RUN clojure -T:build uber && \
-    mv target/spark-query-*-standalone.jar app-standalone.jar
 
-FROM ${RUN_FROM} AS runner
-
-COPY --from=builder /usr/src/app/app-standalone.jar /opt/proxy/app-standalone.jar
-
-WORKDIR /opt/proxy
-
-CMD ["java", "-jar", "app-standalone.jar"]
+CMD ["clojure", "-M:run", "-m", "spark-query.core"]

spark-query/Dockerfile-uberjar

@@ -0,0 +1,25 @@
+ARG BUILD_FROM=clojure:temurin-17-tools-deps-1.12.0.1530-bookworm-slim
+ARG RUN_FROM=eclipse-temurin:11-jdk-ubi9-minimal
+
+FROM ${BUILD_FROM} AS builder
+
+RUN mkdir -p /usr/src/app
+WORKDIR /usr/src/app
+
+# Cache layer with dependencies
+COPY ./deps.edn /usr/src/app/deps.edn
+RUN clojure -X:deps prep && \
+    clojure -P -T:build
+
+COPY ./build.clj /usr/src/app/build.clj
+COPY ./src /usr/src/app/src
+RUN clojure -T:build uber && \
+    mv target/spark-query-*-standalone.jar app-standalone.jar
+
+FROM ${RUN_FROM} AS runner
+
+COPY --from=builder /usr/src/app/app-standalone.jar /opt/proxy/app-standalone.jar
+
+WORKDIR /opt/proxy
+
+CMD ["java", "-jar", "app-standalone.jar"]

spark-query/deps.edn

@@ -1,9 +1,13 @@
 {:deps {http-kit/http-kit {:mvn/version "2.8.0"}
         compojure/compojure {:mvn/version "1.7.1"}
+        ring/ring-devel {:mvn/version "1.14.1"}
         org.clojure/data.json {:mvn/version "2.5.1"}
 
         nrepl/nrepl {:mvn/version "1.3.1"}
         cider/cider-nrepl {:mvn/version "0.55.1"}
+        com.bhauman/rebel-readline-cljs {:mvn/version "0.1.5"}
+
+        org.apache.spark/spark-sql_2.13 {:mvn/version "3.5.5"}
 
         org.apache.iceberg/iceberg-aws-bundle {:mvn/version "1.8.1"}
         ;; software.amazon.awssdk/protocol-core {:mvn/version "2.31.14"}
@@ -19,6 +23,9 @@
  :aliases
  {:build {:deps {io.github.clojure/tools.build {:mvn/version "0.10.8"}}
           :ns-default build}
+  :run {:jvm-opts ["--add-exports" "java.base/sun.nio.ch=ALL-UNNAMED"
+                   "--add-opens=java.base/java.nio=ALL-UNNAMED"
+                   ]}
   :fmt/check {:extra-deps {cljfmt/cljfmt {:mvn/version "0.9.2"}}
               :main-opts ["--main" "cljfmt.main" "check"]}
   :fmt/fix {:extra-deps {cljfmt/cljfmt {:mvn/version "0.9.2"}}

spark-query/src/spark_query/core.clj

@@ -6,6 +6,8 @@
    [cider.nrepl :refer [cider-nrepl-handler]]
    [nrepl.server :as nrepl-server]
    [org.httpkit.server :as server]
+   [ring.middleware.reload :refer [wrap-reload]]
+   [ring.util.request :as ring-req]
    [compojure.core :refer [defroutes GET POST]])
   (:import
    [java.util Properties]
@@ -20,10 +22,12 @@
     Record]
    [org.apache.iceberg.rest RESTCatalog]
    [org.apache.iceberg.aws.s3 S3FileIOProperties]
-   [org.apache.hadoop.conf Configuration])
+   [org.apache.hadoop.conf Configuration]
+   [org.apache.spark.sql SparkSession])
   (:gen-class))
 
 (def PORT 8090)
+(def NREPL-PORT 7890)
 
 (defn- record->vec
   [record]
@@ -49,6 +53,19 @@
             (.initialize "demo" catalog-props))]
     catalog))
 
+(defn spark-session
+  []
+  (-> (SparkSession/builder)
+      (.config "spark.sql.defaultCatalog" "demo")
+      (.config "spark.sql.catalog.demo" "org.apache.iceberg.spark.SparkCatalog")
+      (.config "spark.sql.catalog.demo.type" "rest")
+      (.config "spark.sql.catalog.demo.uri" "http://iceberg-rest:8181")
+      (.config "spark.sql.catalog.demo.io-impl" "org.apache.iceberg.aws.s3.S3FileIO")
+      (.config "spark.sql.catalog.demo.warehouse" "s3://warehouse/wh/")
+      (.config "spark.sql.catalog.demo.s3.endpoint" "http://minio.net:9000")
+      (.master "local")
+      .getOrCreate))
+
 (def CATALOG (atom nil))
 
 (defn get-catalog
@@ -108,15 +125,17 @@
                      .partition
                      partition-data->vec)))))
 
-(defn handle-scan-table [ns-in table-in]
+(defn handle-scan-table
+  [ns-in table-in]
   (let [table (load-table (get-catalog) ns-in table-in)
         rows (scan-table table)
         response-body (->> rows
                            (mapv record->vec)
                            json/write-str)]
     {:body response-body}))
 
-(defn handle-table-partitions [ns-in table-in]
+(defn handle-table-partitions
+  [ns-in table-in]
   (let [table (load-table (get-catalog) ns-in table-in)
         partitions-from-meta (table-partitions-from-meta table)
         partitions-from-data (table-partitions-from-data table)
@@ -125,21 +144,38 @@
         response-body (json/write-str response)]
     {:body response-body}))
 
+(defn handle-spark-sql
+  [request]
+  (let [sql (ring-req/body-string request)
+        session (spark-session)
+        dataset (.sql session sql)
+        _ (.show dataset)
+        response-body (-> dataset
+                          .toJSON
+                          .toLocalIterator
+                          iterator-seq
+                          (->> (map json/read-str))
+                          json/write-str)]
+    {:body response-body}))
+
 (defroutes app-routes
   (GET "/scan/:ns/:table" [ns table] (handle-scan-table ns table))
-  (GET "/partitions/:ns/:table" [ns table] (handle-table-partitions ns table)))
+  (GET "/partitions/:ns/:table" [ns table] (handle-table-partitions ns table))
+  (POST "/sql" request (handle-spark-sql request)))
 
 (defn- block-forever
   []
   (while true
     (Thread/sleep 60000)))
 
 (defn -main
-  [& args]
+  [& _args]
   (try
+    (println "starting nREPL server on port" NREPL-PORT)
+    (nrepl-server/start-server :port NREPL-PORT :bind "0.0.0.0" :handler cider-nrepl-handler)
+    (println "started nREPL server on port" NREPL-PORT)
     (println "starting server on port" PORT)
-    (nrepl-server/start-server :port 7890 :bind "0.0.0.0" :handler cider-nrepl-handler)
-    (server/run-server app-routes {:port PORT})
+    (server/run-server (wrap-reload #'app-routes) {:port PORT})
     (println "started server on port" PORT)
     (block-forever)
     (catch Exception e

spark-query/vsn

@@ -1 +1 @@
-1.0.3
+1.0.4