writing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 0,
   "id": "4fc0a20b",
   "metadata": {},
   "outputs": [],
   "source": [
    "#import libraries\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import DoubleType\n",
    "from pyspark.sql.functions import col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cef1064f",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/06/04 17:55:57 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "23/06/04 17:55:57 WARN DependencyUtils: Local jar /Users/pranibansal/Downloads/spark-3.4.0-bin-hadoop3/redshift-jdbc42-2.0.0.4.jar does not exist, skipping.\n",
      "23/06/04 17:55:58 INFO SparkContext: Running Spark version 3.4.0\n",
      "23/06/04 17:55:58 INFO ResourceUtils: ==============================================================\n",
      "23/06/04 17:55:58 INFO ResourceUtils: No custom resources configured for spark.driver.\n",
      "23/06/04 17:55:58 INFO ResourceUtils: ==============================================================\n",
      "23/06/04 17:55:58 INFO SparkContext: Submitted application: integration_pyspark_redshift\n",
      "23/06/04 17:55:58 INFO ResourceProfile: Default ResourceProfile created, executor resources: Map(cores -> name: cores, amount: 1, script: , vendor: , memory -> name: memory, amount: 1024, script: , vendor: , offHeap -> name: offHeap, amount: 0, script: , vendor: ), task resources: Map(cpus -> name: cpus, amount: 1.0)\n",
      "23/06/04 17:55:58 INFO ResourceProfile: Limiting resource is cpu\n",
      "23/06/04 17:55:58 INFO ResourceProfileManager: Added ResourceProfile id: 0\n",
      "23/06/04 17:55:58 INFO SecurityManager: Changing view acls to: pranibansal\n",
      "23/06/04 17:55:58 INFO SecurityManager: Changing modify acls to: pranibansal\n",
      "23/06/04 17:55:58 INFO SecurityManager: Changing view acls groups to: \n",
      "23/06/04 17:55:58 INFO SecurityManager: Changing modify acls groups to: \n",
      "23/06/04 17:55:58 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: pranibansal; groups with view permissions: EMPTY; users with modify permissions: pranibansal; groups with modify permissions: EMPTY\n",
      "23/06/04 17:55:58 INFO Utils: Successfully started service 'sparkDriver' on port 56012.\n",
      "23/06/04 17:55:58 INFO SparkEnv: Registering MapOutputTracker\n",
      "23/06/04 17:55:58 INFO SparkEnv: Registering BlockManagerMaster\n",
      "23/06/04 17:55:58 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information\n",
      "23/06/04 17:55:58 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up\n",
      "23/06/04 17:55:58 INFO SparkEnv: Registering BlockManagerMasterHeartbeat\n",
      "23/06/04 17:55:58 INFO DiskBlockManager: Created local directory at /private/var/folders/56/plm6ts814hj714s5vcdw5h040000gn/T/blockmgr-f05601af-5e88-4bcd-b835-c961f32a2fa7\n",
      "23/06/04 17:55:58 INFO MemoryStore: MemoryStore started with capacity 434.4 MiB\n",
      "23/06/04 17:55:58 INFO SparkEnv: Registering OutputCommitCoordinator\n",
      "23/06/04 17:55:58 INFO JettyUtils: Start Jetty 0.0.0.0:4040 for SparkUI\n",
      "23/06/04 17:55:58 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n",
      "23/06/04 17:55:58 WARN Utils: Service 'SparkUI' could not bind on port 4041. Attempting port 4042.\n",
      "23/06/04 17:55:58 INFO Utils: Successfully started service 'SparkUI' on port 4042.\n",
      "23/06/04 17:55:58 ERROR SparkContext: Failed to add redshift-jdbc42-2.0.0.4.jar to Spark environment\n",
      "java.io.FileNotFoundException: Jar /Users/pranibansal/Downloads/spark-3.4.0-bin-hadoop3/redshift-jdbc42-2.0.0.4.jar not found\n",
      "\tat org.apache.spark.SparkContext.addLocalJarFile$1(SparkContext.scala:1968)\n",
      "\tat org.apache.spark.SparkContext.addJar(SparkContext.scala:2023)\n",
      "\tat org.apache.spark.SparkContext.$anonfun$new$12(SparkContext.scala:507)\n",
      "\tat org.apache.spark.SparkContext.$anonfun$new$12$adapted(SparkContext.scala:507)\n",
      "\tat scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)\n",
      "\tat scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)\n",
      "\tat org.apache.spark.SparkContext.<init>(SparkContext.scala:507)\n",
      "\tat org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58)\n",
      "\tat java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)\n",
      "\tat java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:76)\n",
      "\tat java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:53)\n",
      "\tat java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499)\n",
      "\tat java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:483)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:238)\n",
      "\tat py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)\n",
      "\tat py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)\n",
      "\tat py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)\n",
      "\tat py4j.ClientServerConnection.run(ClientServerConnection.java:106)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:833)\n",
      "23/06/04 17:55:59 INFO Executor: Starting executor ID driver on host pranis-air.lan\n",
      "23/06/04 17:55:59 INFO Executor: Starting executor with user classpath (userClassPathFirst = false): ''\n",
      "23/06/04 17:55:59 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 56016.\n",
      "23/06/04 17:55:59 INFO NettyBlockTransferService: Server created on pranis-air.lan:56016\n",
      "23/06/04 17:55:59 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy\n",
      "23/06/04 17:55:59 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, pranis-air.lan, 56016, None)\n",
      "23/06/04 17:55:59 INFO BlockManagerMasterEndpoint: Registering block manager pranis-air.lan:56016 with 434.4 MiB RAM, BlockManagerId(driver, pranis-air.lan, 56016, None)\n",
      "23/06/04 17:55:59 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, pranis-air.lan, 56016, None)\n",
      "23/06/04 17:55:59 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, pranis-air.lan, 56016, None)\n"
     ]
    }
   ],
   "source": [
    "jar_path = 'redshift-jdbc42-2.0.0.4.jar'\n",
    "# w/ spark session obj.\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"integration_pyspark_redshift\") \\\n",
    "    .config(\"spark.jars\", jar_path)\\\n",
    "    .config('spark.driver.extraClassPath', jar_path) \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ebec24a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "#provide aws credentials\n",
    "spark._jsc.hadoopConfiguration().set(\"fs.s3.awsAccessKeyId\", \"yourawsAccessKeyId\")\n",
    "spark._jsc.hadoopConfiguration().set(\"fs.s3.awsSecretAccessKey\", \"yourawsSecretAccessKey\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "776fc857",
   "metadata": {},
   "outputs": [],
   "source": [
    "# define variables to store the Redshift URL, Redshift username, Redshift password, and our S3 temporary directory\n",
    "redshift_url = \"jdbc:redshift://redshift-cluster-1.cem25ev4t3ap.us-east-1.redshift.amazonaws.com:5439/dev\"\n",
    "redshift_user = \"datastunt\"\n",
    "redshift_password = \"Datastunt123\"\n",
    "s3_temp_dir = \"s3://airlines123/airline/\"\n",
    "\n",
    "df_filtered.write \\\n",
    "    .format(\"jdbc\") \\\n",
    "    .option(\"url\", redshift_url) \\\n",
    "    .option(\"dbtable\", \"airline\") \\\n",
    "    .option(\"user\", redshift_user) \\\n",
    "    .option(\"password\", redshift_password) \\\n",
    "    .option(\"tempdir\", s3_temp_dir) \\\n",
    "    .mode(\"error\") \\\n",
    "    .save()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.9.7 ('base')",
   "language": "python",
   "name": "python397jvsc74a57bd04090bb9dea958443c796ed706d0ebc3b46bb5181ea941ef1d87768496669018a"
  },
  "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": 5
}