Montagsrunde_basics.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "toc": "true"
   },
   "source": [
    "# Learning Spark Basics - Demos\n",
    "### Yu Feng\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Environment Variables Setting"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "\n",
    "def configure_spark(spark_home=None, pyspark_python=None):\n",
    "    spark_home = spark_home or \"/usr/local/spark\"\n",
    "    os.environ['SPARK_HOME'] = spark_home\n",
    "    \n",
    "    # Add the PySpark directories to the Python path:\n",
    "    sys.path.insert(1, os.path.join(spark_home, 'python'))\n",
    "    sys.path.insert(1, os.path.join(spark_home, 'python', 'build'))\n",
    "    sys.path.insert(1, os.path.join(spark_home, 'python', 'lib','pyspark.zip'))\n",
    "    sys.path.insert(1, os.path.join(spark_home, 'python', 'lib','py4j-0.10.1-src.zip'))\n",
    "    \n",
    "    os.environ['PYTHONPATH'] = os.path.join(spark_home, 'python', 'lib','py4j-0.10.1-src.zip')\n",
    "    \n",
    "    # If PySpark isn't specified, use currently running Python binary:\n",
    "    pyspark_python = pyspark_python or sys.executable\n",
    "    os.environ['PYSPARK_PYTHON'] = pyspark_python\n",
    "\n",
    "configure_spark()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "os.environ['JAVA_HOME'] = \"/usr/lib/jvm/java-1.7.0-openjdk-amd64\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Demo 1: Programming with RDDs\n",
    "\n",
    "The most traditional tasks for Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext\n",
    "\n",
    "conf = SparkConf().setMaster(\"local\").setAppName(\"My App\") \n",
    "sc = SparkContext(conf = conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RDD (resilient distributed dataset)\n",
    "* Spilt into multiple partitions\n",
    "* Be computed on different nodes of the cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Create an RDD from external dataset\n",
    "lines = sc.textFile(\"alice30.txt\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Create an RDD from an existing colleciton\n",
    "localLines = sc.parallelize([\"pandas\", \"i like pandas\"]) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RRD Operations: Actions and Transformation\n",
    "\n",
    "### - Actions: compute results based on RDD, return it to driver program or save it to external storage system (e.g. HDFS)\n",
    "\n",
    "* .count() - Count the number of items in this RDD\n",
    "* .first() - First item in this RDD\n",
    "* .take() - Return first n lines \n",
    "* .collect() - Return all lines, should not used in large datasets\n",
    "* .reduce()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3599\n",
      "                ALICE'S ADVENTURES IN WONDERLAND\n"
     ]
    }
   ],
   "source": [
    "print lines.count()\n",
    "print lines.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "                ALICE'S ADVENTURES IN WONDERLAND\n",
      "\n",
      "                          Lewis Carroll\n",
      "\n",
      "               THE MILLENNIUM FULCRUM EDITION 3.0\n",
      "\n",
      "\n",
      "\n",
      "\n",
      "                            CHAPTER I\n"
     ]
    }
   ],
   "source": [
    "for line in lines.take(10): \n",
    "    print line"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n"
     ]
    }
   ],
   "source": [
    "nums = sc.parallelize([1, 2, 3, 4]) \n",
    "sum = nums.reduce(lambda x, y: x + y)\n",
    "print sum "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### - Transformation: construct a new RDD from a previous one\n",
    "\n",
    "* .filter() - takes a funtion and return the elements fulfill the requirements\n",
    "* .map() - takes a function and return new values for each elements\n",
    "* .distinct(), .union(), .intersection(), .substract() - mathematical sets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Image1](images/mathsetop.PNG)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "74\n"
     ]
    }
   ],
   "source": [
    "# Return an entire new RDD\n",
    "pythonLines = lines.filter(lambda line: \"Queen\" in line)\n",
    "\n",
    "#for line in pythonLines.collect():\n",
    "#    print line\n",
    "\n",
    "print pythonLines.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "136\n"
     ]
    }
   ],
   "source": [
    "kingLines = lines.filter(lambda line: \"King\" in line)\n",
    "king_queen_lines = pythonLines.union(kingLines)\n",
    "\n",
    "print king_queen_lines.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "62\n"
     ]
    }
   ],
   "source": [
    "def containsKing(s):\n",
    "    return \"King\" in s\n",
    "\n",
    "kingLines = lines.filter(containsKing)\n",
    "print kingLines.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 4, 9, 16]\n"
     ]
    }
   ],
   "source": [
    "nums = sc.parallelize([1, 2, 3, 4]) \n",
    "squared = nums.map(lambda x: x * x).collect() \n",
    "print squared"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.stop() \n",
    "del sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Demo 2: Programming with DataFrame\n",
    "\n",
    "Dataframe refers to \"tabular\" data\n",
    "\n",
    "\n",
    "\n",
    "## DataFrame in Spark\n",
    "DataFrame performs much faster than RDDs\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "![Image2](images/dataframes-faster.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Version 2.0.0, use SparkSession to create DataFrame, register DataFrame \n",
    "# as tables, execute SQL over tables, cache tables, and read parquet files. \n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark SQL basic example\") \\\n",
    "    .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total: 3118\n",
      "+---------+-----------+--------------------+-----+--------------------+\n",
      "| Latitude|  Longitude|                Time|Label|            Location|\n",
      "+---------+-----------+--------------------+-----+--------------------+\n",
      "|55.864237|  -4.251806|Mon Oct 05 08:20:...|    1|   Glasgow  Scotland|\n",
      "|33.748995| -84.387982|Mon Oct 05 08:20:...|    1|         Atlanta  GA|\n",
      "|33.448377|-112.074037|Mon Oct 05 08:20:...|    1|     Phoenix Arizona|\n",
      "|34.000710| -81.034814|Mon Oct 05 08:20:...|    1|Columbia  South C...|\n",
      "|40.569789| -79.764770|Mon Oct 05 08:20:...|    1|  New Kensington  PA|\n",
      "|33.748995| -84.387982|Mon Oct 05 08:20:...|    1|         Atlanta  GA|\n",
      "|35.689197|  51.388974|Mon Oct 05 08:20:...|    1|         IRAN TEHRAN|\n",
      "|35.125801|-117.985904|Mon Oct 05 08:20:...|    1|  California City CA|\n",
      "|51.621440|  -3.943646|Mon Oct 05 08:20:...|    1|             swansea|\n",
      "|33.748995| -84.387982|Mon Oct 05 08:20:...|    1|   Atlanta  GA      |\n",
      "+---------+-----------+--------------------+-----+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# spark is an existing SparkSession\n",
    "df = spark.read.csv(\"work-flow/tweet.csv\", header=True)\n",
    "# Displays the content of the DataFrame to stdout\n",
    "print 'Total: %d' % df.count()\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Latitude: string (nullable = true)\n",
      " |-- Longitude: string (nullable = true)\n",
      " |-- Time: string (nullable = true)\n",
      " |-- Label: string (nullable = true)\n",
      " |-- Location: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# spark, df are from the previous example\n",
    "# Print the schema in a tree format\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|            Location|\n",
      "+--------------------+\n",
      "|   Glasgow  Scotland|\n",
      "|         Atlanta  GA|\n",
      "|     Phoenix Arizona|\n",
      "|Columbia  South C...|\n",
      "|  New Kensington  PA|\n",
      "|         Atlanta  GA|\n",
      "|         IRAN TEHRAN|\n",
      "|  California City CA|\n",
      "|             swansea|\n",
      "|   Atlanta  GA      |\n",
      "|Vila   Velha   ES...|\n",
      "|             USA  ma|\n",
      "|         IRAN TEHRAN|\n",
      "|         Phoenix  AZ|\n",
      "|              Lagos |\n",
      "| Cavite  Philippines|\n",
      "|      Barrington  RI|\n",
      "|               Miami|\n",
      "|                U.K.|\n",
      "|      Patiala Punjab|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Select only the \"Location\" column\n",
    "df.select('Location').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|            Location|count|\n",
      "+--------------------+-----+\n",
      "|              London|   86|\n",
      "|        Columbia  SC|   34|\n",
      "|Querétaro Arteaga...|   26|\n",
      "|Florence  South C...|   25|\n",
      "|         Atlanta  GA|   24|\n",
      "|    Myrtle Beach  SC|   24|\n",
      "|        New York  NY|   24|\n",
      "|       Charlotte  NC|   22|\n",
      "|      Wilmington  NC|   22|\n",
      "|         Norfolk  VA|   20|\n",
      "|      Charleston  SC|   18|\n",
      "|      South Carolina|   18|\n",
      "|          Birmingham|   17|\n",
      "|               World|   17|\n",
      "|     London  England|   17|\n",
      "|         Houston  TX|   16|\n",
      "|        Greenbelt MD|   15|\n",
      "| Manchester  England|   15|\n",
      "|               Earth|   15|\n",
      "|      Lagos  Nigeria|   14|\n",
      "+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Count people by Location\n",
    "df.groupBy('Location').count().sort('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------------------+-----+-------------------+\n",
      "| Latitude| Longitude|                Time|Label|           Location|\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "|53.480759| -2.242631|Mon Oct 05 08:21:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:22:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:22:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:23:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:24:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:28:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:30:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:32:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:33:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:33:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:34:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:42:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:43:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 09:11:...|    1|Manchester  England|\n",
      "|-5.398840|105.242250|Mon Oct 05 12:18:...|    2|Manchester  England|\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "| Latitude| Longitude|                Time|Label|           Location|\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "|53.480759| -2.242631|Mon Oct 05 08:21:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:22:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:22:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:23:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:24:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:28:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:30:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:32:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:33:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:33:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:34:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:42:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:43:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 09:11:...|    1|Manchester  England|\n",
      "|-5.398840|105.242250|Mon Oct 05 12:18:...|    2|Manchester  England|\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "| Latitude| Longitude|                Time|Label|           Location|\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "|53.480759| -2.242631|Mon Oct 05 08:21:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:22:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:22:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:23:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:24:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:28:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:30:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:32:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:33:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:33:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:34:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:42:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 08:43:...|    1|Manchester  England|\n",
      "|53.480759| -2.242631|Mon Oct 05 09:11:...|    1|Manchester  England|\n",
      "|-5.398840|105.242250|Mon Oct 05 12:18:...|    2|Manchester  England|\n",
      "+---------+----------+--------------------+-----+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Select by location\n",
    "df.filter(df['Location'] == 'Manchester  England').show()\n",
    "\n",
    "# Pandas-like syntax\n",
    "df[df['Location'] == 'Manchester  England'].show()\n",
    "\n",
    "# Register the DataFrame as a SQL temporary view\n",
    "df.createOrReplaceTempView(\"spatialdata\")\n",
    "sqlDF = spark.sql(\"SELECT * FROM spatialdata WHERE Location == 'Manchester  England' \")\n",
    "sqlDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Pandas Dataframe: \n",
      "Latitude     object\n",
      "Longitude    object\n",
      "Time         object\n",
      "Label        object\n",
      "Location     object\n",
      "dtype: object\n",
      "Spark Dataframe: \n",
      "DataFrame[Latitude: string, Longitude: string, Time: string, Label: string, Location: string]\n"
     ]
    }
   ],
   "source": [
    "# Convert Spark DataFrame to Pandas\n",
    "pandas_df = df.toPandas()\n",
    "print 'Pandas Dataframe: '\n",
    "print pandas_df.dtypes\n",
    "\n",
    "# Create a Spark DataFrame from Pandas\n",
    "spark_df = spark.createDataFrame(pandas_df)\n",
    "print 'Spark Dataframe: '\n",
    "print spark_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "spark.stop()\n",
    "del spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Demo 3: Geospatial Analysis with Apache Spark\n",
    "\n",
    "Spatial funcitonalities do not supported by original Apache Spark distributions\n",
    "\n",
    "![](images/spark_packagess.png)\n",
    "\n",
    "## SpatialSpark (88 🌟 in Github)\n",
    "\n",
    "SpatialSpark aims to provide efficient spatial operations using Apache Spark. It can be used as a Spark library for spatial extension as well as a standalone application to process large scale spatial join operations.\n",
    "\n",
    "SpatialSpark has been compiled and tested on Spark 2.0.2. For geometry operations and data structures for indexes, well known JTS library is used.\n",
    "\n",
    "Reference: \n",
    "1. http://simin.me/projects/spatialspark/\n",
    "2. https://github.com/syoummer/SpatialSpark\n",
    "\n",
    "\n",
    "## GeoSpark (134 🌟 in Github)\n",
    "\n",
    "GeoSpark is a cluster computing system for processing large-scale spatial data. GeoSpark extends Apache Spark with a set of out-of-the-box **Spatial Resilient Distributed Datasets (SRDDs)** that efficiently load, process, and analyze large-scale spatial data across machines. GeoSpark provides APIs for Apache Spark programmer to easily develop their spatial analysis programs with Spatial Resilient Distributed Datasets (SRDDs) which have in house support for **geometrical and Spatial Queries (Range, K Nearest Neighbors, Join)**.\n",
    "\n",
    "Reference:\n",
    "1. http://geospark.datasyslab.org/\n",
    "2. https://github.com/DataSystemsLab/GeoSpark\n",
    "\n",
    "\n",
    "## Magellan (249 🌟 in Github)\n",
    "\n",
    "Magellan is an open source library Geospatial Analytics using Spark as the underlying engine. We leverage Catalyst’s pluggable optimizer to efficiently execute spatial joins, SparkSQL’s powerful operators to express geometric queries in a natural DSL, and Pyspark’s Python integration to **provide Python bindings**.\n",
    "\n",
    "Currently supported files:\n",
    "- **ESRI format files** \n",
    "- **GeoJSON**.\n",
    "\n",
    "Currently supported capabilities:\n",
    "- **Geometries**: Point, LineString, Polygon, MultiPoint\n",
    "- **Predicates**: Intersects, Within, Contains\n",
    "- **Operations**: Intersection\n",
    "\n",
    "Scala and Python API\n",
    "\n",
    "Reference:\n",
    "1. https://de.hortonworks.com/blog/magellan-geospatial-analytics-in-spark/\n",
    "2. https://github.com/harsha2010/magellan"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext, SQLContext\n",
    "\n",
    "conf = SparkConf().setMaster(\"local\").setAppName(\"My App\") \n",
    "sc = SparkContext(conf = conf)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "TypeError",
     "evalue": "'JavaPackage' object is not callable",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m\u001b[0m",
      "\u001b[1;31mTypeError\u001b[0mTraceback (most recent call last)",
      "\u001b[1;32m<ipython-input-24-6e9e960615f0>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      4\u001b[0m \u001b[0mmagellan\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0msc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmagellan\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      5\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 6\u001b[1;33m \u001b[1;32mprint\u001b[0m \u001b[0mmagellan\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mtypes\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPoint\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m-\u001b[0m\u001b[1;36m1.0\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m-\u001b[0m\u001b[1;36m1.0\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      7\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      8\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mnamestr\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mobj\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mnamespace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mTypeError\u001b[0m: 'JavaPackage' object is not callable"
     ]
    }
   ],
   "source": [
    "#from magellan.types import Point, Polygon\n",
    "from pyspark.sql import Row, SQLContext\n",
    "\n",
    "magellan = sc._jvm.magellan\n",
    "\n",
    "print magellan.types.Point(-1.0, -1.0)\n",
    "\n",
    "def namestr(obj, namespace):\n",
    "    return [name for name in namespace if namespace[name] is obj]\n",
    "namestr(magellan, globals())\n",
    "\n",
    "#PointRecord = Row(\"id\", \"point\")\n",
    "\n",
    "#points = sc.parallelize([\n",
    "#  (0, Point(-1.0, -1.0)),\n",
    "#  (1, Point(-1.0, 1.0)),\n",
    "#  (2, Point(1.0, -1.0))])\\\n",
    "#.map(lambda x: PointRecord(*x))\\\n",
    "#.toDF()\n",
    "\n",
    "#points.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc.stop()\n",
    "del sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  },
  "toc": {
   "toc_cell": true,
   "toc_number_sections": true,
   "toc_section_display": "block",
   "toc_threshold": 6,
   "toc_window_display": true
  },
  "toc_position": {
   "height": "594px",
   "left": "1531.83px",
   "right": "20px",
   "top": "158px",
   "width": "327px"
  },
  "widgets": {
   "state": {},
   "version": "1.1.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}