sample_jupyter_spark_nb.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02886506",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "73e5dd7c",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "# Create a SparkSession\n",
    "spark = (\n",
    "        SparkSession.builder.appName(\"Naive Data Pipeline\")\n",
    "        .enableHiveSupport()\n",
    "        .getOrCreate()\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "40875b97",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "def get_upstream_table(table_name: str, spark: SparkSession):\n",
    "    host = os.getenv(\"UPSTREAM_HOST\", \"upstream\")\n",
    "    port = os.getenv(\"UPSTREAM_PORT\", \"5432\")\n",
    "    db = os.getenv(\"UPSTREAM_DATABASE\", \"upstreamdb\")\n",
    "    jdbc_url = f'jdbc:postgresql://{host}:{port}/{db}'\n",
    "    connection_properties = {\n",
    "        \"user\": os.getenv(\"UPSTREAM_USERNAME\", \"sdeuser\"),\n",
    "        \"password\": os.getenv(\"UPSTREAM_PASSWORD\", \"sdepassword\"),\n",
    "        \"driver\": \"org.postgresql.Driver\",\n",
    "    }\n",
    "    return spark.read.jdbc(\n",
    "        url=jdbc_url, table=table_name, properties=connection_properties\n",
    "    )\n",
    "\n",
    "\n",
    "def get_upstream_seller_data(spark):\n",
    "    table_name = \"rainforest.seller\"\n",
    "    return get_upstream_table(table_name,  spark)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b8ea349",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "raw_seller_data = get_upstream_seller_data(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8258b1bf",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "raw_seller_data.show()\n",
    "# note that there will be no data unless you populate the upstream tables with command:\n",
    "# docker exec spark-master bash -c \"python3 /opt/spark/work-dir/capstone/upstream_datagen/datagen.py\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "867be0b5",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}