notebooks/load-postgres-notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: An illegal reflective access operation has occurred\n",
      "WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/usr/local/spark-3.1.2-bin-hadoop3.2/jars/spark-unsafe_2.12-3.1.2.jar) to constructor java.nio.DirectByteBuffer(long,int)\n",
      "WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform\n",
      "WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations\n",
      "WARNING: All illegal access operations will be denied in a future release\n",
      "21/10/31 05:58:31 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Spark session & context\n",
    "spark = (SparkSession\n",
    "         .builder\n",
    "         .master(\"local\")\n",
    "         .appName(\"load-postgres\")\n",
    "         # Add postgres jar\n",
    "         .config(\"spark.driver.extraClassPath\", \"/home/jovyan/work/jars/postgresql-9.4.1207.jar\")\n",
    "         .getOrCreate())\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Read CSV data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_movies_csv = (\n",
    "    spark.read\n",
    "    .format(\"csv\")\n",
    "    .option(\"header\", True)\n",
    "    .load(\"/home/jovyan/work/data/movies.csv\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movieId: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- genres: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_movies_csv.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_ratings_csv = (\n",
    "    spark.read\n",
    "    .format(\"csv\")\n",
    "    .option(\"header\", True)\n",
    "    .load(\"/home/jovyan/work/data/ratings.csv\")\n",
    "    .withColumnRenamed(\"timestamp\",\"timestamp_epoch\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert epoch to timestamp and rating to DoubleType\n",
    "from pyspark.sql.functions import from_unixtime, col, to_timestamp\n",
    "from pyspark.sql.types import DoubleType\n",
    "\n",
    "df_ratings_csv_fmt = (\n",
    "    df_ratings_csv\n",
    "    .withColumn('rating', col(\"rating\").cast(DoubleType()))\n",
    "    .withColumn('timestamp', to_timestamp(from_unixtime(col(\"timestamp_epoch\"))))\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load data to Postgres"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "(df_movies_csv.write\n",
    " .format(\"jdbc\")\n",
    " .option(\"url\", \"jdbc:postgresql://postgres/test\")\n",
    " .option(\"dbtable\", \"public.movies\")\n",
    " .option(\"user\", \"test\")\n",
    " .option(\"password\", \"postgres\")\n",
    " .mode(\"overwrite\")\n",
    " .save())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "(df_ratings_csv_fmt\n",
    " .select([c for c in df_ratings_csv_fmt.columns if c != \"timestamp_epoch\"])\n",
    " .write\n",
    " .format(\"jdbc\")\n",
    " .option(\"url\", \"jdbc:postgresql://postgres/test\")\n",
    " .option(\"dbtable\", \"public.ratings\")\n",
    " .option(\"user\", \"test\")\n",
    " .option(\"password\", \"postgres\")\n",
    " .mode(\"overwrite\")\n",
    " .save())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------------+-------------------+\n",
      "|userId|movieId|rating|timestamp_epoch|          timestamp|\n",
      "+------+-------+------+---------------+-------------------+\n",
      "|     1|      1|   4.0|      964982703|2000-07-30 18:45:03|\n",
      "|     1|      3|   4.0|      964981247|2000-07-30 18:20:47|\n",
      "|     1|      6|   4.0|      964982224|2000-07-30 18:37:04|\n",
      "|     1|     47|   5.0|      964983815|2000-07-30 19:03:35|\n",
      "|     1|     50|   5.0|      964982931|2000-07-30 18:48:51|\n",
      "|     1|     70|   3.0|      964982400|2000-07-30 18:40:00|\n",
      "|     1|    101|   5.0|      964980868|2000-07-30 18:14:28|\n",
      "|     1|    110|   4.0|      964982176|2000-07-30 18:36:16|\n",
      "|     1|    151|   5.0|      964984041|2000-07-30 19:07:21|\n",
      "|     1|    157|   5.0|      964984100|2000-07-30 19:08:20|\n",
      "|     1|    163|   5.0|      964983650|2000-07-30 19:00:50|\n",
      "|     1|    216|   5.0|      964981208|2000-07-30 18:20:08|\n",
      "|     1|    223|   3.0|      964980985|2000-07-30 18:16:25|\n",
      "|     1|    231|   5.0|      964981179|2000-07-30 18:19:39|\n",
      "|     1|    235|   4.0|      964980908|2000-07-30 18:15:08|\n",
      "|     1|    260|   5.0|      964981680|2000-07-30 18:28:00|\n",
      "|     1|    296|   3.0|      964982967|2000-07-30 18:49:27|\n",
      "|     1|    316|   3.0|      964982310|2000-07-30 18:38:30|\n",
      "|     1|    333|   5.0|      964981179|2000-07-30 18:19:39|\n",
      "|     1|    349|   4.0|      964982563|2000-07-30 18:42:43|\n",
      "+------+-------+------+---------------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_ratings_csv_fmt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[userId: string, movieId: string, rating: double, timestamp_epoch: string, timestamp: timestamp]"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}