spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using iguazio DB and File System with Spark "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark SQL basic example\") \\\n",
    "    .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read a CSV file from iguazio file system into Spark dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+--------+----+-----+----+----+----+----+----+----+----+----+----+----+\n",
      "|              _c0|     _c1| _c2|  _c3| _c4| _c5| _c6| _c7| _c8| _c9|_c10|_c11|_c12|_c13|\n",
      "+-----------------+--------+----+-----+----+----+----+----+----+----+----+----+----+----+\n",
      "|          STATION|    DATE|PRCP| SNWD|SNOW|TMAX|TMIN|AWND|WDF2|WDF5|WSF2|WSF5|PGTM|FMTM|\n",
      "|GHCND:USW00094728|20000101|   0|-9999|   0| 100|  11|  26| 250| 230|  72|  94|1337|1337|\n",
      "|GHCND:USW00094728|20000102|   0|-9999|   0| 156|  61|  21| 260| 260|  72| 112|2313|2314|\n",
      "|GHCND:USW00094728|20000103|   0|-9999|   0| 178| 106|  30| 260| 250|  67|  94| 320| 321|\n",
      "|GHCND:USW00094728|20000104| 178|-9999|   0| 156|  78|  35| 320| 350|  67| 107|1819|1840|\n",
      "|GHCND:USW00094728|20000105|   0|-9999|   0|  83| -17|  51| 330| 340| 107| 143| 843| 844|\n",
      "|GHCND:USW00094728|20000106|   0|-9999|   0|  56| -22|  30| 220| 250|  67|  98|1833|1834|\n",
      "|GHCND:USW00094728|20000107|   0|-9999|   0|  94|  17|  42| 300| 310| 103| 156|1521|1601|\n",
      "|GHCND:USW00094728|20000109|   5|-9999|   0| 106|  28|  26| 270| 270|  63|  89|  22| 601|\n",
      "|GHCND:USW00094728|20000110| 213|-9999|   0| 144|  67|  41| 280| 260|  94| 139|1736|1758|\n",
      "|GHCND:USW00094728|20000111|   0|-9999|   0| 111|  44|  49| 300| 310| 112| 174|1203|1203|\n",
      "|GHCND:USW00094728|20000112|   0|-9999|   0|  83|  39|  39| 330| 330|  94| 161| 536| 610|\n",
      "|GHCND:USW00094728|20000113|  13|-9999|   0|  39| -78|  51|  90|  10| 103| 143|1539| 843|\n",
      "|GHCND:USW00094728|20000114|   0|    0|   0| -56|-106|  72| 330| 340| 125| 165|1422|1350|\n",
      "|GHCND:USW00094728|20000115|   0|-9999|   0| -11| -83|  31| 320| 330|  80| 130| 241| 201|\n",
      "|GHCND:USW00094728|20000116|   0|-9999|   0|  78| -61|  52| 320| 300| 125| 170|2319|2219|\n",
      "|GHCND:USW00094728|20000117|   0|-9999|   0| -61|-139|  63| 330| 310| 134| 201| 250| 253|\n",
      "|GHCND:USW00094728|20000118|   0|-9999|   0| -72|-161|  33|  10|  10|  72| 107| 543| 540|\n",
      "|GHCND:USW00094728|20000119|   0|-9999|   0| -17|-122|  29| 330| 320|  63|  80|1607|1544|\n",
      "|GHCND:USW00094728|20000120|  30|-9999|  64| -22| -50|  41|  70|  70|  89| 116|1302|1744|\n",
      "+-----------------+--------+----+-----+----+----+----+----+----+----+----+----+----+----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.read.option(\"inferSchema\", \"true\").csv('v3io://bigdata/weather.csv')\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Write the dataframe into iguazio Key/Value DB API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "df.write.format(\"io.iguaz.v3io.spark.sql.kv\").mode(\"append\").option(\"key\", \"_c0\").save(\"v3io://bigdata/weather_table\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read the data from the DB with predicate offload (push-down)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+--------+----+-----+----+----+----+----+----+----+----+----+----+----+\n",
      "|              _c0|     _c1| _c2|  _c3| _c4| _c5| _c6| _c7| _c8| _c9|_c10|_c11|_c12|_c13|\n",
      "+-----------------+--------+----+-----+----+----+----+----+----+----+----+----+----+----+\n",
      "|GHCND:USW00094728|20000101|   0|-9999|   0| 100|  11|  26| 250| 230|  72|  94|1337|1337|\n",
      "|          STATION|    DATE|PRCP| SNWD|SNOW|TMAX|TMIN|AWND|WDF2|WDF5|WSF2|WSF5|PGTM|FMTM|\n",
      "+-----------------+--------+----+-----+----+----+----+----+----+----+----+----+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"io.iguaz.v3io.spark.sql.kv\").load(\"v3io://bigdata/weather_table\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}