spark_yarn.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [],
   "source": [
    "#SparkSession의 연결을 끊기 위하여\n",
    "quit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark 데이터 파이프라인 - yarn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import Row"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## SparkSession\n",
    "\n",
    "- spark의 기능의 시작점\n",
    "- 데이터 프레임 기능을 사용하기 위함"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/usr/local/spark-3.3.2-bin-hadoop3/jars/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /home/bigdata/.ivy2/cache\n",
      "The jars for the packages stored in: /home/bigdata/.ivy2/jars\n",
      "org.mongodb.spark#mongo-spark-connector_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-c0dbcf04-e066-4fe2-aca1-4ba4a862bfab;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.mongodb.spark#mongo-spark-connector_2.12;3.0.1 in central\n",
      "\tfound org.mongodb#mongodb-driver-sync;4.0.5 in central\n",
      "\tfound org.mongodb#bson;4.0.5 in central\n",
      "\tfound org.mongodb#mongodb-driver-core;4.0.5 in central\n",
      ":: resolution report :: resolve 266ms :: artifacts dl 11ms\n",
      "\t:: modules in use:\n",
      "\torg.mongodb#bson;4.0.5 from central in [default]\n",
      "\torg.mongodb#mongodb-driver-core;4.0.5 from central in [default]\n",
      "\torg.mongodb#mongodb-driver-sync;4.0.5 from central in [default]\n",
      "\torg.mongodb.spark#mongo-spark-connector_2.12;3.0.1 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   4   |   0   |   0   |   0   ||   4   |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-c0dbcf04-e066-4fe2-aca1-4ba4a862bfab\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 4 already retrieved (0kB/11ms)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/03/27 23:55:25 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/03/27 23:55:28 WARN Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.\n",
      "23/03/27 23:55:38 WARN Client: Same path resource file:///home/bigdata/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.12-3.0.1.jar added multiple times to distributed cache.\n",
      "23/03/27 23:55:38 WARN Client: Same path resource file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-sync-4.0.5.jar added multiple times to distributed cache.\n",
      "23/03/27 23:55:38 WARN Client: Same path resource file:///home/bigdata/.ivy2/jars/org.mongodb_bson-4.0.5.jar added multiple times to distributed cache.\n",
      "23/03/27 23:55:38 WARN Client: Same path resource file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-core-4.0.5.jar added multiple times to distributed cache.\n",
      "CPU times: user 63.4 ms, sys: 82 ms, total: 145 ms\n",
      "Wall time: 36.4 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "yarn = SparkSession\\\n",
    "        .builder\\\n",
    "        .master('yarn')\\\n",
    "        .appName('twitter_yarn')\\\n",
    "        .config('spark.submit.deployMode', 'client')\\\n",
    "        .config('spark.executor.instances', 4)\\\n",
    "        .config('spark.executor.memory', '2g')\\\n",
    "        .config('spark.driver.memory', '2g')\\\n",
    "        .config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector_2.12:3.0.1')\\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('spark.eventLog.enabled', 'true'),\n",
       " ('spark.executor.instances', '4'),\n",
       " ('spark.driver.extraJavaOptions',\n",
       "  '-XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED'),\n",
       " ('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector_2.12:3.0.1'),\n",
       " ('spark.driver.appUIAddress', 'http://master:4040'),\n",
       " ('spark.org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter.param.PROXY_HOSTS',\n",
       "  'master'),\n",
       " ('spark.sql.warehouse.dir', 'hdfs://master/user/bigdata/spark-warehouse'),\n",
       " ('spark.executorEnv.PYTHONPATH',\n",
       "  '{{PWD}}/pyspark.zip<CPS>{{PWD}}/py4j-0.10.9.5-src.zip<CPS>{{PWD}}/org.mongodb.spark_mongo-spark-connector_2.12-3.0.1.jar<CPS>{{PWD}}/org.mongodb_mongodb-driver-sync-4.0.5.jar<CPS>{{PWD}}/org.mongodb_bson-4.0.5.jar<CPS>{{PWD}}/org.mongodb_mongodb-driver-core-4.0.5.jar'),\n",
       " ('spark.app.id', 'application_1679906880075_0002'),\n",
       " ('spark.org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter.param.PROXY_URI_BASES',\n",
       "  'http://master:8088/proxy/application_1679906880075_0002'),\n",
       " ('spark.ui.killEnabled', 'true'),\n",
       " ('spark.eventLog.dir', 'hdfs://master/user/bigdata/spark_eventLog'),\n",
       " ('spark.ui.port', '4040'),\n",
       " ('spark.repl.local.jars',\n",
       "  'file:///home/bigdata/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.12-3.0.1.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-sync-4.0.5.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_bson-4.0.5.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-core-4.0.5.jar'),\n",
       " ('spark.yarn.secondary.jars',\n",
       "  'org.mongodb.spark_mongo-spark-connector_2.12-3.0.1.jar,org.mongodb_mongodb-driver-sync-4.0.5.jar,org.mongodb_bson-4.0.5.jar,org.mongodb_mongodb-driver-core-4.0.5.jar'),\n",
       " ('spark.serializer.objectStreamReset', '100'),\n",
       " ('spark.submit.pyFiles',\n",
       "  '/home/bigdata/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.12-3.0.1.jar,/home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-sync-4.0.5.jar,/home/bigdata/.ivy2/jars/org.mongodb_bson-4.0.5.jar,/home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-core-4.0.5.jar'),\n",
       " ('spark.submit.deployMode', 'client'),\n",
       " ('spark.ui.filters',\n",
       "  'org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter'),\n",
       " ('spark.executor.memory', '2g'),\n",
       " ('spark.app.startTime', '1679928925513'),\n",
       " ('spark.driver.host', 'master'),\n",
       " ('spark.ui.enabled', 'true'),\n",
       " ('spark.app.name', 'twitter_yarn'),\n",
       " ('spark.executor.id', 'driver'),\n",
       " ('spark.yarn.dist.pyFiles',\n",
       "  'file:///home/bigdata/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.12-3.0.1.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-sync-4.0.5.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_bson-4.0.5.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-core-4.0.5.jar'),\n",
       " ('spark.master', 'yarn'),\n",
       " ('spark.driver.port', '38373'),\n",
       " ('spark.rdd.compress', 'True'),\n",
       " ('spark.executor.extraJavaOptions',\n",
       "  '-XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED'),\n",
       " ('spark.ui.proxyBase', '/proxy/application_1679906880075_0002'),\n",
       " ('spark.driver.memory', '2g'),\n",
       " ('spark.yarn.dist.jars',\n",
       "  'file:///home/bigdata/.ivy2/jars/org.mongodb.spark_mongo-spark-connector_2.12-3.0.1.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-sync-4.0.5.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_bson-4.0.5.jar,file:///home/bigdata/.ivy2/jars/org.mongodb_mongodb-driver-core-4.0.5.jar'),\n",
       " ('spark.yarn.isPython', 'true'),\n",
       " ('spark.ui.showConsoleProgress', 'true'),\n",
       " ('spark.app.submitTime', '1679928925288')]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "yarn.sparkContext.getConf().getAll()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 데이터 파이프라인"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.5 ms, sys: 13 ms, total: 15.5 ms\n",
      "Wall time: 7 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df = yarn.read.format('mongo')\\\n",
    "               .option('uri', 'mongodb://192.168.56.101:27017/twitter.sample')\\\n",
    "               .load()\n",
    "\n",
    "df.createOrReplaceTempView('tweets')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2 µs, sys: 2 µs, total: 4 µs\n",
      "Wall time: 4.77 µs\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "type(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "mongoDB로부터 저장되어있는 데이터를 불러오고 데이터프레임의 형태로 저장합니다.\n",
    "\n",
    "`createOrReplaceTempView`함수로 `sql`을 사용하여 핸들링할 수 있도록 `view`에 `tweets`라는 이름으로 등록합니다."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 전체 데이터 수 확인"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### sql을 사용하여 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 1:=======================================================> (42 + 1) / 43]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "| 7739109|\n",
      "+--------+\n",
      "\n",
      "CPU times: user 57.9 ms, sys: 86.6 ms, total: 145 ms\n",
      "Wall time: 36.9 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "query ='''\n",
    "SELECT count(*)\n",
    "FROM tweets\n",
    "'''\n",
    "yarn.sql(query).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Dataframe 함수를 사용하여 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 4:=======================================================> (42 + 1) / 43]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 92.6 ms, sys: 54.7 ms, total: 147 ms\n",
      "Wall time: 30.7 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "7739109"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "전체 데이터의 수는 약 770만개입니다."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 상위 5개 데이터 조회"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sql 사용하여 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 7:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+\n",
      "|                 _id|          _timestamp|                data|\n",
      "+--------------------+--------------------+--------------------+\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "+--------------------+--------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 2.38 ms, sys: 7.63 ms, total: 10 ms\n",
      "Wall time: 2.14 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 데이터 확인\n",
    "\n",
    "query = \"\"\"\n",
    "  SELECT *\n",
    "  FROM tweets\n",
    "\"\"\"\n",
    "yarn.sql(query).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Dataframe 함수를 사용하여 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 8:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+\n",
      "|                 _id|          _timestamp|                data|\n",
      "+--------------------+--------------------+--------------------+\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "|{64102551db4069ef...|2023-03-14T16:42:...|{2023-03-14T07:42...|\n",
      "+--------------------+--------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 3.26 ms, sys: 3.04 ms, total: 6.3 ms\n",
      "Wall time: 1.79 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(_id=Row(oid='64102551db4069eff534f0c2'), _timestamp='2023-03-14T16:42:09.451647+09:00', data=Row(created_at='2023-03-14T07:42:03.000Z', edit_history_tweet_ids=['1635546820548993024'], id='1635546820548993024', lang='fr', text='RT @Mweilc: Après @AmiensMetropole fin de parcours des tableaux de la Chambre du Roi enfin visibles à hauteur d’homme  @CVersailles avec en…', withheld=None)),\n",
       " Row(_id=Row(oid='64102551db4069eff534f0c3'), _timestamp='2023-03-14T16:42:09.474800+09:00', data=Row(created_at='2023-03-14T07:42:03.000Z', edit_history_tweet_ids=['1635546820523728897'], id='1635546820523728897', lang='en', text='@erdincciftci Good morning forever Erdin 🙏🔥', withheld=None)),\n",
       " Row(_id=Row(oid='64102551db4069eff534f0c4'), _timestamp='2023-03-14T16:42:09.476270+09:00', data=Row(created_at='2023-03-14T07:42:03.000Z', edit_history_tweet_ids=['1635546820540788736'], id='1635546820540788736', lang='en', text='@CheapremeBrand Am i right https://t.co/34FUg5Ed7e', withheld=None)),\n",
       " Row(_id=Row(oid='64102551db4069eff534f0c5'), _timestamp='2023-03-14T16:42:09.476917+09:00', data=Row(created_at='2023-03-14T07:42:03.000Z', edit_history_tweet_ids=['1635546820528029698'], id='1635546820528029698', lang='lt', text='@homerhsimpson Kad smo mazge', withheld=None)),\n",
       " Row(_id=Row(oid='64102551db4069eff534f0c6'), _timestamp='2023-03-14T16:42:09.477979+09:00', data=Row(created_at='2023-03-14T07:42:03.000Z', edit_history_tweet_ids=['1635546820549177345'], id='1635546820549177345', lang='en', text='RT @Joon_record: Goodmorning Fam 🌸🍬🍫 https://t.co/Sh1digrFFs', withheld=None))]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.types.Row"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df.head(5)[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`df.head(5)`를 통하여 조회할 수 있지만 `df.head(5)`는 로우 오브젝트를 반환합니다."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 언어별 count 확인"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sql 사용하여 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 11:======================================================> (42 + 1) / 43]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "|lang|  count|\n",
      "+----+-------+\n",
      "|  en|2197064|\n",
      "|  ja|1708222|\n",
      "|  th| 484936|\n",
      "|  es| 467246|\n",
      "|  ko| 316200|\n",
      "+----+-------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 65.1 ms, sys: 103 ms, total: 168 ms\n",
      "Wall time: 35 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#언어별 count 확인\n",
    "\n",
    "query = '''\n",
    "  SELECT data.lang as lang, count(*) count\n",
    "  FROM tweets\n",
    "  WHERE data.lang is not NULL\n",
    "  GROUP BY lang\n",
    "  ORDER BY count DESC;\n",
    "'''\n",
    "\n",
    "yarn.sql(query).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Dataframe 함수를 사용하여 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 14:======================================================> (42 + 1) / 43]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "|lang|  count|\n",
      "+----+-------+\n",
      "| art|  15009|\n",
      "|  en|2197064|\n",
      "|  ne|   2113|\n",
      "|  vi|   8691|\n",
      "|  ps|   1232|\n",
      "+----+-------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 77.2 ms, sys: 82.6 ms, total: 160 ms\n",
      "Wall time: 31.7 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df.groupby('data.lang').count().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 영어로 작성된 tweet만 추출"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sql 사용"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+\n",
      "|               time|                text|\n",
      "+-------------------+--------------------+\n",
      "|2023-03-14 16:42:09|@erdincciftci Goo...|\n",
      "|2023-03-14 16:42:09|@CheapremeBrand A...|\n",
      "|2023-03-14 16:42:09|RT @Joon_record: ...|\n",
      "|2023-03-14 16:42:09|RT @TheRavenCrypt...|\n",
      "|2023-03-14 16:42:09|@gareth_lane He i...|\n",
      "+-------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 0 ns, sys: 7.5 ms, total: 7.5 ms\n",
      "Wall time: 597 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#영어로 작성된 tweet만 추출\n",
    "\n",
    "query = \"\"\"\n",
    "  SELECT\n",
    "    DATE_FORMAT(_timestamp, \"yyyy-MM-dd HH:mm:ss\") as time,\n",
    "    data.text as text\n",
    "  FROM tweets\n",
    "  WHERE data.lang = 'en';\n",
    "\"\"\"\n",
    "\n",
    "en_tweets_sql = yarn.sql(query)\n",
    "en_tweets_sql.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### DataFrame 함수 사용"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+\n",
      "|               time|                text|\n",
      "+-------------------+--------------------+\n",
      "|2023-03-14 16:42:09|RT @Mweilc: Après...|\n",
      "|2023-03-14 16:42:09|@erdincciftci Goo...|\n",
      "|2023-03-14 16:42:09|@CheapremeBrand A...|\n",
      "|2023-03-14 16:42:09|@homerhsimpson Ka...|\n",
      "|2023-03-14 16:42:09|RT @Joon_record: ...|\n",
      "+-------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 8.81 ms, sys: 3.51 ms, total: 12.3 ms\n",
      "Wall time: 428 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "from pyspark.sql.functions import date_format\n",
    "en_tweets_sdf = df[['_timestamp', 'data.text']].withColumnRenamed('_timestamp','time')\n",
    "en_tweets_sdf = en_tweets_sdf.select(date_format('time', 'yyyy-MM-dd HH:mm:ss').alias('time'), en_tweets_sdf.text)\n",
    "en_tweets_sdf.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 샘플 추출"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 6.61 ms, sys: 1.34 ms, total: 7.95 ms\n",
      "Wall time: 585 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(time='2023-03-14 16:42:09', text='@erdincciftci Good morning forever Erdin 🙏🔥')]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "#샘플하나 추출\n",
    "en_tweets_sql.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 7.38 ms, total: 7.38 ms\n",
      "Wall time: 434 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(time='2023-03-14 16:42:09', text='RT @Mweilc: Après @AmiensMetropole fin de parcours des tableaux de la Chambre du Roi enfin visibles à hauteur d’homme  @CVersailles avec en…')]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "en_tweets_sdf.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### tweet의 text를 단어별로 split하는 함수\n",
    "- spark의 함수를 사용하여 진행"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 2 µs, total: 2 µs\n",
      "Wall time: 4.05 µs\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#tweet의 text를 단어별로 split하는 함수\n",
    "def text_split(row):\n",
    "  for word in row.text.split():\n",
    "    yield Row(time=row.time, word=word)\n",
    "  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### en_tweets_sql "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 21:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 7.28 ms, sys: 12.4 ms, total: 19.7 ms\n",
      "Wall time: 2.08 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(time='2023-03-14 16:42:09', word='@erdincciftci'),\n",
       " Row(time='2023-03-14 16:42:09', word='Good'),\n",
       " Row(time='2023-03-14 16:42:09', word='morning'),\n",
       " Row(time='2023-03-14 16:42:09', word='forever'),\n",
       " Row(time='2023-03-14 16:42:09', word='Erdin')]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "#단어별로 split한뒤에 5개 show\n",
    "en_tweets_sql.rdd.flatMap(text_split).take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------------+\n",
      "|               time|         word|\n",
      "+-------------------+-------------+\n",
      "|2023-03-14 16:42:09|@erdincciftci|\n",
      "|2023-03-14 16:42:09|         Good|\n",
      "|2023-03-14 16:42:09|      morning|\n",
      "|2023-03-14 16:42:09|      forever|\n",
      "|2023-03-14 16:42:09|        Erdin|\n",
      "+-------------------+-------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 184 ms, sys: 138 ms, total: 322 ms\n",
      "Wall time: 2.22 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#단어별로 split한뒤에 데이터프레임으로 변경후 5개 show\n",
    "en_tweets_sql.rdd.flatMap(text_split).toDF().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### en_tweets_sdf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 10.5 ms, sys: 263 µs, total: 10.8 ms\n",
      "Wall time: 964 ms\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 24:>                                                         (0 + 1) / 1]\r",
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(time='2023-03-14 16:42:09', word='RT'),\n",
       " Row(time='2023-03-14 16:42:09', word='@Mweilc:'),\n",
       " Row(time='2023-03-14 16:42:09', word='Après'),\n",
       " Row(time='2023-03-14 16:42:09', word='@AmiensMetropole'),\n",
       " Row(time='2023-03-14 16:42:09', word='fin')]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "en_tweets_sdf.rdd.flatMap(text_split).take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------------+\n",
      "|               time|            word|\n",
      "+-------------------+----------------+\n",
      "|2023-03-14 16:42:09|              RT|\n",
      "|2023-03-14 16:42:09|        @Mweilc:|\n",
      "|2023-03-14 16:42:09|           Après|\n",
      "|2023-03-14 16:42:09|@AmiensMetropole|\n",
      "|2023-03-14 16:42:09|             fin|\n",
      "+-------------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 2.28 ms, sys: 27.7 ms, total: 30 ms\n",
      "Wall time: 1.07 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#단어별로 split한뒤에 데이터프레임으로 변경후 5개 show\n",
    "en_tweets_sdf.rdd.flatMap(text_split).toDF().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 단어별로 분리하고 데이터프레임으로 추출"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sql 사용"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 27:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 8.35 ms, sys: 17 ms, total: 25.3 ms\n",
      "Wall time: 1.36 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#단어별로 분리하고 데이터프레임으로 추출\n",
    "words_sql = en_tweets_sql.rdd.flatMap(text_split).toDF()\n",
    "words_sql.createOrReplaceTempView('words_sql')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------------+\n",
      "|               time|         word|\n",
      "+-------------------+-------------+\n",
      "|2023-03-14 16:42:09|@erdincciftci|\n",
      "|2023-03-14 16:42:09|         Good|\n",
      "|2023-03-14 16:42:09|      morning|\n",
      "|2023-03-14 16:42:09|      forever|\n",
      "|2023-03-14 16:42:09|        Erdin|\n",
      "+-------------------+-------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 3.55 ms, sys: 3.52 ms, total: 7.06 ms\n",
      "Wall time: 507 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#words의 상위 5개 show\n",
    "query = '''\n",
    "    SELECT *\n",
    "    FROM words_sql\n",
    "'''\n",
    "\n",
    "yarn.sql(query).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 29:======================================================> (42 + 1) / 43]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|34827421|\n",
      "+--------+\n",
      "\n",
      "CPU times: user 97 ms, sys: 118 ms, total: 215 ms\n",
      "Wall time: 1min 1s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#개수 확인\n",
    "query = \"\"\"\n",
    "    SELECT count(*) \n",
    "    FROM words_sql\n",
    "\"\"\"\n",
    "\n",
    "yarn.sql(query).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### DataFrame 함수 사용"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.18 ms, sys: 22.4 ms, total: 24.6 ms\n",
      "Wall time: 468 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#단어별로 분리하고 데이터프레임으로 추출\n",
    "words_sdf = en_tweets_sdf.rdd.flatMap(text_split).toDF()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------------+\n",
      "|               time|            word|\n",
      "+-------------------+----------------+\n",
      "|2023-03-14 16:42:09|              RT|\n",
      "|2023-03-14 16:42:09|        @Mweilc:|\n",
      "|2023-03-14 16:42:09|           Après|\n",
      "|2023-03-14 16:42:09|@AmiensMetropole|\n",
      "|2023-03-14 16:42:09|             fin|\n",
      "+-------------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 4.01 ms, sys: 4.15 ms, total: 8.16 ms\n",
      "Wall time: 508 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#words_sdf 상위 5개 show\n",
    "words_sdf.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1 µs, sys: 0 ns, total: 1 µs\n",
      "Wall time: 3.58 µs\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#개수 확인\n",
    "#words_sdf.count()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "개수확인할때 Error 발생"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 물리적인 테이블로 보관\n",
    "- hdfs의 spark-warehour에 저장"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rmr: DEPRECATED: Please use '-rm -r' instead.\n",
      "Deleted spark-warehouse/yarn_twitter_sample_words\n"
     ]
    }
   ],
   "source": [
    "!hdfs dfs -rmr spark-warehouse/yarn_twitter_sample_words"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "!hdfs dfs -expunge"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "기존폴더 제거"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ls: `spark-warehouse/yarn_twitter_sample_words': No such file or directory\r\n"
     ]
    }
   ],
   "source": [
    "!hdfs dfs -ls spark-warehouse/yarn_twitter_sample_words"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 99.2 ms, sys: 134 ms, total: 234 ms\n",
      "Wall time: 1min 12s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#words를 hdfs warehouse에 저장\n",
    "words_sql.write.saveAsTable('yarn_twitter_sample_words')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`saveAsTable()`함수를 사용하여 `spark-warehouse`에 저장"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-rw-r--r--   3 bigdata supergroup          0 2023-03-28 00:05 spark-warehouse/yarn_twitter_sample_words/_SUCCESS\n"
     ]
    }
   ],
   "source": [
    "!hdfs dfs -ls spark-warehouse/yarn_twitter_sample_words | grep _SUCCESS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "hdfs를 확인해보니 잘 저장되었습니다."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 36:===========================================>              (3 + 1) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 5.98 ms, sys: 5.91 ms, total: 11.9 ms\n",
      "Wall time: 1.52 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "34827421"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "#hdfs warehouse에 저장된 테이블 개수파악\n",
    "yarn.table('yarn_twitter_sample_words').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "물리적인 테이블의 형태로 보관하니 집계 시간이 빨라짐"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 단어별로 카운트"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "기존의 word_sql"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 41:>                                                         (0 + 4) / 5]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "|word|  count|\n",
      "+----+-------+\n",
      "|  RT|1232866|\n",
      "| the| 770781|\n",
      "|  to| 601721|\n",
      "|   a| 435703|\n",
      "| and| 416036|\n",
      "+----+-------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 81.8 ms, sys: 121 ms, total: 202 ms\n",
      "Wall time: 1min 4s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "query = '''\n",
    "    SELECT word, count(*) count\n",
    "    FROM words_sql\n",
    "    GROUP BY word\n",
    "    ORDER BY count DESC\n",
    "'''\n",
    "\n",
    "yarn.sql(query).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "hdfs에 저장되어있는 물리적인 테이블"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 44:===========================================>              (3 + 1) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "|word|  count|\n",
      "+----+-------+\n",
      "|  RT|1232866|\n",
      "| the| 770781|\n",
      "|  to| 601721|\n",
      "|   a| 435703|\n",
      "| and| 416036|\n",
      "+----+-------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 10.9 ms, sys: 10.9 ms, total: 21.8 ms\n",
      "Wall time: 7.77 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "query = '''\n",
    "    SELECT word, count(*) count\n",
    "    FROM yarn_twitter_sample_words\n",
    "    GROUP BY word\n",
    "    ORDER BY count DESC\n",
    "'''\n",
    "\n",
    "yarn.sql(query).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "물리적인 테이블에서 조회할때 훨씬 빠른속도를 보임"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 데이터 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------+\n",
      "|               time|   word|\n",
      "+-------------------+-------+\n",
      "|2023-03-17 22:58:36|Another|\n",
      "|2023-03-17 22:58:36|fucking|\n",
      "|2023-03-17 22:58:36|    day|\n",
      "|2023-03-17 22:58:36|     😕|\n",
      "|2023-03-17 22:58:36|   Was.|\n",
      "+-------------------+-------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 2.38 ms, sys: 2.22 ms, total: 4.6 ms\n",
      "Wall time: 381 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "yarn.sql('SELECT * FROM yarn_twitter_sample_words').show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1시간마다 그룹화하여 데이터 집계"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 48:=============================================>            (7 + 2) / 9]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 17.8 ms, sys: 28.4 ms, total: 46.2 ms\n",
      "Wall time: 23.2 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 48:===================================================>      (8 + 1) / 9]\r",
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "28603658"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "query = '''\n",
    " SELECT substr(time, 1, 13) time,\n",
    " word,\n",
    " count(*) count\n",
    " FROM yarn_twitter_sample_words\n",
    " GROUP BY time, word\n",
    "''' \n",
    "yarn.sql(query).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 등록 횟수가 적은 단어 조사"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 54:===========================================>              (3 + 1) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------+\n",
      "|count|  words|\n",
      "+-----+-------+\n",
      "|    1|2259387|\n",
      "|    2| 321173|\n",
      "|    3| 134057|\n",
      "|    4|  75277|\n",
      "|    5|  48716|\n",
      "+-----+-------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 4.39 ms, sys: 14.6 ms, total: 19 ms\n",
      "Wall time: 7.39 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#등록 횟수가 적은 단어의 수 조사\n",
    "\n",
    "query = '''\n",
    "    SELECT t.count, count(*) words\n",
    "    FROM (\n",
    "    SELECT word, count(*) count FROM yarn_twitter_sample_words\n",
    "    GROUP BY word\n",
    "    ) t\n",
    "    GROUP BY 1\n",
    "    ORDER BY 1\n",
    "'''\n",
    "\n",
    "yarn.sql(query).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 단어를 카테고리로 나누는 디멘전 테이블 조회"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 58:===========================================>              (3 + 1) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+---------+\n",
      "|                word|count| category|\n",
      "+--------------------+-----+---------+\n",
      "|https://t.co/JqnL...|    2|  COUNT=2|\n",
      "|https://t.co/sYzG...|    1|  COUNT=1|\n",
      "|               those|13352|    those|\n",
      "|          distancing|   54| COUNT=54|\n",
      "|               still|24013|    still|\n",
      "|      @lawyersarab17|    2|  COUNT=2|\n",
      "|           @4KTFeed:|    4|  COUNT=4|\n",
      "|                some|28064|     some|\n",
      "|                hope|10093|     hope|\n",
      "|            Colluded|  125|COUNT=125|\n",
      "+--------------------+-----+---------+\n",
      "only showing top 10 rows\n",
      "\n",
      "CPU times: user 8.32 ms, sys: 12.8 ms, total: 21.1 ms\n",
      "Wall time: 6.91 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 60:>                                                         (0 + 1) / 1]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 단어를 카테고리로 나누는 디멘전 테이블\n",
    "\n",
    "query = '''\n",
    " SELECT word, count,\n",
    "     IF(count > 1000, word, concat('COUNT=', count)) category\n",
    " FROM (\n",
    "     SELECT word, count(*) count FROM yarn_twitter_sample_words\n",
    "     GROUP BY 1\n",
    " ) t\n",
    "'''\n",
    "\n",
    "yarn.sql(query).show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 디멘전 테이블을 view: word_category 생성"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 4.48 ms, sys: 0 ns, total: 4.48 ms\n",
      "Wall time: 44.6 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "yarn.sql(query).createOrReplaceTempView('word_category')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 63:==============================================>           (4 + 1) / 5]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 5 ms, sys: 18.8 ms, total: 23.8 ms\n",
      "Wall time: 6.21 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "3100631"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "yarn.sql(query).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 1시간마다 카테고리별로 그룹화"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 71:===================================================>      (8 + 1) / 9]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------+-----+\n",
      "|         time|category|count|\n",
      "+-------------+--------+-----+\n",
      "|2023-03-20 23| COUNT=8| 4700|\n",
      "|2023-03-20 22|COUNT=20|  781|\n",
      "|2023-03-21 16| COUNT=3| 8350|\n",
      "|2023-03-16 00|COUNT=29| 1609|\n",
      "|2023-03-16 19|COUNT=26| 1282|\n",
      "+-------------+--------+-----+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 34.6 ms, sys: 29.9 ms, total: 64.5 ms\n",
      "Wall time: 27.4 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# hdfs에 저장되어있는 테이블\n",
    "# 1시간마다 카테고리별로 그룹화하여 집계\n",
    "query = '''\n",
    " SELECT substr(a.time, 1, 13) time, b.category, count(*) count\n",
    " FROM yarn_twitter_sample_words a\n",
    " LEFT JOIN word_category b ON a.word = b.word\n",
    " GROUP BY 1, 2\n",
    "'''\n",
    "\n",
    "yarn.sql(query).show(5)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 개수 확인"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 80:===================================================>      (8 + 1) / 9]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 17.6 ms, sys: 73.5 ms, total: 91 ms\n",
      "Wall time: 25.1 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "220243"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "yarn.sql(query).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### pandas의 DataFrame으로 변환"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 710 ms, sys: 374 ms, total: 1.08 s\n",
      "Wall time: 27.1 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>time</th>\n",
       "      <th>category</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2023-03-22 00</td>\n",
       "      <td>COUNT=85</td>\n",
       "      <td>71</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2023-03-21 16</td>\n",
       "      <td>COUNT=3</td>\n",
       "      <td>8350</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2023-03-17 14</td>\n",
       "      <td>COUNT=146</td>\n",
       "      <td>222</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2023-03-17 21</td>\n",
       "      <td>COUNT=40</td>\n",
       "      <td>996</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2023-03-20 23</td>\n",
       "      <td>COUNT=8</td>\n",
       "      <td>4700</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "            time   category  count\n",
       "0  2023-03-22 00   COUNT=85     71\n",
       "1  2023-03-21 16    COUNT=3   8350\n",
       "2  2023-03-17 14  COUNT=146    222\n",
       "3  2023-03-17 21   COUNT=40    996\n",
       "4  2023-03-20 23    COUNT=8   4700"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "result = yarn.sql(query).toPandas()\n",
    "result.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 64 ms, sys: 5.13 ms, total: 69.1 ms\n",
      "Wall time: 66 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "time        220243\n",
       "category    220243\n",
       "count       220243\n",
       "dtype: int64"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "result.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}