From 90a1d4931384da1837ba490bd55c0afc8990c6c4 Mon Sep 17 00:00:00 2001 From: veerarr <39897491+veerarr@users.noreply.github.com> Date: Fri, 8 May 2026 13:16:23 -0400 Subject: [PATCH 1/6] Create test --- test | 1 + 1 file changed, 1 insertion(+) create mode 100644 test diff --git a/test b/test new file mode 100644 index 0000000..b90cb87 --- /dev/null +++ b/test @@ -0,0 +1 @@ +sgfsdtr From f1734189bb30d77062184598a7db38052a6da8f3 Mon Sep 17 00:00:00 2001 From: veerarr <39897491+veerarr@users.noreply.github.com> Date: Fri, 8 May 2026 13:56:26 -0400 Subject: [PATCH 2/6] Delete test --- test | 1 - 1 file changed, 1 deletion(-) delete mode 100644 test diff --git a/test b/test deleted file mode 100644 index b90cb87..0000000 --- a/test +++ /dev/null @@ -1 +0,0 @@ -sgfsdtr From c2b65c26e716e1cf91ae5f639cec4aa256294137 Mon Sep 17 00:00:00 2001 From: Rekha Veeraraghavan <39897491+veerarr@users.noreply.github.com> Date: Thu, 14 May 2026 12:21:43 -0400 Subject: [PATCH 3/6] Add files via upload --- .../direct-s3-jdbc-without-catalog.ipynb | 620 ++++++++++++++++++ 1 file changed, 620 insertions(+) create mode 100644 examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb diff --git a/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb b/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb new file mode 100644 index 0000000..f81a8a9 --- /dev/null +++ b/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb @@ -0,0 +1,620 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "b0ef68ac-9b8b-41d0-9e21-2f398bc6ea13", + "metadata": { + "tags": [] + }, + "source": [ + "# Direct S3 + JDBC Access Without a Catalog on EMR Serverless via SparkConnect" + ] + }, + { + "cell_type": "markdown", + "id": "e92709c0-3dfb-44f9-a4e7-e9f643fa98ff", + "metadata": { + "tags": [] + }, + "source": [ + "## Topics covered in this example\n", + "\n", + "1) [Prerequisites](#prerequisites)
\n", + "2) [Configuring SparkConnect Session](#configure_spark_connect)
\n", + "3) [Reading Files Directly from S3](#read_s3)
\n", + "    a) [Parquet Files](#parquet)
\n", + "    b) [CSV Files](#csv)
\n", + "    c) [JSON Files](#json)
\n", + "4) [Reading from MySQL RDS via JDBC](#jdbc)
\n", + "5) [Cross-Source Joins](#cross_source_joins)
\n", + "6) [Writing Results Back to S3](#write_s3)
\n", + "7) [Cleanup](#cleanup)
" + ] + }, + { + "cell_type": "markdown", + "id": "f0bac8b2-99cc-40e8-bce9-ac52e0c4a135", + "metadata": { + "tags": [] + }, + "source": [ + "***\n", + "\n", + "\n", + "## Prerequisites\n", + "
\n", + "NOTE : In order to execute this notebook successfully as is, please ensure the following prerequisites are completed.
\n", + "\n", + "* This notebook demonstrates connecting from a **local Jupyter notebook** to **EMR Serverless** via **SparkConnect** — no catalog (Glue, Hive, or Iceberg) is used.\n", + "* EMR Serverless application with **Spark 3.4+** and **SparkConnect endpoint** enabled.\n", + "* A MySQL RDS instance accessible from the EMR Serverless application's VPC.\n", + "* S3 bucket with sample data files (Parquet, CSV, JSON).\n", + "* The MySQL JDBC driver JAR available in S3 (e.g., `mysql-connector-java-8.0.33.jar`).\n", + "* IAM role (job execution role) with permissions to:\n", + " - Read/write to the S3 bucket\n", + " - Access the EMR Serverless application\n", + "\n", + "* Install the required Python packages on your local machine:\n", + "\n", + " pip install pyspark==3.4.0\n", + "\n", + "* Set the following environment variables for SparkConnect:\n", + "\n", + " export EMR_SERVERLESS_APP_ID=\n", + " export EMR_SERVERLESS_ENDPOINT=\n", + "\n", + "***" + ] + }, + { + "cell_type": "markdown", + "id": "04e47cf4-bebd-4425-9aed-44ddc8a35d76", + "metadata": { + "tags": [] + }, + "source": [ + "## Introduction\n", + "This notebook demonstrates a **catalog-free** approach to data access on EMR Serverless. Instead of registering tables in AWS Glue Data Catalog, Hive Metastore, or Apache Iceberg, we read files directly from S3 paths and query relational databases via JDBC.\n", + "\n", + "This approach is useful when:\n", + "- You need quick, ad-hoc exploration of data files without catalog setup overhead.\n", + "- You want to join data across S3 files and relational databases in a single Spark session.\n", + "- Your data is not yet organized into a formal lakehouse architecture.\n", + "\n", + "We connect from a local Jupyter notebook to EMR Serverless using **SparkConnect**, which provides a thin client connection to a remote Spark cluster without requiring a full Spark installation locally.\n", + "\n", + "***" + ] + }, + { + "cell_type": "markdown", + "id": "deca2149-344c-4baf-a3b6-4bd19d7afbcb", + "metadata": { + "tags": [] + }, + "source": [ + "## Setup\n", + "Ensure you have the following S3 paths ready with sample data. Update the placeholders below:\n", + "\n", + "- **S3 Parquet path:** s3://my-s3-bucket/data/parquet/\n", + "- **S3 CSV path:** s3://my-s3-bucket/data/csv/\n", + "- **S3 JSON path:** s3://my-s3-bucket/data/json/\n", + "- **S3 output path:** s3://my-s3-bucket/output/\n", + "- **JDBC URL:** jdbc:mysql://your-rds-endpoint:3306/your_database\n", + "- **JDBC driver JAR:** s3://my-s3-bucket/jars/mysql-connector-java-8.0.33.jar\n", + "\n", + "***" + ] + }, + { + "cell_type": "markdown", + "id": "5af5f2a5-bd04-461b-8617-1f9fcfe92e83", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Configuring SparkConnect Session\n", + "\n", + "Connect to EMR Serverless from your local Jupyter notebook using SparkConnect. This creates a remote Spark session — no local Spark cluster is needed.\n", + "\n", + "Update your-spark-connect-endpoint with your EMR Serverless SparkConnect endpoint URL." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b42dd813-a4a5-42da-b29c-bbf9e02d9046", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "from pyspark.sql import SparkSession\n", + "\n", + "spark = SparkSession.builder \\\n", + " .remote(\"sc://\") \\\n", + " .appName(\"DirectS3JDBCWithoutCatalog\") \\\n", + " .getOrCreate()\n", + "\n", + "print(f\"Spark version: {spark.version}\")\n", + "print(\"SparkConnect session established successfully.\")" + ] + }, + { + "cell_type": "markdown", + "id": "a0b1c2d3-e4f5-6789-abcd-ef0123456789", + "metadata": { + "tags": [] + }, + "source": [ + "**Define common variables used throughout the notebook.**\n", + "\n", + "Update the placeholder values below with your actual bucket name, RDS endpoint, and credentials." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c1d2e3f4-a5b6-7890-cdef-123456789abc", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "# S3 paths\n", + "S3_BUCKET = \"my-s3-bucket\"\n", + "PARQUET_PATH = f\"s3://{S3_BUCKET}/data/parquet/\"\n", + "CSV_PATH = f\"s3://{S3_BUCKET}/data/csv/\"\n", + "JSON_PATH = f\"s3://{S3_BUCKET}/data/json/\"\n", + "OUTPUT_PATH = f\"s3://{S3_BUCKET}/output/\"\n", + "\n", + "# JDBC connection details\n", + "JDBC_URL = \"jdbc:mysql://:3306/\"\n", + "JDBC_USER = \"\"\n", + "JDBC_PASSWORD = \"\"\n", + "JDBC_DRIVER = \"com.mysql.cj.jdbc.Driver\"" + ] + }, + { + "cell_type": "markdown", + "id": "8725d217-b26c-4fa8-84de-18969f5260c7", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Reading Files Directly from S3\n", + "\n", + "We read files directly from S3 paths using `spark.read` — **no catalog registration required**. Spark infers or uses the specified schema to load data as DataFrames." + ] + }, + { + "cell_type": "markdown", + "id": "58730a9c-431d-49b8-985f-ef2177f439a2", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "### Parquet Files\n", + "Parquet is a columnar storage format. Schema is embedded in the file, so Spark auto-infers it." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "7c898a47-33eb-4882-9990-e2dbaf257d68", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_parquet = spark.read.parquet(PARQUET_PATH)\n", + "\n", + "print(f\"Parquet record count: {df_parquet.count()}\")\n", + "df_parquet.printSchema()\n", + "df_parquet.show(5)" + ] + }, + { + "cell_type": "markdown", + "id": "bb85126b-117e-4ddd-b0b6-8652e6025e95", + "metadata": { + "tags": [] + }, + "source": [ + "**Create a temporary view so we can use SQL syntax — still no catalog involved.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4d1ac4ee-7b08-4842-9d53-99bb5b22cfa6", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_parquet.createOrReplaceTempView(\"orders\")\n", + "\n", + "spark.sql(\"SELECT * FROM orders LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "3190ca8c-803c-42f4-98b9-62c26be29959", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "### CSV Files\n", + "CSV files require specifying options like header and schema inference." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4054056c-fae1-44b9-ba89-3243ce0d8ea5", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_csv = spark.read \\\n", + " .option(\"header\", \"true\") \\\n", + " .option(\"inferSchema\", \"true\") \\\n", + " .csv(CSV_PATH)\n", + "\n", + "print(f\"CSV record count: {df_csv.count()}\")\n", + "df_csv.printSchema()\n", + "df_csv.show(5)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "653dc08f-b2be-4bec-afce-4b9a88f19712", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_csv.createOrReplaceTempView(\"customers\")\n", + "\n", + "spark.sql(\"SELECT * FROM customers LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "4504fd86-29c5-4c5f-a71d-f7e9643db3c8", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "### JSON Files\n", + "JSON files are semi-structured. Spark infers the schema automatically." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "cdf9839a-a352-4aca-9de7-59a784d4373a", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_json = spark.read.json(JSON_PATH)\n", + "\n", + "print(f\"JSON record count: {df_json.count()}\")\n", + "df_json.printSchema()\n", + "df_json.show(5)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "19bdb3bd-b7d3-458c-8fff-2e0dbc035eb7", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_json.createOrReplaceTempView(\"products\")\n", + "\n", + "spark.sql(\"SELECT * FROM products LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "85f8e445-d0a1-4bb1-9dfe-f1bf0ff946e1", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Reading from MySQL RDS via JDBC\n", + "\n", + "We connect directly to a MySQL RDS instance using JDBC. No catalog registration is needed — Spark reads the table directly from the database.\n", + "\n", + "Update the JDBC connection details defined in the Setup section above." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9b34f116-db73-4f80-92f4-10400a717066", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_mysql = spark.read \\\n", + " .format(\"jdbc\") \\\n", + " .option(\"url\", JDBC_URL) \\\n", + " .option(\"dbtable\", \"inventory\") \\\n", + " .option(\"user\", JDBC_USER) \\\n", + " .option(\"password\", JDBC_PASSWORD) \\\n", + " .option(\"driver\", JDBC_DRIVER) \\\n", + " .load()\n", + "\n", + "print(f\"MySQL record count: {df_mysql.count()}\")\n", + "df_mysql.printSchema()\n", + "df_mysql.show(5)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "fb7863b4-a453-490d-a638-7cb29368a76f", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_mysql.createOrReplaceTempView(\"inventory\")\n", + "\n", + "spark.sql(\"SELECT * FROM inventory LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "34824f0a-4334-45d3-8a0d-9a575875488a", + "metadata": { + "tags": [] + }, + "source": [ + "**You can also push down a query to the database to reduce data transfer.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "91cd1eac-c422-4c44-bdf7-d23284bc87d6", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "query = \"(SELECT product_id, quantity, warehouse_location FROM inventory WHERE quantity > 100) AS filtered_inventory\"\n", + "\n", + "df_filtered = spark.read \\\n", + " .format(\"jdbc\") \\\n", + " .option(\"url\", JDBC_URL) \\\n", + " .option(\"dbtable\", query) \\\n", + " .option(\"user\", JDBC_USER) \\\n", + " .option(\"password\", JDBC_PASSWORD) \\\n", + " .option(\"driver\", JDBC_DRIVER) \\\n", + " .load()\n", + "\n", + "df_filtered.show(5)" + ] + }, + { + "cell_type": "markdown", + "id": "9d19f6c3-9f08-402f-85ac-8aa487f51c05", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Cross-Source Joins\n", + "\n", + "Now we join data across all sources — S3 Parquet, S3 CSV, S3 JSON, and MySQL RDS — in a single Spark SQL query. This is the power of the catalog-free approach: temporary views from any source can be joined seamlessly." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d338d82a-f7b0-4ef7-8e1b-ebb2c88c9081", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "cross_join_df = spark.sql(\"\"\"\n", + " SELECT \n", + " c.customer_id,\n", + " c.customer_name,\n", + " o.order_id,\n", + " o.order_date,\n", + " p.product_name,\n", + " p.category,\n", + " o.amount,\n", + " i.quantity AS inventory_quantity,\n", + " i.warehouse_location\n", + " FROM orders o\n", + " JOIN customers c ON o.customer_id = c.customer_id\n", + " JOIN products p ON o.product_id = p.product_id\n", + " JOIN inventory i ON o.product_id = i.product_id\n", + "\"\"\")\n", + "\n", + "print(f\"Cross-source join result count: {cross_join_df.count()}\")\n", + "cross_join_df.show(10)" + ] + }, + { + "cell_type": "markdown", + "id": "fb22bf8b-3c83-45b1-aac6-c37e61b41e20", + "metadata": { + "tags": [] + }, + "source": [ + "**Run aggregations on the joined data.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "906629e3-5c0d-4ab6-a2dc-5d8a205c2288", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "spark.sql(\"\"\"\n", + " SELECT \n", + " p.category,\n", + " COUNT(o.order_id) AS total_orders,\n", + " ROUND(SUM(o.amount), 2) AS total_revenue,\n", + " ROUND(AVG(o.amount), 2) AS avg_order_value\n", + " FROM orders o\n", + " JOIN products p ON o.product_id = p.product_id\n", + " GROUP BY p.category\n", + " ORDER BY total_revenue DESC\n", + "\"\"\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "1d035fa4-6fc5-4c1c-b846-3c386ca0222b", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Writing Results Back to S3\n", + "\n", + "Write the cross-source join results back to S3 in Parquet format — again, no catalog involved. The output is simply a set of files in an S3 path." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3073f8a6-121f-4768-b112-7ebae68d28f4", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "cross_join_df.write \\\n", + " .mode(\"overwrite\") \\\n", + " .parquet(f\"{OUTPUT_PATH}cross_source_join/\")\n", + "\n", + "print(f\"Results written to {OUTPUT_PATH}cross_source_join/\")" + ] + }, + { + "cell_type": "markdown", + "id": "d4f8bc24-5ad0-4b1b-8e94-34d65b6d62c3", + "metadata": { + "tags": [] + }, + "source": [ + "**Verify the written data by reading it back.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3dce569e-93c4-43c3-a962-c72fcc83f8ef", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_verify = spark.read.parquet(f\"{OUTPUT_PATH}cross_source_join/\")\n", + "\n", + "print(f\"Verified record count: {df_verify.count()}\")\n", + "df_verify.show(5)" + ] + }, + { + "cell_type": "markdown", + "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "metadata": { + "tags": [] + }, + "source": [ + "**You can also write in other formats like CSV or JSON.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b1c2d3e4-f5a6-7890-bcde-f12345678901", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "# Write as CSV\n", + "cross_join_df.write \\\n", + " .mode(\"overwrite\") \\\n", + " .option(\"header\", \"true\") \\\n", + " .csv(f\"{OUTPUT_PATH}cross_source_join_csv/\")\n", + "\n", + "print(f\"CSV results written to {OUTPUT_PATH}cross_source_join_csv/\")" + ] + }, + { + "cell_type": "markdown", + "id": "c2d3e4f5-a6b7-8901-cdef-234567890abc", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Cleanup\n", + "\n", + "Stop the Spark session to release EMR Serverless resources." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d3e4f5a6-b7c8-9012-defg-345678901bcd", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "spark.stop()\n", + "print(\"SparkConnect session stopped. EMR Serverless resources released.\")" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "pygments_lexer": "ipython3", + "version": "3.9.0" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} From fab9b3ae8927386c69366900906ea35144d03386 Mon Sep 17 00:00:00 2001 From: Rekha Veeraraghavan <39897491+veerarr@users.noreply.github.com> Date: Thu, 14 May 2026 13:58:21 -0400 Subject: [PATCH 4/6] Update direct-s3-jdbc-without-catalog.ipynb --- examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb b/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb index f81a8a9..139d22f 100644 --- a/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb +++ b/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb @@ -60,8 +60,7 @@ "\n", "* Set the following environment variables for SparkConnect:\n", "\n", - " export EMR_SERVERLESS_APP_ID=\n", - " export EMR_SERVERLESS_ENDPOINT=\n", + " export EMR_SERVERLESS_APP_ID=\n" "\n", "***" ] From faff3e5fe13ba3f55f9986a3fd012d59db5da56f Mon Sep 17 00:00:00 2001 From: Rekha Veeraraghavan <39897491+veerarr@users.noreply.github.com> Date: Thu, 14 May 2026 13:59:11 -0400 Subject: [PATCH 5/6] Delete examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb --- .../direct-s3-jdbc-without-catalog.ipynb | 619 ------------------ 1 file changed, 619 deletions(-) delete mode 100644 examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb diff --git a/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb b/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb deleted file mode 100644 index 139d22f..0000000 --- a/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb +++ /dev/null @@ -1,619 +0,0 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "id": "b0ef68ac-9b8b-41d0-9e21-2f398bc6ea13", - "metadata": { - "tags": [] - }, - "source": [ - "# Direct S3 + JDBC Access Without a Catalog on EMR Serverless via SparkConnect" - ] - }, - { - "cell_type": "markdown", - "id": "e92709c0-3dfb-44f9-a4e7-e9f643fa98ff", - "metadata": { - "tags": [] - }, - "source": [ - "## Topics covered in this example\n", - "\n", - "1) [Prerequisites](#prerequisites)
\n", - "2) [Configuring SparkConnect Session](#configure_spark_connect)
\n", - "3) [Reading Files Directly from S3](#read_s3)
\n", - "    a) [Parquet Files](#parquet)
\n", - "    b) [CSV Files](#csv)
\n", - "    c) [JSON Files](#json)
\n", - "4) [Reading from MySQL RDS via JDBC](#jdbc)
\n", - "5) [Cross-Source Joins](#cross_source_joins)
\n", - "6) [Writing Results Back to S3](#write_s3)
\n", - "7) [Cleanup](#cleanup)
" - ] - }, - { - "cell_type": "markdown", - "id": "f0bac8b2-99cc-40e8-bce9-ac52e0c4a135", - "metadata": { - "tags": [] - }, - "source": [ - "***\n", - "\n", - "\n", - "## Prerequisites\n", - "
\n", - "NOTE : In order to execute this notebook successfully as is, please ensure the following prerequisites are completed.
\n", - "\n", - "* This notebook demonstrates connecting from a **local Jupyter notebook** to **EMR Serverless** via **SparkConnect** — no catalog (Glue, Hive, or Iceberg) is used.\n", - "* EMR Serverless application with **Spark 3.4+** and **SparkConnect endpoint** enabled.\n", - "* A MySQL RDS instance accessible from the EMR Serverless application's VPC.\n", - "* S3 bucket with sample data files (Parquet, CSV, JSON).\n", - "* The MySQL JDBC driver JAR available in S3 (e.g., `mysql-connector-java-8.0.33.jar`).\n", - "* IAM role (job execution role) with permissions to:\n", - " - Read/write to the S3 bucket\n", - " - Access the EMR Serverless application\n", - "\n", - "* Install the required Python packages on your local machine:\n", - "\n", - " pip install pyspark==3.4.0\n", - "\n", - "* Set the following environment variables for SparkConnect:\n", - "\n", - " export EMR_SERVERLESS_APP_ID=\n" - "\n", - "***" - ] - }, - { - "cell_type": "markdown", - "id": "04e47cf4-bebd-4425-9aed-44ddc8a35d76", - "metadata": { - "tags": [] - }, - "source": [ - "## Introduction\n", - "This notebook demonstrates a **catalog-free** approach to data access on EMR Serverless. Instead of registering tables in AWS Glue Data Catalog, Hive Metastore, or Apache Iceberg, we read files directly from S3 paths and query relational databases via JDBC.\n", - "\n", - "This approach is useful when:\n", - "- You need quick, ad-hoc exploration of data files without catalog setup overhead.\n", - "- You want to join data across S3 files and relational databases in a single Spark session.\n", - "- Your data is not yet organized into a formal lakehouse architecture.\n", - "\n", - "We connect from a local Jupyter notebook to EMR Serverless using **SparkConnect**, which provides a thin client connection to a remote Spark cluster without requiring a full Spark installation locally.\n", - "\n", - "***" - ] - }, - { - "cell_type": "markdown", - "id": "deca2149-344c-4baf-a3b6-4bd19d7afbcb", - "metadata": { - "tags": [] - }, - "source": [ - "## Setup\n", - "Ensure you have the following S3 paths ready with sample data. Update the placeholders below:\n", - "\n", - "- **S3 Parquet path:** s3://my-s3-bucket/data/parquet/\n", - "- **S3 CSV path:** s3://my-s3-bucket/data/csv/\n", - "- **S3 JSON path:** s3://my-s3-bucket/data/json/\n", - "- **S3 output path:** s3://my-s3-bucket/output/\n", - "- **JDBC URL:** jdbc:mysql://your-rds-endpoint:3306/your_database\n", - "- **JDBC driver JAR:** s3://my-s3-bucket/jars/mysql-connector-java-8.0.33.jar\n", - "\n", - "***" - ] - }, - { - "cell_type": "markdown", - "id": "5af5f2a5-bd04-461b-8617-1f9fcfe92e83", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "## Configuring SparkConnect Session\n", - "\n", - "Connect to EMR Serverless from your local Jupyter notebook using SparkConnect. This creates a remote Spark session — no local Spark cluster is needed.\n", - "\n", - "Update your-spark-connect-endpoint with your EMR Serverless SparkConnect endpoint URL." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "b42dd813-a4a5-42da-b29c-bbf9e02d9046", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "from pyspark.sql import SparkSession\n", - "\n", - "spark = SparkSession.builder \\\n", - " .remote(\"sc://\") \\\n", - " .appName(\"DirectS3JDBCWithoutCatalog\") \\\n", - " .getOrCreate()\n", - "\n", - "print(f\"Spark version: {spark.version}\")\n", - "print(\"SparkConnect session established successfully.\")" - ] - }, - { - "cell_type": "markdown", - "id": "a0b1c2d3-e4f5-6789-abcd-ef0123456789", - "metadata": { - "tags": [] - }, - "source": [ - "**Define common variables used throughout the notebook.**\n", - "\n", - "Update the placeholder values below with your actual bucket name, RDS endpoint, and credentials." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "c1d2e3f4-a5b6-7890-cdef-123456789abc", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "# S3 paths\n", - "S3_BUCKET = \"my-s3-bucket\"\n", - "PARQUET_PATH = f\"s3://{S3_BUCKET}/data/parquet/\"\n", - "CSV_PATH = f\"s3://{S3_BUCKET}/data/csv/\"\n", - "JSON_PATH = f\"s3://{S3_BUCKET}/data/json/\"\n", - "OUTPUT_PATH = f\"s3://{S3_BUCKET}/output/\"\n", - "\n", - "# JDBC connection details\n", - "JDBC_URL = \"jdbc:mysql://:3306/\"\n", - "JDBC_USER = \"\"\n", - "JDBC_PASSWORD = \"\"\n", - "JDBC_DRIVER = \"com.mysql.cj.jdbc.Driver\"" - ] - }, - { - "cell_type": "markdown", - "id": "8725d217-b26c-4fa8-84de-18969f5260c7", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "## Reading Files Directly from S3\n", - "\n", - "We read files directly from S3 paths using `spark.read` — **no catalog registration required**. Spark infers or uses the specified schema to load data as DataFrames." - ] - }, - { - "cell_type": "markdown", - "id": "58730a9c-431d-49b8-985f-ef2177f439a2", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "### Parquet Files\n", - "Parquet is a columnar storage format. Schema is embedded in the file, so Spark auto-infers it." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "7c898a47-33eb-4882-9990-e2dbaf257d68", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_parquet = spark.read.parquet(PARQUET_PATH)\n", - "\n", - "print(f\"Parquet record count: {df_parquet.count()}\")\n", - "df_parquet.printSchema()\n", - "df_parquet.show(5)" - ] - }, - { - "cell_type": "markdown", - "id": "bb85126b-117e-4ddd-b0b6-8652e6025e95", - "metadata": { - "tags": [] - }, - "source": [ - "**Create a temporary view so we can use SQL syntax — still no catalog involved.**" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "4d1ac4ee-7b08-4842-9d53-99bb5b22cfa6", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_parquet.createOrReplaceTempView(\"orders\")\n", - "\n", - "spark.sql(\"SELECT * FROM orders LIMIT 10\").show()" - ] - }, - { - "cell_type": "markdown", - "id": "3190ca8c-803c-42f4-98b9-62c26be29959", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "### CSV Files\n", - "CSV files require specifying options like header and schema inference." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "4054056c-fae1-44b9-ba89-3243ce0d8ea5", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_csv = spark.read \\\n", - " .option(\"header\", \"true\") \\\n", - " .option(\"inferSchema\", \"true\") \\\n", - " .csv(CSV_PATH)\n", - "\n", - "print(f\"CSV record count: {df_csv.count()}\")\n", - "df_csv.printSchema()\n", - "df_csv.show(5)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "653dc08f-b2be-4bec-afce-4b9a88f19712", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_csv.createOrReplaceTempView(\"customers\")\n", - "\n", - "spark.sql(\"SELECT * FROM customers LIMIT 10\").show()" - ] - }, - { - "cell_type": "markdown", - "id": "4504fd86-29c5-4c5f-a71d-f7e9643db3c8", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "### JSON Files\n", - "JSON files are semi-structured. Spark infers the schema automatically." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "cdf9839a-a352-4aca-9de7-59a784d4373a", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_json = spark.read.json(JSON_PATH)\n", - "\n", - "print(f\"JSON record count: {df_json.count()}\")\n", - "df_json.printSchema()\n", - "df_json.show(5)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "19bdb3bd-b7d3-458c-8fff-2e0dbc035eb7", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_json.createOrReplaceTempView(\"products\")\n", - "\n", - "spark.sql(\"SELECT * FROM products LIMIT 10\").show()" - ] - }, - { - "cell_type": "markdown", - "id": "85f8e445-d0a1-4bb1-9dfe-f1bf0ff946e1", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "## Reading from MySQL RDS via JDBC\n", - "\n", - "We connect directly to a MySQL RDS instance using JDBC. No catalog registration is needed — Spark reads the table directly from the database.\n", - "\n", - "Update the JDBC connection details defined in the Setup section above." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "9b34f116-db73-4f80-92f4-10400a717066", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_mysql = spark.read \\\n", - " .format(\"jdbc\") \\\n", - " .option(\"url\", JDBC_URL) \\\n", - " .option(\"dbtable\", \"inventory\") \\\n", - " .option(\"user\", JDBC_USER) \\\n", - " .option(\"password\", JDBC_PASSWORD) \\\n", - " .option(\"driver\", JDBC_DRIVER) \\\n", - " .load()\n", - "\n", - "print(f\"MySQL record count: {df_mysql.count()}\")\n", - "df_mysql.printSchema()\n", - "df_mysql.show(5)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "fb7863b4-a453-490d-a638-7cb29368a76f", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_mysql.createOrReplaceTempView(\"inventory\")\n", - "\n", - "spark.sql(\"SELECT * FROM inventory LIMIT 10\").show()" - ] - }, - { - "cell_type": "markdown", - "id": "34824f0a-4334-45d3-8a0d-9a575875488a", - "metadata": { - "tags": [] - }, - "source": [ - "**You can also push down a query to the database to reduce data transfer.**" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "91cd1eac-c422-4c44-bdf7-d23284bc87d6", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "query = \"(SELECT product_id, quantity, warehouse_location FROM inventory WHERE quantity > 100) AS filtered_inventory\"\n", - "\n", - "df_filtered = spark.read \\\n", - " .format(\"jdbc\") \\\n", - " .option(\"url\", JDBC_URL) \\\n", - " .option(\"dbtable\", query) \\\n", - " .option(\"user\", JDBC_USER) \\\n", - " .option(\"password\", JDBC_PASSWORD) \\\n", - " .option(\"driver\", JDBC_DRIVER) \\\n", - " .load()\n", - "\n", - "df_filtered.show(5)" - ] - }, - { - "cell_type": "markdown", - "id": "9d19f6c3-9f08-402f-85ac-8aa487f51c05", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "## Cross-Source Joins\n", - "\n", - "Now we join data across all sources — S3 Parquet, S3 CSV, S3 JSON, and MySQL RDS — in a single Spark SQL query. This is the power of the catalog-free approach: temporary views from any source can be joined seamlessly." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "d338d82a-f7b0-4ef7-8e1b-ebb2c88c9081", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "cross_join_df = spark.sql(\"\"\"\n", - " SELECT \n", - " c.customer_id,\n", - " c.customer_name,\n", - " o.order_id,\n", - " o.order_date,\n", - " p.product_name,\n", - " p.category,\n", - " o.amount,\n", - " i.quantity AS inventory_quantity,\n", - " i.warehouse_location\n", - " FROM orders o\n", - " JOIN customers c ON o.customer_id = c.customer_id\n", - " JOIN products p ON o.product_id = p.product_id\n", - " JOIN inventory i ON o.product_id = i.product_id\n", - "\"\"\")\n", - "\n", - "print(f\"Cross-source join result count: {cross_join_df.count()}\")\n", - "cross_join_df.show(10)" - ] - }, - { - "cell_type": "markdown", - "id": "fb22bf8b-3c83-45b1-aac6-c37e61b41e20", - "metadata": { - "tags": [] - }, - "source": [ - "**Run aggregations on the joined data.**" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "906629e3-5c0d-4ab6-a2dc-5d8a205c2288", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "spark.sql(\"\"\"\n", - " SELECT \n", - " p.category,\n", - " COUNT(o.order_id) AS total_orders,\n", - " ROUND(SUM(o.amount), 2) AS total_revenue,\n", - " ROUND(AVG(o.amount), 2) AS avg_order_value\n", - " FROM orders o\n", - " JOIN products p ON o.product_id = p.product_id\n", - " GROUP BY p.category\n", - " ORDER BY total_revenue DESC\n", - "\"\"\").show()" - ] - }, - { - "cell_type": "markdown", - "id": "1d035fa4-6fc5-4c1c-b846-3c386ca0222b", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "## Writing Results Back to S3\n", - "\n", - "Write the cross-source join results back to S3 in Parquet format — again, no catalog involved. The output is simply a set of files in an S3 path." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "3073f8a6-121f-4768-b112-7ebae68d28f4", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "cross_join_df.write \\\n", - " .mode(\"overwrite\") \\\n", - " .parquet(f\"{OUTPUT_PATH}cross_source_join/\")\n", - "\n", - "print(f\"Results written to {OUTPUT_PATH}cross_source_join/\")" - ] - }, - { - "cell_type": "markdown", - "id": "d4f8bc24-5ad0-4b1b-8e94-34d65b6d62c3", - "metadata": { - "tags": [] - }, - "source": [ - "**Verify the written data by reading it back.**" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "3dce569e-93c4-43c3-a962-c72fcc83f8ef", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "df_verify = spark.read.parquet(f\"{OUTPUT_PATH}cross_source_join/\")\n", - "\n", - "print(f\"Verified record count: {df_verify.count()}\")\n", - "df_verify.show(5)" - ] - }, - { - "cell_type": "markdown", - "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", - "metadata": { - "tags": [] - }, - "source": [ - "**You can also write in other formats like CSV or JSON.**" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "b1c2d3e4-f5a6-7890-bcde-f12345678901", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "# Write as CSV\n", - "cross_join_df.write \\\n", - " .mode(\"overwrite\") \\\n", - " .option(\"header\", \"true\") \\\n", - " .csv(f\"{OUTPUT_PATH}cross_source_join_csv/\")\n", - "\n", - "print(f\"CSV results written to {OUTPUT_PATH}cross_source_join_csv/\")" - ] - }, - { - "cell_type": "markdown", - "id": "c2d3e4f5-a6b7-8901-cdef-234567890abc", - "metadata": { - "tags": [] - }, - "source": [ - "\n", - "## Cleanup\n", - "\n", - "Stop the Spark session to release EMR Serverless resources." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "d3e4f5a6-b7c8-9012-defg-345678901bcd", - "metadata": { - "tags": [] - }, - "outputs": [], - "source": [ - "spark.stop()\n", - "print(\"SparkConnect session stopped. EMR Serverless resources released.\")" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "pygments_lexer": "ipython3", - "version": "3.9.0" - } - }, - "nbformat": 4, - "nbformat_minor": 5 -} From ecbd4ff7373e902119885f65800e8d677fc74cf0 Mon Sep 17 00:00:00 2001 From: Rekha Veeraraghavan <39897491+veerarr@users.noreply.github.com> Date: Thu, 14 May 2026 15:34:14 -0400 Subject: [PATCH 6/6] Add files via upload --- .../direct-s3-jdbc-without-catalog.ipynb | 839 ++++++++++++++++++ 1 file changed, 839 insertions(+) create mode 100644 examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb diff --git a/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb b/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb new file mode 100644 index 0000000..302d606 --- /dev/null +++ b/examples/spark-connect/direct-s3-jdbc-without-catalog.ipynb @@ -0,0 +1,839 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "b0ef68ac-9b8b-41d0-9e21-2f398bc6ea13", + "metadata": { + "tags": [] + }, + "source": [ + "# Direct S3 + JDBC Access Without a Catalog on EMR Serverless via SparkConnect" + ] + }, + { + "cell_type": "markdown", + "id": "e92709c0-3dfb-44f9-a4e7-e9f643fa98ff", + "metadata": { + "tags": [] + }, + "source": [ + "## Topics covered in this example\n", + "\n", + "1) [Prerequisites](#prerequisites)
\n", + "2) [Configuring SparkConnect Session](#configure_spark_connect)
\n", + "3) [Reading Files Directly from S3](#read_s3)
\n", + "    a) [Parquet Files](#parquet)
\n", + "    b) [CSV Files](#csv)
\n", + "    c) [JSON Files](#json)
\n", + "4) [Reading from MySQL RDS via JDBC](#jdbc)
\n", + "5) [Cross-Source Joins](#cross_source_joins)
\n", + "6) [Writing Results Back to S3](#write_s3)
\n", + "7) [Cleanup](#cleanup)
" + ] + }, + { + "cell_type": "markdown", + "id": "f0bac8b2-99cc-40e8-bce9-ac52e0c4a135", + "metadata": { + "tags": [] + }, + "source": [ + "***\n", + "\n", + "\n", + "## Prerequisites\n", + "
\n", + "NOTE : In order to execute this notebook successfully as is, please ensure the following prerequisites are completed.
\n", + "\n", + "**What This Scenario Covers:**\n", + "\n", + "Connect from a local Jupyter notebook to EMR Serverless via SparkConnect, query files directly from S3 (Parquet, CSV, JSON) and a MySQL RDS instance via JDBC, perform cross-source joins — all **without** registering tables in any catalog (Glue, Hive, or Iceberg).\n", + "\n", + "**Requirements:**\n", + "\n", + "* An EMR Serverless application on emr-7.13.0 or later.\n", + "* A MySQL RDS instance accessible from the EMR Serverless application's VPC.\n", + "* S3 bucket with sample data files (Parquet, CSV, JSON).\n", + "* The MySQL JDBC driver JAR available in S3 (e.g., `mysql-connector-java-8.0.33.jar`).\n", + "* IAM role (job execution role) with permissions to:\n", + " - Read/write to the S3 bucket\n", + " - Access the EMR Serverless application\n", + "\n", + "* Install the required Python packages on your local machine:\n", + "\n", + " pip install pyspark==3.5.6\n", + " pip install pandas numpy pyarrow grpcio grpcio-status googleapis-common-protos\n", + "\n", + "* Set the following environment variables for SparkConnect:\n", + "\n", + " export EMR_SERVERLESS_APP_ID=\n", + "\n", + "***" + ] + }, + { + "cell_type": "markdown", + "id": "04e47cf4-bebd-4425-9aed-44ddc8a35d76", + "metadata": { + "tags": [] + }, + "source": [ + "## Introduction\n", + "This notebook demonstrates a **catalog-free** approach to data access on EMR Serverless. Instead of registering tables in AWS Glue Data Catalog, Hive Metastore, or Apache Iceberg, we read files directly from S3 paths and query relational databases via JDBC.\n", + "\n", + "This approach is useful when:\n", + "- You need quick, ad-hoc exploration of data files without catalog setup overhead.\n", + "- You want to join data across S3 files and relational databases in a single Spark session.\n", + "- Your data is not yet organized into a formal lakehouse architecture.\n", + "\n", + "We connect from a local Jupyter notebook to EMR Serverless using **SparkConnect**, which provides a thin client connection to a remote Spark cluster without requiring a full Spark installation locally.\n", + "\n", + "***" + ] + }, + { + "cell_type": "markdown", + "id": "deca2149-344c-4baf-a3b6-4bd19d7afbcb", + "metadata": { + "tags": [] + }, + "source": [ + "## Setup\n", + "Ensure you have the following S3 paths ready with sample data. Update the placeholders below:\n", + "\n", + "- **S3 Parquet path:** s3://my-s3-bucket/data/parquet/\n", + "- **S3 CSV path:** s3://my-s3-bucket/data/csv/\n", + "- **S3 JSON path:** s3://my-s3-bucket/data/json/\n", + "- **S3 output path:** s3://my-s3-bucket/output/\n", + "- **JDBC URL:** jdbc:mysql://your-rds-endpoint:3306/your_database\n", + "- **JDBC driver JAR:** s3://my-s3-bucket/jars/mysql-connector-java-8.0.33.jar\n", + "\n", + "***" + ] + }, + { + "cell_type": "markdown", + "id": "5af5f2a5-bd04-461b-8617-1f9fcfe92e83", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Configuring SparkConnect Session\n", + "\n", + "Connect to EMR Serverless from your local Jupyter notebook using SparkConnect. This creates a remote Spark session — no local Spark cluster is needed." + ] + }, + { + "cell_type": "markdown", + "id": "cb5bf380", + "metadata": {}, + "source": [ + "### Create EMR Serverless Application \n", + "\n", + "To create a Spark Connect–enabled application, see the command in the README or the AWS documentation linked above." + ] + }, + { + "cell_type": "markdown", + "id": "731defda", + "metadata": {}, + "source": [ + "```bash\n", + "aws emr-serverless create-application \\\n", + " --name \"direct-s3-jdbc-sparkconnect\" \\\n", + " --type SPARK \\\n", + " --release-label emr-7.13.0 \\\n", + " --initial-capacity '{\n", + " \"Driver\": { \"workerCount\": 2, \"workerConfiguration\": {\"cpu\": \"4vcpu\", \"memory\": \"16gb\", \"disk\": \"20GB\"} },\n", + " \"Executor\": { \"workerCount\": 4, \"workerConfiguration\": {\"cpu\": \"4vcpu\", \"memory\": \"16gb\", \"disk\": \"20GB\"} }\n", + " }' \\\n", + " --runtime-configuration '[{\n", + " \"classification\": \"spark-defaults\",\n", + " \"properties\": {\n", + " \"spark.jars\": \"s3://my-s3-bucket/jars/mysql-connector-j-8.3.0.jar\"\n", + " }\n", + " }]' \\\n", + " --monitoring-configuration '{\n", + " \"managedPersistenceMonitoringConfiguration\": {\"enabled\": true},\n", + " \"s3MonitoringConfiguration\": {\"logUri\": \"s3://my-s3-bucket/sparkconnect-logs/\"}\n", + " }' \\\n", + " --interactive-configuration '{\"sessionEnabled\": true}' \\\n", + " --region ${REGION}\n", + "```" + ] + }, + { + "cell_type": "markdown", + "id": "67c62bc2", + "metadata": {}, + "source": [ + "\n", + "### Start EMR Serverless Session \n", + "\n", + "`StartSession` creates a Spark driver and executors for this session on the application. With pre-initialized capacity (1 driver + 2 executors), the session is typically ready in under 90 seconds.\n", + "\n", + "```bash\n", + "aws emr-serverless start-session \\\n", + " --application-id ${APP_ID} \\\n", + " --execution-role-arn ${ROLE_ARN} \\\n", + " --name \"directS3JDBCwithoutcatalog\" \\\n", + " --idle-timeout-minutes 60 \\\n", + " --region ${REGION}\n", + "```\n", + "\n", + "```bash\n", + "export SESSION_ID=\n", + "```\n" + ] + }, + { + "cell_type": "markdown", + "id": "4f4418b8", + "metadata": {}, + "source": [ + "### Wait for Session IDLE State\n" + ] + }, + { + "cell_type": "markdown", + "id": "9ab108c9", + "metadata": {}, + "source": [ + "```bash\n", + "aws emr-serverless list-sessions \\\n", + " --application-id ${APP_ID} \\\n", + " --region ${REGION}\n", + "```\n", + "\n", + "Expected: `\"state\": \"IDLE\"`, `\"stateDetails\": \"Session is running.\"`" + ] + }, + { + "cell_type": "markdown", + "id": "22c54579", + "metadata": {}, + "source": [ + "## Connect a local `SparkSession` to the remote driver\n", + "\n", + "`GetSessionEndpoint` returns a host and a one-hour authentication token. The Spark Connect URL must include `:443`, `use_ssl=true`, and `x-aws-proxy-auth=`; the PySpark client will not connect without them.\n", + "\n", + "```bash\n", + "aws emr-serverless get-session-endpoint \\\n", + " --application-id ${APP_ID} \\\n", + " --session-id ${SESSION_ID} \\\n", + " --region ${REGION}\n", + "```\n", + "\n", + "```bash\n", + "export SPARK_CONNECT_ENDPOINT=\n", + "export AUTH_TOKEN=\n", + "```" + ] + }, + { + "cell_type": "markdown", + "id": "e09bc32c", + "metadata": {}, + "source": [ + "\n", + "## Connect from Local Jupyter via SparkConnect" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "db9d57a5", + "metadata": {}, + "outputs": [], + "source": [ + "from pyspark.sql import SparkSession\n", + "\n", + "# Stop any existing session first\n", + "try:\n", + " spark.stop()\n", + " print(\"✅ Old session stopped\")\n", + "except:\n", + " pass\n", + "\n", + "SPARK_CONNECT_ENDPOINT = \"\"\n", + "AUTH_TOKEN = \"\"\n", + "\n", + "url = (\n", + " f\"sc://{SPARK_CONNECT_ENDPOINT}:443/;\"\n", + " f\"use_ssl=true;\"\n", + " f\"x-aws-proxy-auth={AUTH_TOKEN}\"\n", + ")\n", + "\n", + "spark = (\n", + " SparkSession.builder\n", + " .remote(url)\n", + " .getOrCreate()\n", + ")\n", + "\n", + "print(f\"✅ Connected! Spark version: {spark.version}\")\n", + "spark.sql(\"SELECT 1 + 1 AS result\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "a0b1c2d3-e4f5-6789-abcd-ef0123456789", + "metadata": { + "tags": [] + }, + "source": [ + "**Define common variables used throughout the notebook.**\n", + "\n", + "Update the placeholder values below with your actual bucket name, RDS endpoint, and credentials." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c1d2e3f4-a5b6-7890-cdef-123456789abc", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "# S3 paths\n", + "S3_BUCKET = \"my-s3-bucket\"\n", + "PARQUET_PATH = f\"s3://{S3_BUCKET}/data/parquet/\"\n", + "CSV_PATH = f\"s3://{S3_BUCKET}/data/csv/\"\n", + "JSON_PATH = f\"s3://{S3_BUCKET}/data/json/\"\n", + "OUTPUT_PATH = f\"s3://{S3_BUCKET}/output/\"\n", + "\n", + "# JDBC connection details\n", + "JDBC_URL = \"jdbc:mysql://:3306/\"\n", + "JDBC_USER = \"\"\n", + "JDBC_PASSWORD = \"\"\n", + "JDBC_DRIVER = \"com.mysql.cj.jdbc.Driver\"" + ] + }, + { + "cell_type": "markdown", + "id": "8725d217-b26c-4fa8-84de-18969f5260c7", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Reading Files Directly from S3\n", + "\n", + "We read files directly from S3 paths using `spark.read` — **no catalog registration required**. Spark infers or uses the specified schema to load data as DataFrames." + ] + }, + { + "cell_type": "markdown", + "id": "58730a9c-431d-49b8-985f-ef2177f439a2", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "### Parquet Files\n", + "Parquet is a columnar storage format. Schema is embedded in the file, so Spark auto-infers it." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "7c898a47-33eb-4882-9990-e2dbaf257d68", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_parquet = spark.read.parquet(PARQUET_PATH)\n", + "\n", + "print(f\"Parquet record count: {df_parquet.count()}\")\n", + "df_parquet.printSchema()\n", + "df_parquet.show(5)" + ] + }, + { + "cell_type": "markdown", + "id": "bb85126b-117e-4ddd-b0b6-8652e6025e95", + "metadata": { + "tags": [] + }, + "source": [ + "**Create a temporary view so we can use SQL syntax — still no catalog involved.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4d1ac4ee-7b08-4842-9d53-99bb5b22cfa6", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_parquet.createOrReplaceTempView(\"orders\")\n", + "\n", + "spark.sql(\"SELECT * FROM orders LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "3190ca8c-803c-42f4-98b9-62c26be29959", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "### CSV Files\n", + "CSV files require specifying options like header and schema inference." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4054056c-fae1-44b9-ba89-3243ce0d8ea5", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_csv = spark.read \\\n", + " .option(\"header\", \"true\") \\\n", + " .option(\"inferSchema\", \"true\") \\\n", + " .csv(CSV_PATH)\n", + "\n", + "print(f\"CSV record count: {df_csv.count()}\")\n", + "df_csv.printSchema()\n", + "df_csv.show(5)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "653dc08f-b2be-4bec-afce-4b9a88f19712", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_csv.createOrReplaceTempView(\"customers\")\n", + "\n", + "spark.sql(\"SELECT * FROM customers LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "4504fd86-29c5-4c5f-a71d-f7e9643db3c8", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "### JSON Files\n", + "JSON files are semi-structured. Spark infers the schema automatically." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "cdf9839a-a352-4aca-9de7-59a784d4373a", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_json = spark.read.json(JSON_PATH)\n", + "\n", + "print(f\"JSON record count: {df_json.count()}\")\n", + "df_json.printSchema()\n", + "df_json.show(5)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "19bdb3bd-b7d3-458c-8fff-2e0dbc035eb7", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_json.createOrReplaceTempView(\"products\")\n", + "\n", + "spark.sql(\"SELECT * FROM products LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "85f8e445-d0a1-4bb1-9dfe-f1bf0ff946e1", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Reading from MySQL RDS via JDBC\n", + "\n", + "We connect directly to a MySQL RDS instance using JDBC. No catalog registration is needed — Spark reads the table directly from the database.\n", + "\n", + "Update the JDBC connection details defined in the Setup section above." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9b34f116-db73-4f80-92f4-10400a717066", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_mysql = spark.read \\\n", + " .format(\"jdbc\") \\\n", + " .option(\"url\", JDBC_URL) \\\n", + " .option(\"dbtable\", \"inventory\") \\\n", + " .option(\"user\", JDBC_USER) \\\n", + " .option(\"password\", JDBC_PASSWORD) \\\n", + " .option(\"driver\", JDBC_DRIVER) \\\n", + " .load()\n", + "\n", + "print(f\"MySQL record count: {df_mysql.count()}\")\n", + "df_mysql.printSchema()\n", + "df_mysql.show(5)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "fb7863b4-a453-490d-a638-7cb29368a76f", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_mysql.createOrReplaceTempView(\"inventory\")\n", + "\n", + "spark.sql(\"SELECT * FROM inventory LIMIT 10\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "34824f0a-4334-45d3-8a0d-9a575875488a", + "metadata": { + "tags": [] + }, + "source": [ + "**You can also push down a query to the database to reduce data transfer.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "91cd1eac-c422-4c44-bdf7-d23284bc87d6", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "query = \"(SELECT product_id, quantity, warehouse_location FROM inventory WHERE quantity > 100) AS filtered_inventory\"\n", + "\n", + "df_filtered = spark.read \\\n", + " .format(\"jdbc\") \\\n", + " .option(\"url\", JDBC_URL) \\\n", + " .option(\"dbtable\", query) \\\n", + " .option(\"user\", JDBC_USER) \\\n", + " .option(\"password\", JDBC_PASSWORD) \\\n", + " .option(\"driver\", JDBC_DRIVER) \\\n", + " .load()\n", + "\n", + "df_filtered.show(5)" + ] + }, + { + "cell_type": "markdown", + "id": "9d19f6c3-9f08-402f-85ac-8aa487f51c05", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Cross-Source Joins\n", + "\n", + "Now we join data across all sources — S3 Parquet, S3 CSV, S3 JSON, and MySQL RDS — in a single Spark SQL query. This is the power of the catalog-free approach: temporary views from any source can be joined seamlessly." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d338d82a-f7b0-4ef7-8e1b-ebb2c88c9081", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "cross_join_df = spark.sql(\"\"\"\n", + " SELECT \n", + " c.customer_id,\n", + " c.customer_name,\n", + " o.order_id,\n", + " o.order_date,\n", + " p.product_name,\n", + " p.category,\n", + " o.amount,\n", + " i.quantity AS inventory_quantity,\n", + " i.warehouse_location\n", + " FROM orders o\n", + " JOIN customers c ON o.customer_id = c.customer_id\n", + " JOIN products p ON o.product_id = p.product_id\n", + " JOIN inventory i ON o.product_id = i.product_id\n", + "\"\"\")\n", + "\n", + "print(f\"Cross-source join result count: {cross_join_df.count()}\")\n", + "cross_join_df.show(10)" + ] + }, + { + "cell_type": "markdown", + "id": "fb22bf8b-3c83-45b1-aac6-c37e61b41e20", + "metadata": { + "tags": [] + }, + "source": [ + "**Run aggregations on the joined data.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "906629e3-5c0d-4ab6-a2dc-5d8a205c2288", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "spark.sql(\"\"\"\n", + " SELECT \n", + " p.category,\n", + " COUNT(o.order_id) AS total_orders,\n", + " ROUND(SUM(o.amount), 2) AS total_revenue,\n", + " ROUND(AVG(o.amount), 2) AS avg_order_value\n", + " FROM orders o\n", + " JOIN products p ON o.product_id = p.product_id\n", + " GROUP BY p.category\n", + " ORDER BY total_revenue DESC\n", + "\"\"\").show()" + ] + }, + { + "cell_type": "markdown", + "id": "1d035fa4-6fc5-4c1c-b846-3c386ca0222b", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Writing Results Back to S3\n", + "\n", + "Write the cross-source join results back to S3 in Parquet format — again, no catalog involved. The output is simply a set of files in an S3 path." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3073f8a6-121f-4768-b112-7ebae68d28f4", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "cross_join_df.write \\\n", + " .mode(\"overwrite\") \\\n", + " .parquet(f\"{OUTPUT_PATH}cross_source_join/\")\n", + "\n", + "print(f\"Results written to {OUTPUT_PATH}cross_source_join/\")" + ] + }, + { + "cell_type": "markdown", + "id": "d4f8bc24-5ad0-4b1b-8e94-34d65b6d62c3", + "metadata": { + "tags": [] + }, + "source": [ + "**Verify the written data by reading it back.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3dce569e-93c4-43c3-a962-c72fcc83f8ef", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "df_verify = spark.read.parquet(f\"{OUTPUT_PATH}cross_source_join/\")\n", + "\n", + "print(f\"Verified record count: {df_verify.count()}\")\n", + "df_verify.show(5)" + ] + }, + { + "cell_type": "markdown", + "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", + "metadata": { + "tags": [] + }, + "source": [ + "**You can also write in other formats like CSV or JSON.**" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b1c2d3e4-f5a6-7890-bcde-f12345678901", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "# Write as CSV\n", + "cross_join_df.write \\\n", + " .mode(\"overwrite\") \\\n", + " .option(\"header\", \"true\") \\\n", + " .csv(f\"{OUTPUT_PATH}cross_source_join_csv/\")\n", + "\n", + "print(f\"CSV results written to {OUTPUT_PATH}cross_source_join_csv/\")" + ] + }, + { + "cell_type": "markdown", + "id": "9e977e5f", + "metadata": {}, + "source": [ + "## Spark UI\n", + "`GetResourceDashboard` returns a URL to the Spark UI for this session. While the session is running, it shows the live UI (jobs, stages, executors, SQL queries). After termination, the same URL serves the Spark History Server." + ] + }, + { + "cell_type": "markdown", + "id": "b1ea3632", + "metadata": {}, + "source": [ + "### Get Spark UI / Resource Dashboard\n", + "\n", + "Two-step process required:\n", + "\n", + "**Step 1 — Run in Terminal to Get URL:**\n", + "```bash\n", + "aws emr-serverless get-resource-dashboard \\\n", + " —application-id ${APP_ID} \\\n", + " —resource-id ${SESSION_ID} \\\n", + " —resource-type SESSION \\\n", + " —endpoint-url ${EP} \\\n", + " —region ${REGION}\n", + "```\n", + "\n", + "Copy the `url` value from the response.\n", + "\n", + "**Step 2 — Paste URL in Jupyter and Open:**\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "54d7372d", + "metadata": {}, + "outputs": [], + "source": [ + "import webbrowser\n", + "\n", + "# Paste URL from terminal output here\n", + "url = \"\"\n", + "\n", + "print(\"✅ Opening Spark UI in browser...\")\n", + "print(f\"\\n🔗 Session : \")\n", + "print(f\"🔗 App ID : \")\n", + "print(f\"🔗 Region : \")\n", + "webbrowser.open(url)\n" + ] + }, + { + "cell_type": "markdown", + "id": "c2d3e4f5-a6b7-8901-cdef-234567890abc", + "metadata": { + "tags": [] + }, + "source": [ + "\n", + "## Cleanup\n", + "\n", + "Stop the Spark session to release EMR Serverless resources." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d3e4f5a6-b7c8-9012-defg-345678901bcd", + "metadata": { + "tags": [] + }, + "outputs": [], + "source": [ + "# Close local client, terminate remote session to stop billing\n", + "spark.stop()\n", + "print(\"SparkConnect session stopped. EMR Serverless resources released.\")" + ] + }, + { + "cell_type": "markdown", + "id": "e48f8ac5", + "metadata": {}, + "source": [ + "### From the Terminal — Terminate Session and Stop Application\n", + "\n", + "```bash\n", + "# Terminate session\n", + "aws emr-serverless terminate-session \\\n", + " --application-id ${APP_ID} \\\n", + " --session-id ${SESSION_ID} \\\n", + " --region ${REGION}\n", + "\n", + "# Stop application\n", + "aws emr-serverless stop-application \\\n", + " --application-id ${APP_ID} \\\n", + " --region ${REGION}\n", + "\n", + "# Delete RDS instance\n", + "aws rds delete-db-instance \\\n", + " --db-instance-identifier sparkconnect-demo-mysql \\\n", + " --skip-final-snapshot \\\n", + " --region ${REGION}\n", + "\n", + "# Delete S3 sample data (keep bucket if needed)\n", + "aws s3 rm s3://$BUCKET/data/ --recursive\n", + "aws s3 rm s3://$BUCKET/jars/ --recursive\n", + "aws s3 rm s3://$BUCKET/sparkconnect-logs/ --recursive\n", + "```" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "pygments_lexer": "ipython3", + "version": "3.9.0" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +}