notebooks/spark_basics/DataFrame.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DataFrame"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark SQL provides a special type of collection of data called [*DataFrame*](https://spark.apache.org/docs/latest/sql-programming-guide.html#datasets-and-dataframes) which is organized into named columns, built on top of RDD. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations. DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs. In the following example, we will show that how to create a Dateframe for a JSON dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----+\n",
      "|             address|name|\n",
      "+--------------------+----+\n",
      "|[Los Angeles,Cali...| Bob|\n",
      "+--------------------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "from bigdl.util.common import *\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "sc = SparkContext('local')\n",
    "spark= SparkSession(sc)\n",
    "\n",
    "# Defines a Python list storing one JSON object.\n",
    "json_strings = ['{\"name\":\"Bob\",\"address\":{\"city\":\"Los Angeles\",\"state\":\"California\"}}', ]\n",
    "# Defines an RDD from the Python list.\n",
    "peopleRDD = sc.parallelize(json_strings)\n",
    "# Creates an DataFrame from an RDD[String].\n",
    "people = spark.read.json(peopleRDD)\n",
    "people.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Text Search"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this example we show how to search through the log file for the number of error messages using *Dataframe*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import Row\n",
    "\n",
    "text_data = sc.parallelize([\"MYSQL ERROR 1\\n\",\"MYSQL ERROR 2\\n\",\"MYSQL\\n\"])\n",
    "# Creates a DataFrame having a single column named \"line\"\n",
    "df = text_data.map(lambda r: Row(r)).toDF([\"line\"])\n",
    "# Counts ERRORs\n",
    "errors = df.filter(df[\"line\"].like(\"%ERROR%\"))\n",
    "# Counts all the errors\n",
    "errors.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To find out more DataFrame APIs, you can refer to this website [DataFrames APIs](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2.0
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}