part03/week04_160525/BIGBIO_LearningSpark_ch9_SPARKSQL.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "참고할 문서 링크\n",
    "\n",
    "http://www.slideshare.net/juhuipark37/spark-sql-55494235 # 책 내용 CH9 sparkSQL 아꿈사 박주희님 자료   \n",
    "http://www.slideshare.net/KangDognhyun/spark-sql-57229920 # SPARK RDD, DataFrame, Dataset 비교 등 - 강동현님 자료 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## CH9. SPARKSQL 실습"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 실습파일 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!git clone https://github.com/databricks/learning-spark.git"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext() # SparkContext 객체가 sc 변수에 만들어진다. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 예제 9-8 파이썬에서 SQL 컨텍스트 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hiveCtx = HiveContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### HiveContext를 사용할 경우 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import HiveContext, Row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "hiveCtx = HiveContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### SQLContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 기본쿼리예제"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 예제 9-11 파이썬에서 트윗 데이터 읽어서 질의하기 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/home/jovyan/work/learning-spark/files\n"
     ]
    }
   ],
   "source": [
    "cd /home/jovyan/work/learning-spark/files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 72\r\n",
      "-rw-r--r-- 1 jovyan   69 May 24 08:14 callsigns\r\n",
      "-rw-r--r-- 1 jovyan    7 May 24 08:14 call_signs2.txt\r\n",
      "-rw-r--r-- 1 jovyan 8194 May 24 08:14 callsign_tbl\r\n",
      "-rw-r--r-- 1 jovyan 8194 May 24 08:14 callsign_tbl_sorted\r\n",
      "-rw-r--r-- 1 jovyan  288 May 24 08:14 cqlsh_setup\r\n",
      "drwxr-xr-x 2 jovyan 4096 May 24 08:14 \u001b[0m\u001b[01;34mfake_logs\u001b[0m/\r\n",
      "-rw-r--r-- 1 jovyan   42 May 24 08:14 favourite_animals.csv\r\n",
      "-rw-r--r-- 1 jovyan  656 May 24 08:14 flumeconf.cfg\r\n",
      "-rw-r--r-- 1 jovyan  764 May 24 08:14 ham.txt\r\n",
      "-rw-r--r-- 1 jovyan   34 May 24 08:14 happypandas\r\n",
      "-rw-r--r-- 1 jovyan   25 May 24 08:14 int_string.csv\r\n",
      "-rw-r--r-- 1 jovyan  146 May 24 08:14 pandainfo.json\r\n",
      "-rw-r--r-- 1 jovyan  501 May 24 08:14 spam.txt\r\n",
      "-rw-r--r-- 1 jovyan 1493 May 24 08:14 testweet.json\r\n"
     ]
    }
   ],
   "source": [
    "ll"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 여기부터 보시는 분들을 위해 처음부터 다시"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 파이썬에서 SQL Context 생성 : HiveContext 로 시작 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import HiveContext, Row\n",
    "hiveCtx = HiveContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### testweet.json을 데이터로 사용 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/home/jovyan/work/learning-spark/files\n"
     ]
    }
   ],
   "source": [
    "cd ~/work/learning-spark/files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "callsigns            \u001b[0m\u001b[01;34mfake_logs\u001b[0m/             int_string.csv  \u001b[01;34mspark\u001b[0m/\r\n",
      "call_signs2.txt      favourite_animals.csv  pandainfo.json  testjoin.join\r\n",
      "callsign_tbl         flumeconf.cfg          people.json     testjson.json\r\n",
      "callsign_tbl_sorted  ham.txt                people.txt      testweet.json\r\n",
      "cqlsh_setup          happypandas            spam.txt\r\n"
     ]
    }
   ],
   "source": [
    "ls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\"createdAt\":\"Nov 4, 2014 4:56:59 PM\",\"id\":529799371026485248,\"text\":\"Adventures With Coffee, Code, and Writing.\",\"source\":\"\\u003ca href\\u003d\\\"http://twitter.com\\\" rel\\u003d\\\"nofollow\\\"\\u003eTwitter Web Client\\u003c/a\\u003e\",\"isTruncated\":false,\"inReplyToStatusId\":-1,\"inReplyToUserId\":-1,\"isFavorited\":false,\"retweetCount\":0,\"isPossiblySensitive\":false,\"contributorsIDs\":[],\"userMentionEntities\":[],\"urlEntities\":[],\"hashtagEntities\":[],\"mediaEntities\":[],\"currentUserRetweetId\":-1,\"user\":{\"id\":15594928,\"name\":\"Holden Karau\",\"screenName\":\"holdenkarau\",\"location\":\"\",\"description\":\"\",\"descriptionURLEntities\":[],\"isContributorsEnabled\":false,\"profileImageUrl\":\"http://pbs.twimg.com/profile_images/3005696115/2036374bbadbed85249cdd50aac6e170_normal.jpeg\",\"profileImageUrlHttps\":\"https://pbs.twimg.com/profile_images/3005696115/2036374bbadbed85249cdd50aac6e170_normal.jpeg\",\"isProtected\":false,\"followersCount\":1231,\"profileBackgroundColor\":\"C0DEED\",\"profileTextColor\":\"333333\",\"profileLinkColor\":\"0084B4\",\"profileSidebarFillColor\":\"DDEEF6\",\"profileSidebarBorderColor\":\"FFFFFF\",\"profileUseBackgroundImage\":true,\"showAllInlineMedia\":false,\"friendsCount\":600,\"createdAt\":\"Aug 5, 2011 9:42:44 AM\",\"favouritesCount\":1095,\"utcOffset\":-3,\"profileBackgroundImageUrl\":\"\",\"profileBackgroundImageUrlHttps\":\"\",\"profileBannerImageUrl\":\"\",\"profileBackgroundTiled\":true,\"lang\":\"en\",\"statusesCount\":6234,\"isGeoEnabled\":true,\"isVerified\":false,\"translator\":false,\"listedCount\":0,\"isFollowRequestSent\":false}}\r\n"
     ]
    }
   ],
   "source": [
    "!cat testweet.json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1차 오류"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "ename": "Py4JJavaError",
     "evalue": "An error occurred while calling o18.json.\n: java.io.IOException: No input paths specified in job\n\tat org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:201)\n\tat org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313)\n\tat org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.rdd.RDD.partitions(RDD.scala:237)\n\tat org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.rdd.RDD.partitions(RDD.scala:237)\n\tat org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.rdd.RDD.partitions(RDD.scala:237)\n\tat org.apache.spark.rdd.RDD$$anonfun$treeAggregate$1.apply(RDD.scala:1115)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:316)\n\tat org.apache.spark.rdd.RDD.treeAggregate(RDD.scala:1113)\n\tat org.apache.spark.sql.execution.datasources.json.InferSchema$.infer(InferSchema.scala:65)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation$$anonfun$4.apply(JSONRelation.scala:114)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation$$anonfun$4.apply(JSONRelation.scala:109)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation.dataSchema$lzycompute(JSONRelation.scala:109)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation.dataSchema(JSONRelation.scala:108)\n\tat org.apache.spark.sql.sources.HadoopFsRelation.schema$lzycompute(interfaces.scala:636)\n\tat org.apache.spark.sql.sources.HadoopFsRelation.schema(interfaces.scala:635)\n\tat org.apache.spark.sql.execution.datasources.LogicalRelation.<init>(LogicalRelation.scala:37)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:125)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:109)\n\tat org.apache.spark.sql.DataFrameReader.json(DataFrameReader.scala:244)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.lang.reflect.Method.invoke(Method.java:606)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)\n\tat py4j.Gateway.invoke(Gateway.java:259)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:209)\n\tat java.lang.Thread.run(Thread.java:745)\n",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-5-0ad854ca8bb5>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[0minputFile\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;34m\"testweet.json\"\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[1;31m#input = hiveCtx.jsonFile(inputFile)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 3\u001b[1;33m \u001b[0minput\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mhiveCtx\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mjson\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minputFile\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32m/usr/local/spark/python/pyspark/sql/readwriter.py\u001b[0m in \u001b[0;36mjson\u001b[1;34m(self, path, schema)\u001b[0m\n\u001b[0;32m    174\u001b[0m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mschema\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mschema\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    175\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0misinstance\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mbasestring\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 176\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_df\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jreader\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mjson\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    177\u001b[0m         \u001b[1;32melif\u001b[0m \u001b[0mtype\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;33m==\u001b[0m \u001b[0mlist\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    178\u001b[0m             \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_df\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jreader\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mjson\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sqlContext\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPythonUtils\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mtoSeq\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/spark/python/lib/py4j-0.9-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m    811\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    812\u001b[0m         return_value = get_return_value(\n\u001b[1;32m--> 813\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m    814\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    815\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/spark/python/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     43\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mdeco\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     44\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 45\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     46\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     47\u001b[0m             \u001b[0ms\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mjava_exception\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mtoString\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/spark/python/lib/py4j-0.9-src.zip/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    306\u001b[0m                 raise Py4JJavaError(\n\u001b[0;32m    307\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 308\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    309\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    310\u001b[0m                 raise Py4JError(\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o18.json.\n: java.io.IOException: No input paths specified in job\n\tat org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:201)\n\tat org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313)\n\tat org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:199)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.rdd.RDD.partitions(RDD.scala:237)\n\tat org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.rdd.RDD.partitions(RDD.scala:237)\n\tat org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)\n\tat org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.rdd.RDD.partitions(RDD.scala:237)\n\tat org.apache.spark.rdd.RDD$$anonfun$treeAggregate$1.apply(RDD.scala:1115)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:316)\n\tat org.apache.spark.rdd.RDD.treeAggregate(RDD.scala:1113)\n\tat org.apache.spark.sql.execution.datasources.json.InferSchema$.infer(InferSchema.scala:65)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation$$anonfun$4.apply(JSONRelation.scala:114)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation$$anonfun$4.apply(JSONRelation.scala:109)\n\tat scala.Option.getOrElse(Option.scala:120)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation.dataSchema$lzycompute(JSONRelation.scala:109)\n\tat org.apache.spark.sql.execution.datasources.json.JSONRelation.dataSchema(JSONRelation.scala:108)\n\tat org.apache.spark.sql.sources.HadoopFsRelation.schema$lzycompute(interfaces.scala:636)\n\tat org.apache.spark.sql.sources.HadoopFsRelation.schema(interfaces.scala:635)\n\tat org.apache.spark.sql.execution.datasources.LogicalRelation.<init>(LogicalRelation.scala:37)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:125)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:109)\n\tat org.apache.spark.sql.DataFrameReader.json(DataFrameReader.scala:244)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.lang.reflect.Method.invoke(Method.java:606)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)\n\tat py4j.Gateway.invoke(Gateway.java:259)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:209)\n\tat java.lang.Thread.run(Thread.java:745)\n"
     ]
    }
   ],
   "source": [
    "inputFile = \"testweet.json\"\n",
    "input = hiveCtx.jsonFile(inputFile)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 이제 jsonFile 이 아니라 read.json 으로 바뀌었다고 합니다 . (책에는 jsonFile로 나와있음)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "http://spark.apache.org/docs/latest/sql-programming-guide.html#json-datasets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 예제를 한번 보고 넘어갈까요~!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# sc is an existing SparkContext.\n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "# A JSON dataset is pointed to by path.\n",
    "# The path can be either a single text file or a directory storing text files.\n",
    "people = sqlContext.read.json(\"examples/src/main/resources/people.json\")\n",
    "\n",
    "# The inferred schema can be visualized using the printSchema() method.\n",
    "people.printSchema()\n",
    "# root\n",
    "#  |-- age: integer (nullable = true)\n",
    "#  |-- name: string (nullable = true)\n",
    "\n",
    "# Register this DataFrame as a table.\n",
    "people.registerTempTable(\"people\")\n",
    "\n",
    "# SQL statements can be run by using the sql methods provided by `sqlContext`.\n",
    "teenagers = sqlContext.sql(\"SELECT name FROM people WHERE age >= 13 AND age <= 19\")\n",
    "\n",
    "# Alternatively, a DataFrame can be created for a JSON dataset represented by\n",
    "# an RDD[String] storing one JSON object per string.\n",
    "anotherPeopleRDD = sc.parallelize([\n",
    "  '{\"name\":\"Yin\",\"address\":{\"city\":\"Columbus\",\"state\":\"Ohio\"}}'])\n",
    "anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 하지만 다시 에러발생"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "inputFile = \"testweet.json\"\n",
    "#input = hiveCtx.jsonFile(inputFile)\n",
    "input = hiveCtx.read.json(inputFile)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "아무리 해도 에러를 잡을 수가~!!! ㅠ.ㅠ  \n",
    "그냥 파일이름만 쓰지 맙시다.   \n",
    "그리고 반드시 상대경로가 아닌 절대경로를 사용.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import HiveContext, Row\n",
    "hiveCtx = HiveContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "ename": "Exception",
     "evalue": "(\"You must build Spark with Hive. Export 'SPARK_HIVE=true' and run build/sbt assembly\", Py4JJavaError(u'An error occurred while calling None.org.apache.spark.sql.hive.HiveContext.\\n', JavaObject id=o868))",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mException\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-56-59035fe0f6f0>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[0minputFile\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;34m\"/home/jovyan/work/learning-spark/files/testweet.json\"\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 2\u001b[1;33m \u001b[0minput\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mhiveCtx\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mjson\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minputFile\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32m/usr/local/spark/python/pyspark/sql/context.py\u001b[0m in \u001b[0;36mread\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    661\u001b[0m         \u001b[1;33m:\u001b[0m\u001b[1;32mreturn\u001b[0m\u001b[1;33m:\u001b[0m \u001b[1;33m:\u001b[0m\u001b[1;32mclass\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m`\u001b[0m\u001b[0mDataFrameReader\u001b[0m\u001b[1;33m`\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    662\u001b[0m         \"\"\"\n\u001b[1;32m--> 663\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mDataFrameReader\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    664\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    665\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/spark/python/pyspark/sql/readwriter.py\u001b[0m in \u001b[0;36m__init__\u001b[1;34m(self, sqlContext)\u001b[0m\n\u001b[0;32m     54\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     55\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0m__init__\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0msqlContext\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 56\u001b[1;33m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jreader\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0msqlContext\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_ssql_ctx\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     57\u001b[0m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sqlContext\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0msqlContext\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     58\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/spark/python/pyspark/sql/context.py\u001b[0m in \u001b[0;36m_ssql_ctx\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    689\u001b[0m             raise Exception(\"You must build Spark with Hive. \"\n\u001b[0;32m    690\u001b[0m                             \u001b[1;34m\"Export 'SPARK_HIVE=true' and run \"\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 691\u001b[1;33m                             \"build/sbt assembly\", e)\n\u001b[0m\u001b[0;32m    692\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    693\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0m_get_hive_ctx\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mException\u001b[0m: (\"You must build Spark with Hive. Export 'SPARK_HIVE=true' and run build/sbt assembly\", Py4JJavaError(u'An error occurred while calling None.org.apache.spark.sql.hive.HiveContext.\\n', JavaObject id=o868))"
     ]
    }
   ],
   "source": [
    "inputFile = \"/home/jovyan/work/learning-spark/files/testweet.json\"\n",
    "input = hiveCtx.read.json(inputFile)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "경로 문제는 해결하였으나, 현재 사용하는 서버에는 HIVE가 연결되어있지 않아서 hiveCtx 는 사용할 수가 없었음. \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "대신 SQLContext 사용하기로 함"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "inputFile = \"/home/jovyan/work/learning-spark/files/testweet.json\"\n",
    "input = sqlContext.read.json(inputFile)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 입력 데이터프레임 등록\n",
    "input.registerTempTable(\"tweets\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# retweetCount 기준으로 트윗들을 가져온다. \n",
    "topTweets = sqlContext.sql(\"\"\"SELECT text, \n",
    "retweetCount FROM tweets ORDER BY retweetCount LIMIT 10\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[text: string, retweetCount: bigint]"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "topTweets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------+\n",
      "|                text|retweetCount|\n",
      "+--------------------+------------+\n",
      "|Adventures With C...|           0|\n",
      "+--------------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "topTweets.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "그럼 한가지 더 해볼까요."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://github.com/trustedanalytics/jupyter-default-notebooks/blob/master/notebooks/examples/spark/spark-sql-example.ipynb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Create a SqlContext from the SparkContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\n",
    "data = [\n",
    "    (1, 'a'), \n",
    "    (2, 'b'), \n",
    "    (3, 'c'), \n",
    "    (4, 'd'), \n",
    "    (5, 'e'), \n",
    "    (6, 'a'), \n",
    "    (7, 'b'), \n",
    "    (8, 'c'), \n",
    "    (9, 'd'), \n",
    "    (10, 'e')\n",
    "]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>numbers</th>\n",
       "      <th>letters</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>3</td>\n",
       "      <td>c</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>4</td>\n",
       "      <td>d</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>5</td>\n",
       "      <td>e</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>6</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>7</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>8</td>\n",
       "      <td>c</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>9</td>\n",
       "      <td>d</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>10</td>\n",
       "      <td>e</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   numbers letters\n",
       "0        1       a\n",
       "1        2       b\n",
       "2        3       c\n",
       "3        4       d\n",
       "4        5       e\n",
       "5        6       a\n",
       "6        7       b\n",
       "7        8       c\n",
       "8        9       d\n",
       "9       10       e"
      ]
     },
     "execution_count": 69,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Convert a local data set into a DataFrame\n",
    "df = sqlContext.createDataFrame(data, ['numbers', 'letters'])\n",
    "\n",
    "# Convert to a Pandas DataFrame for easy display\n",
    "df.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>numbers</th>\n",
       "      <th>letters</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>3</td>\n",
       "      <td>c</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>4</td>\n",
       "      <td>d</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>5</td>\n",
       "      <td>e</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>6</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>7</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>8</td>\n",
       "      <td>c</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>9</td>\n",
       "      <td>d</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>10</td>\n",
       "      <td>e</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   numbers letters\n",
       "0        1       a\n",
       "1        2       b\n",
       "2        3       c\n",
       "3        4       d\n",
       "4        5       e\n",
       "5        6       a\n",
       "6        7       b\n",
       "7        8       c\n",
       "8        9       d\n",
       "9       10       e"
      ]
     },
     "execution_count": 70,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Register the DataFrame as a table\n",
    "df.registerTempTable(\"mytable\")\n",
    "\n",
    "# Peform a simple select from the table\n",
    "results = sqlContext.sql(\"select * from mytable\")\n",
    "\n",
    "# Convert the results to a Pandas DataFrame for easy viewing\n",
    "results.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>numbers</th>\n",
       "      <th>letters</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>7</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>6</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>5</td>\n",
       "      <td>e</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>4</td>\n",
       "      <td>d</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>3</td>\n",
       "      <td>c</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>2</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>1</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   numbers letters\n",
       "0        7       b\n",
       "1        6       a\n",
       "2        5       e\n",
       "3        4       d\n",
       "4        3       c\n",
       "5        2       b\n",
       "6        1       a"
      ]
     },
     "execution_count": 71,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "# Perform a query with a where clause and order by\n",
    "results = sqlContext.sql(\"select * from mytable where numbers < 8 order by numbers desc\")\n",
    "\n",
    "# Convert the results to a Pandas DataFrame for easy viewing\n",
    "results.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>letters</th>\n",
       "      <th>count</th>\n",
       "      <th>avg</th>\n",
       "      <th>sum</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>a</td>\n",
       "      <td>2</td>\n",
       "      <td>3.5</td>\n",
       "      <td>7</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>b</td>\n",
       "      <td>2</td>\n",
       "      <td>4.5</td>\n",
       "      <td>9</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>c</td>\n",
       "      <td>2</td>\n",
       "      <td>5.5</td>\n",
       "      <td>11</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>d</td>\n",
       "      <td>2</td>\n",
       "      <td>6.5</td>\n",
       "      <td>13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>e</td>\n",
       "      <td>2</td>\n",
       "      <td>7.5</td>\n",
       "      <td>15</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  letters  count  avg  sum\n",
       "0       a      2  3.5    7\n",
       "1       b      2  4.5    9\n",
       "2       c      2  5.5   11\n",
       "3       d      2  6.5   13\n",
       "4       e      2  7.5   15"
      ]
     },
     "execution_count": 72,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "# Perform a more complex query on the table\n",
    "results = sqlContext.sql(\"select letters, count(*) as count, avg(numbers) as avg, sum(numbers) as sum from mytable group by letters\")\n",
    "\n",
    "# Convert the results to a Pandas DataFrame for easy viewing\n",
    "results.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 데이터 프레임 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "http://www.slideshare.net/juhuipark37/spark-sql-55494235"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 예제 9-14 파이썬의 topTweets 데이터프레임에서 문자열 칼럼 접근하기"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "topTweetText = topTweets.rdd.map(lambda row: row.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 캐싱"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "cacheTable(“tableName”) 을 호출하여\u000b",
    "메모리상의 컬럼 지향 포맷으로 테이블을 캐쉬할 수 있다.  \n",
    "메모리에 드라이버 프로그램이 실행되는 동안 유효  \n",
    "HiveSQL이나 SQL 문장으로도 테이블 캐시 가능   \n",
    "Cache Table “TableName” or UnCache  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "스파크 SQL을 기존의 하이브에 연결하려면, 하이브 설정을 제공해주어야 함.  \n",
    "hive-site.xml 파일을 스파크의 ./conf 디렉터리에 복사해주면 된다. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 데이터 불러오고 저장하기 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 아파치 하이브 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 예제 9-15 파이썬에서 하이브 읽기"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import HiveContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "hiveCtx = HiveContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rows = hiveCtx.sql(\"SELECT key, value FROM mytable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "keys = rows.map(lambda row: row[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 데이터소스 / 파케이 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Cloning into 'spark'...\n",
      "remote: Counting objects: 360682, done.\u001b[K\n",
      "remote: Compressing objects: 100% (18/18), done.\u001b[K\n",
      "remote: Total 360682 (delta 4), reused 0 (delta 0), pack-reused 360664\u001b[K\n",
      "Receiving objects: 100% (360682/360682), 177.41 MiB | 9.78 MiB/s, done.\n",
      "Resolving deltas: 100% (137322/137322), done.\n",
      "Checking connectivity... done.\n",
      "Checking out files: 100% (13303/13303), done.\n"
     ]
    }
   ],
   "source": [
    "!git clone https://github.com/apache/spark.git"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 파이썬에서 파케이 읽기"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rows = hiveCtx.load(parquetFile, \"parquet\")\n",
    "names = rows.map(lambda row: row.name)\n",
    "print \"Everyone\"\n",
    "print names.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 파케이에 질의(쿼리)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "tbl = rows.registerTempTable(\"people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pandaFriends = hiveCtx.sql(\"SELECT name FROM people WHERE favoriteAnimal = \\\"panda\\\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "print \"Panda friends\"\n",
    "print pandaFriends.map(lambda row: row.name).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 파케이 파일 저장 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pandaFriends.save(\"hdfs://...\", \"parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## JSON"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 입력레코드"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "{\"name\": \"Holden\"}\n",
    "{\"name\": \"Sparky The Bear\", \"lovesPandas\":true, \"knows\":{\"friends\":[\"holden\"]}}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 파이썬에서 스파크SQL로 JSON 불러오기 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "input = hiveCtx.read.json(inputFile)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 스키마 출력"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### SQL 쿼리로 중접 및 배열 데이터 요청"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "select hashtagEntities[0].text from tweets LIMIT 1;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### RDD에서 가져오기 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "RDD로부터 데이터프레임을 만들어낼수 있다. \n",
    "파이썬에서는 Row 객체들의 RDD를 만들고 inferSchema()를 부르면 된다. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 파이썬에서 Row와 이름있는 튜플로 DataFrame 만들기"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "happyPeopleRDD = sc.paralleliza([Row(name=\"holden\", favouriteBeverage=\"coffee\")])\n",
    "happyPeopleDataFrame = hiveCtx.inferSchema(happyPeopleRDD)\n",
    "happyPeopleDataFrame.registerTempTable(\"happy_people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### JDBC/ODBC -- BI TOOL과 연결"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Tableau와 Spark 연결"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://www.youtube.com/watch?v=OKcIf6UdK7c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 참고할 만한 SPARK 예제들이 많이 있습니다. 꼭 한번 읽어보세요"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://github.com/hibariba/all-spark-notebook-jupyter-demo/tree/master/notebooks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://github.com/hibariba/all-spark-notebook-jupyter-demo/blob/master/notebooks/research-notebook-example.ipynb"
   ]
  }
 ],
 "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.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}