Pyspark Keyword Clustering.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.ml.feature import CountVectorizer\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.ml.linalg import SparseVector, Vectors, VectorUDT\n",
    "from pyspark.sql.types import ArrayType, IntegerType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Keyword clustering\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'3.1.1'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv('./search_results.csv', header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.drop('Title', '_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|             Keyword|                 Url|\n",
      "+--------------------+--------------------+\n",
      "|sump pump next to...|https://www.mlive...|\n",
      "|sump pump next to...|https://www.famil...|\n",
      "|sump pump next to...|https://diy.stack...|\n",
      "+--------------------+--------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.groupBy('Keyword') \\\n",
    "  .agg(F.collect_list('url').alias('urlIndex'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "cv = CountVectorizer(inputCol='urlIndex', outputCol='urlVec')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "transformed_df = cv.fit(df).transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "27665"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "transformed_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2 = transformed_df.drop('urlIndex')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|             Keyword|              urlVec|\n",
      "+--------------------+--------------------+\n",
      "|1/4 vs 1/3 sump pump|(34430,[7,17,42,6...|\n",
      "|2 stage dry sump ...|(34430,[1913,2573...|\n",
      "|3 phase sump pump...|(34430,[2079,3983...|\n",
      "|6-cia-ml 1/3 hp s...|(34430,[1405,1678...|\n",
      "|add battery to ex...|(34430,[16,74,111...|\n",
      "|add battery to su...|(34430,[16,25,41,...|\n",
      "|adjusting zoeller...|(34430,[151,159,4...|\n",
      "|average sump pump...|(34430,[7,62,63,6...|\n",
      "|battery backup su...|(34430,[16,25,41,...|\n",
      "|best sump pump fo...|(34430,[7,17,120,...|\n",
      "|best way to drain...|(34430,[2,14,23,5...|\n",
      "|big john automati...|(34430,[2836,3618...|\n",
      "|black friday sump...|(34430,[36,120,15...|\n",
      "|bur cam 1/3 hp su...|(34430,[4407,7850...|\n",
      "|campbell hausfeld...|(34430,[694,1343,...|\n",
      "|can you connect s...|(34430,[88,105,17...|\n",
      "|changing out sump...|(34430,[4,20,21,2...|\n",
      "|clogged sump pump...|(34430,[0,119,344...|\n",
      "|coleman effluent ...|(34430,[300,928,1...|\n",
      "|consumer report b...|(34430,[18,201,28...|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "get_indices = udf(lambda v: [i.item() for i in v.indices], ArrayType(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "df3 = df2.withColumn(\"urlIndices\", get_indices('urlVec'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "df3 = df3.drop('urlVec')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|             Keyword|          urlIndices|\n",
      "+--------------------+--------------------+\n",
      "|1/4 vs 1/3 sump pump|[7, 17, 42, 64, 8...|\n",
      "|2 stage dry sump ...|[1913, 2573, 3316...|\n",
      "|3 phase sump pump...|[2079, 3983, 8216...|\n",
      "|6-cia-ml 1/3 hp s...|[1405, 1678, 5892...|\n",
      "|add battery to ex...|[16, 74, 111, 117...|\n",
      "|add battery to su...|[16, 25, 41, 74, ...|\n",
      "|adjusting zoeller...|[151, 159, 436, 5...|\n",
      "|average sump pump...|[7, 62, 63, 64, 8...|\n",
      "|battery backup su...|[16, 25, 41, 74, ...|\n",
      "|best sump pump fo...|[7, 17, 120, 246,...|\n",
      "|best way to drain...|[2, 14, 23, 50, 1...|\n",
      "|big john automati...|[2836, 3618, 4872...|\n",
      "|black friday sump...|[36, 120, 1566, 6...|\n",
      "|bur cam 1/3 hp su...|[4407, 7850, 8039...|\n",
      "|campbell hausfeld...|[694, 1343, 2428,...|\n",
      "|can you connect s...|[88, 105, 172, 18...|\n",
      "|changing out sump...|[4, 20, 21, 29, 5...|\n",
      "|clogged sump pump...|[0, 119, 344, 575...|\n",
      "|coleman effluent ...|[300, 928, 1632, ...|\n",
      "|consumer report b...|[18, 201, 289, 10...|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df3.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [],
   "source": [
    "def find_url_intersections(df1, df2):\n",
    "    self_join = df1.alias('Key1').join(df2.alias('Key2')).where('Key1.Keyword < Key2.Keyword')\\\n",
    "        .withColumn(\"urlIntersect\", F.array_intersect('Key1.urlIndices','Key2.urlIndices'))\\\n",
    "        .where(F.size(F.col('urlIntersect'))>0)\\\n",
    "        .drop('urlIndices')\n",
    "    return self_join.toDF(\"Keyword1\", \"Keyword2\", \"urlIntersect\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [],
   "source": [
    "self_join = find_url_intersections(df3, df3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "40000"
      ]
     },
     "execution_count": 89,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "self_join.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [],
   "source": [
    "self_join = self_join.repartition(8*3)\n",
    "#self_join = self_join.coalesce(8*3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.63 s, sys: 1.29 s, total: 3.92 s\n",
      "Wall time: 26min 14s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "15139127"
      ]
     },
     "execution_count": 91,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "self_join.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+\n",
      "|            Keyword1|            Keyword2|        urlIntersect|\n",
      "+--------------------+--------------------+--------------------+\n",
      "|1/4 vs 1/3 sump pump|average sump pump...|         [7, 64, 84]|\n",
      "|1/4 vs 1/3 sump pump|best sump pump fo...|             [7, 17]|\n",
      "|1/4 vs 1/3 sump pump|how big should a ...|                [84]|\n",
      "|1/4 vs 1/3 sump pump|how do i know whi...|             [7, 42]|\n",
      "|1/4 vs 1/3 sump pump|how to determine ...|         [7, 64, 84]|\n",
      "|1/4 vs 1/3 sump pump|  low watt sump pump|             [7, 17]|\n",
      "|1/4 vs 1/3 sump pump| pipe size sump pump|                [84]|\n",
      "|1/4 vs 1/3 sump pump|zoeller vs hydrom...|            [17, 42]|\n",
      "|add battery to ex...|add battery to su...|[16, 74, 111, 117...|\n",
      "|add battery to ex...|battery backup su...|            [16, 74]|\n",
      "|add battery to ex...|how to hook up a ...| [16, 111, 117, 403]|\n",
      "|add battery to ex...|sump pump battery...|            [16, 74]|\n",
      "|add battery to ex...|sump pump battery...|                [16]|\n",
      "|add battery to su...|battery backup su...|    [16, 25, 41, 74]|\n",
      "|add battery to su...|how to hook up a ...|      [16, 111, 117]|\n",
      "|add battery to su...|  low watt sump pump|                [25]|\n",
      "|add battery to su...|sump pump 12v bat...|                [25]|\n",
      "|add battery to su...|sump pump battery...|        [16, 41, 74]|\n",
      "|add battery to su...|sump pump battery...|                [16]|\n",
      "|adjusting zoeller...|how to adjust flo...|          [151, 159]|\n",
      "+--------------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "self_join.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "tmp = self_join.limit(8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+------------+\n",
      "|            Keyword1|            Keyword2|urlIntersect|\n",
      "+--------------------+--------------------+------------+\n",
      "|1/4 vs 1/3 sump pump|average sump pump...| [7, 64, 84]|\n",
      "|1/4 vs 1/3 sump pump|best sump pump fo...|     [7, 17]|\n",
      "|1/4 vs 1/3 sump pump|how big should a ...|        [84]|\n",
      "|1/4 vs 1/3 sump pump|how do i know whi...|     [7, 42]|\n",
      "|1/4 vs 1/3 sump pump|how to determine ...| [7, 64, 84]|\n",
      "|1/4 vs 1/3 sump pump|  low watt sump pump|     [7, 17]|\n",
      "|1/4 vs 1/3 sump pump| pipe size sump pump|        [84]|\n",
      "|1/4 vs 1/3 sump pump|zoeller vs hydrom...|    [17, 42]|\n",
      "+--------------------+--------------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "tmp.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "self_join = tmp.alias('Key1').join(tmp.drop('Keyword1').alias('Key2')).where('Key1.Keyword2 < Key2.Keyword2')\\\n",
    "    .withColumn(\"urlIntersect3\", F.array_intersect('Key1.urlIntersect','Key2.urlIntersect'))\\\n",
    "    .where(F.size(F.col('urlIntersect3'))>0)\\\n",
    "    .drop('urlIntersect')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+-------------+\n",
      "|            Keyword1|            Keyword2|            Keyword2|urlIntersect3|\n",
      "+--------------------+--------------------+--------------------+-------------+\n",
      "|1/4 vs 1/3 sump pump|average sump pump...|best sump pump fo...|          [7]|\n",
      "|1/4 vs 1/3 sump pump|average sump pump...|how big should a ...|         [84]|\n",
      "|1/4 vs 1/3 sump pump|average sump pump...|how do i know whi...|          [7]|\n",
      "|1/4 vs 1/3 sump pump|average sump pump...|how to determine ...|  [7, 64, 84]|\n",
      "|1/4 vs 1/3 sump pump|average sump pump...|  low watt sump pump|          [7]|\n",
      "|1/4 vs 1/3 sump pump|average sump pump...| pipe size sump pump|         [84]|\n",
      "|1/4 vs 1/3 sump pump|best sump pump fo...|how do i know whi...|          [7]|\n",
      "|1/4 vs 1/3 sump pump|best sump pump fo...|how to determine ...|          [7]|\n",
      "|1/4 vs 1/3 sump pump|best sump pump fo...|  low watt sump pump|      [7, 17]|\n",
      "|1/4 vs 1/3 sump pump|best sump pump fo...|zoeller vs hydrom...|         [17]|\n",
      "|1/4 vs 1/3 sump pump|how big should a ...|how to determine ...|         [84]|\n",
      "|1/4 vs 1/3 sump pump|how big should a ...| pipe size sump pump|         [84]|\n",
      "|1/4 vs 1/3 sump pump|how do i know whi...|how to determine ...|          [7]|\n",
      "|1/4 vs 1/3 sump pump|how do i know whi...|  low watt sump pump|          [7]|\n",
      "|1/4 vs 1/3 sump pump|how do i know whi...|zoeller vs hydrom...|         [42]|\n",
      "|1/4 vs 1/3 sump pump|how to determine ...|  low watt sump pump|          [7]|\n",
      "|1/4 vs 1/3 sump pump|how to determine ...| pipe size sump pump|         [84]|\n",
      "|1/4 vs 1/3 sump pump|  low watt sump pump|zoeller vs hydrom...|         [17]|\n",
      "+--------------------+--------------------+--------------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "self_join.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "18"
      ]
     },
     "execution_count": 79,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "self_join.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "return self_join.toDF(\"Keyword1\", \"Keyword2\", \"urlIntersect\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "cannot resolve '`Key1.Keyword`' given input columns: [Key1.Keyword1, Key2.Keyword1, Key1.Keyword2, Key2.Keyword2, Key1.urlIntersect, Key2.urlIntersect]; line 1 pos 0;\n'Filter NOT ('Key1.Keyword = 'Key2.Keyword)\n+- Join Inner\n   :- SubqueryAlias Key1\n   :  +- GlobalLimit 8\n   :     +- LocalLimit 8\n   :        +- Repartition 24, false\n   :           +- Project [Keyword#16 AS Keyword1#1111, Keyword#1091 AS Keyword2#1112, urlIntersect#1102 AS urlIntersect#1113]\n   :              +- Project [Keyword#16, Keyword#1091, urlIntersect#1102]\n   :                 +- Filter (size(urlIntersect#1102, true) > 0)\n   :                    +- Project [Keyword#16, urlIndices#958, Keyword#1091, urlIndices#1090, array_intersect(urlIndices#958, urlIndices#1090) AS urlIntersect#1102]\n   :                       +- Filter NOT (Keyword#16 = Keyword#1091)\n   :                          +- Join Inner\n   :                             :- SubqueryAlias Key1\n   :                             :  +- Project [Keyword#16, urlIndices#958]\n   :                             :     +- Project [Keyword#16, urlVec#890, <lambda>(urlVec#890) AS urlIndices#958]\n   :                             :        +- Project [Keyword#16, urlVec#890]\n   :                             :           +- Project [Keyword#16, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n   :                             :              +- Aggregate [Keyword#16], [Keyword#16, collect_list(url#18, 0, 0) AS urlIndex#44]\n   :                             :                 +- Project [Keyword#16, Url#18]\n   :                             :                    +- Relation[Keyword#16,Title#17,Url#18,_id#19] csv\n   :                             +- SubqueryAlias Key2\n   :                                +- Project [Keyword#1091, urlIndices#1090]\n   :                                   +- Project [Keyword#1091, urlVec#890, <lambda>(urlVec#890) AS urlIndices#1090]\n   :                                      +- Project [Keyword#1091, urlVec#890]\n   :                                         +- Project [Keyword#1091, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n   :                                            +- Aggregate [Keyword#1091], [Keyword#1091, collect_list(url#1093, 0, 0) AS urlIndex#44]\n   :                                               +- Project [Keyword#1091, Url#1093]\n   :                                                  +- Relation[Keyword#1091,Title#1092,Url#1093,_id#1094] csv\n   +- SubqueryAlias Key2\n      +- GlobalLimit 8\n         +- LocalLimit 8\n            +- Repartition 24, false\n               +- Project [Keyword#16 AS Keyword1#1492, Keyword#1091 AS Keyword2#1493, urlIntersect#1102 AS urlIntersect#1494]\n                  +- Project [Keyword#16, Keyword#1091, urlIntersect#1102]\n                     +- Filter (size(urlIntersect#1102, true) > 0)\n                        +- Project [Keyword#16, urlIndices#958, Keyword#1091, urlIndices#1090, array_intersect(urlIndices#958, urlIndices#1090) AS urlIntersect#1102]\n                           +- Filter NOT (Keyword#16 = Keyword#1091)\n                              +- Join Inner\n                                 :- SubqueryAlias Key1\n                                 :  +- Project [Keyword#16, urlIndices#958]\n                                 :     +- Project [Keyword#16, urlVec#890, <lambda>(urlVec#890) AS urlIndices#958]\n                                 :        +- Project [Keyword#16, urlVec#890]\n                                 :           +- Project [Keyword#16, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n                                 :              +- Aggregate [Keyword#16], [Keyword#16, collect_list(url#18, 0, 0) AS urlIndex#44]\n                                 :                 +- Project [Keyword#16, Url#18]\n                                 :                    +- Relation[Keyword#16,Title#17,Url#18,_id#19] csv\n                                 +- SubqueryAlias Key2\n                                    +- Project [Keyword#1091, urlIndices#1090]\n                                       +- Project [Keyword#1091, urlVec#890, <lambda>(urlVec#890) AS urlIndices#1090]\n                                          +- Project [Keyword#1091, urlVec#890]\n                                             +- Project [Keyword#1091, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n                                                +- Aggregate [Keyword#1091], [Keyword#1091, collect_list(url#1093, 0, 0) AS urlIndex#44]\n                                                   +- Project [Keyword#1091, Url#1093]\n                                                      +- Relation[Keyword#1091,Title#1092,Url#1093,_id#1094] csv\n",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-38-55c0c33c8957>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mfind_url_intersections\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtmp\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0mtmp\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m<ipython-input-29-cb11217d8f55>\u001b[0m in \u001b[0;36mfind_url_intersections\u001b[0;34m(df1, df2)\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32mdef\u001b[0m \u001b[0mfind_url_intersections\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdf1\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdf2\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m     \u001b[0mself_join\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mdf1\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0malias\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'Key1'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjoin\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdf2\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0malias\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'Key2'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwhere\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'Key1.Keyword != Key2.Keyword'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      3\u001b[0m         \u001b[0;34m.\u001b[0m\u001b[0mwithColumn\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"urlIntersect\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mF\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0marray_intersect\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'Key1.urlIndices'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m'Key2.urlIndices'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      4\u001b[0m         \u001b[0;34m.\u001b[0m\u001b[0mwhere\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mF\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msize\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mF\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcol\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'urlIntersect'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m>\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      5\u001b[0m         \u001b[0;34m.\u001b[0m\u001b[0mdrop\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'urlIndices'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/lib/python3.7/site-packages/pyspark/sql/dataframe.py\u001b[0m in \u001b[0;36mfilter\u001b[0;34m(self, condition)\u001b[0m\n\u001b[1;32m   1713\u001b[0m         \"\"\"\n\u001b[1;32m   1714\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcondition\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstr\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1715\u001b[0;31m             \u001b[0mjdf\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jdf\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcondition\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1716\u001b[0m         \u001b[0;32melif\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcondition\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mColumn\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1717\u001b[0m             \u001b[0mjdf\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jdf\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfilter\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcondition\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jc\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/lib/python3.7/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1303\u001b[0m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1304\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1305\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1306\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1307\u001b[0m         \u001b[0;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/lib/python3.7/site-packages/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    115\u001b[0m                 \u001b[0;31m# Hide where the exception came from that shows a non-Pythonic\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    116\u001b[0m                 \u001b[0;31m# JVM exception message.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 117\u001b[0;31m                 \u001b[0;32mraise\u001b[0m \u001b[0mconverted\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    118\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    119\u001b[0m                 \u001b[0;32mraise\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mAnalysisException\u001b[0m: cannot resolve '`Key1.Keyword`' given input columns: [Key1.Keyword1, Key2.Keyword1, Key1.Keyword2, Key2.Keyword2, Key1.urlIntersect, Key2.urlIntersect]; line 1 pos 0;\n'Filter NOT ('Key1.Keyword = 'Key2.Keyword)\n+- Join Inner\n   :- SubqueryAlias Key1\n   :  +- GlobalLimit 8\n   :     +- LocalLimit 8\n   :        +- Repartition 24, false\n   :           +- Project [Keyword#16 AS Keyword1#1111, Keyword#1091 AS Keyword2#1112, urlIntersect#1102 AS urlIntersect#1113]\n   :              +- Project [Keyword#16, Keyword#1091, urlIntersect#1102]\n   :                 +- Filter (size(urlIntersect#1102, true) > 0)\n   :                    +- Project [Keyword#16, urlIndices#958, Keyword#1091, urlIndices#1090, array_intersect(urlIndices#958, urlIndices#1090) AS urlIntersect#1102]\n   :                       +- Filter NOT (Keyword#16 = Keyword#1091)\n   :                          +- Join Inner\n   :                             :- SubqueryAlias Key1\n   :                             :  +- Project [Keyword#16, urlIndices#958]\n   :                             :     +- Project [Keyword#16, urlVec#890, <lambda>(urlVec#890) AS urlIndices#958]\n   :                             :        +- Project [Keyword#16, urlVec#890]\n   :                             :           +- Project [Keyword#16, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n   :                             :              +- Aggregate [Keyword#16], [Keyword#16, collect_list(url#18, 0, 0) AS urlIndex#44]\n   :                             :                 +- Project [Keyword#16, Url#18]\n   :                             :                    +- Relation[Keyword#16,Title#17,Url#18,_id#19] csv\n   :                             +- SubqueryAlias Key2\n   :                                +- Project [Keyword#1091, urlIndices#1090]\n   :                                   +- Project [Keyword#1091, urlVec#890, <lambda>(urlVec#890) AS urlIndices#1090]\n   :                                      +- Project [Keyword#1091, urlVec#890]\n   :                                         +- Project [Keyword#1091, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n   :                                            +- Aggregate [Keyword#1091], [Keyword#1091, collect_list(url#1093, 0, 0) AS urlIndex#44]\n   :                                               +- Project [Keyword#1091, Url#1093]\n   :                                                  +- Relation[Keyword#1091,Title#1092,Url#1093,_id#1094] csv\n   +- SubqueryAlias Key2\n      +- GlobalLimit 8\n         +- LocalLimit 8\n            +- Repartition 24, false\n               +- Project [Keyword#16 AS Keyword1#1492, Keyword#1091 AS Keyword2#1493, urlIntersect#1102 AS urlIntersect#1494]\n                  +- Project [Keyword#16, Keyword#1091, urlIntersect#1102]\n                     +- Filter (size(urlIntersect#1102, true) > 0)\n                        +- Project [Keyword#16, urlIndices#958, Keyword#1091, urlIndices#1090, array_intersect(urlIndices#958, urlIndices#1090) AS urlIntersect#1102]\n                           +- Filter NOT (Keyword#16 = Keyword#1091)\n                              +- Join Inner\n                                 :- SubqueryAlias Key1\n                                 :  +- Project [Keyword#16, urlIndices#958]\n                                 :     +- Project [Keyword#16, urlVec#890, <lambda>(urlVec#890) AS urlIndices#958]\n                                 :        +- Project [Keyword#16, urlVec#890]\n                                 :           +- Project [Keyword#16, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n                                 :              +- Aggregate [Keyword#16], [Keyword#16, collect_list(url#18, 0, 0) AS urlIndex#44]\n                                 :                 +- Project [Keyword#16, Url#18]\n                                 :                    +- Relation[Keyword#16,Title#17,Url#18,_id#19] csv\n                                 +- SubqueryAlias Key2\n                                    +- Project [Keyword#1091, urlIndices#1090]\n                                       +- Project [Keyword#1091, urlVec#890, <lambda>(urlVec#890) AS urlIndices#1090]\n                                          +- Project [Keyword#1091, urlVec#890]\n                                             +- Project [Keyword#1091, urlIndex#44, UDF(urlIndex#44) AS urlVec#890]\n                                                +- Aggregate [Keyword#1091], [Keyword#1091, collect_list(url#1093, 0, 0) AS urlIndex#44]\n                                                   +- Project [Keyword#1091, Url#1093]\n                                                      +- Relation[Keyword#1091,Title#1092,Url#1093,_id#1094] csv\n"
     ]
    }
   ],
   "source": [
    "find_url_intersections(tmp,tmp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "ename": "Py4JJavaError",
     "evalue": "An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 4 in stage 23.0 failed 1 times, most recent failure: Lost task 4.0 in stage 23.0 (TID 900) (192.168.0.142 executor driver): org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire 65536 bytes of memory, got 0\n\tat org.apache.spark.memory.MemoryConsumer.throwOom(MemoryConsumer.java:157)\n\tat org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:97)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.<init>(UnsafeInMemorySorter.java:128)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.<init>(UnsafeExternalSorter.java:164)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.create(UnsafeExternalSorter.java:131)\n\tat org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:91)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationMap.dumpToExternalSorter(ObjectAggregationMap.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.processInputs(ObjectAggregationIterator.scala:177)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.<init>(ObjectAggregationIterator.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2(ObjectHashAggregateExec.scala:107)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2$adapted(ObjectHashAggregateExec.scala:85)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2(RDD.scala:885)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2$adapted(RDD.scala:885)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.sql.execution.joins.UnsafeCartesianRDD.compute(CartesianProductExec.scala:47)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.CoalescedRDD.$anonfun$compute$1(CoalescedRDD.scala:99)\n\tat scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)\n\tat scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:490)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.agg_doAggregateWithKeys_0$(Unknown Source)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.processNext(Unknown Source)\n\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:755)\n\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:132)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:131)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:834)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2253)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2202)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2201)\n\tat scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)\n\tat scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)\n\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2201)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1078)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1078)\n\tat scala.Option.foreach(Option.scala:407)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1078)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2440)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2382)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2371)\n\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:868)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2202)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2223)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2242)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2267)\n\tat org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1030)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:414)\n\tat org.apache.spark.rdd.RDD.collect(RDD.scala:1029)\n\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:180)\n\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:566)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n\tat py4j.Gateway.invoke(Gateway.java:282)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n\tat java.base/java.lang.Thread.run(Thread.java:834)\nCaused by: org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire 65536 bytes of memory, got 0\n\tat org.apache.spark.memory.MemoryConsumer.throwOom(MemoryConsumer.java:157)\n\tat org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:97)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.<init>(UnsafeInMemorySorter.java:128)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.<init>(UnsafeExternalSorter.java:164)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.create(UnsafeExternalSorter.java:131)\n\tat org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:91)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationMap.dumpToExternalSorter(ObjectAggregationMap.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.processInputs(ObjectAggregationIterator.scala:177)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.<init>(ObjectAggregationIterator.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2(ObjectHashAggregateExec.scala:107)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2$adapted(ObjectHashAggregateExec.scala:85)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2(RDD.scala:885)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2$adapted(RDD.scala:885)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.sql.execution.joins.UnsafeCartesianRDD.compute(CartesianProductExec.scala:47)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.CoalescedRDD.$anonfun$compute$1(CoalescedRDD.scala:99)\n\tat scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)\n\tat scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:490)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.agg_doAggregateWithKeys_0$(Unknown Source)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.processNext(Unknown Source)\n\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:755)\n\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:132)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:131)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\t... 1 more\n",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[0;32m<timed exec>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/lib/python3.7/site-packages/pyspark/rdd.py\u001b[0m in \u001b[0;36mcollect\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    947\u001b[0m         \"\"\"\n\u001b[1;32m    948\u001b[0m         \u001b[0;32mwith\u001b[0m \u001b[0mSCCallSiteSync\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcontext\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0mcss\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 949\u001b[0;31m             \u001b[0msock_info\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mctx\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPythonRDD\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcollectAndServe\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jrdd\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrdd\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    950\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mlist\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0m_load_from_socket\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msock_info\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jrdd_deserializer\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    951\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/lib/python3.7/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1303\u001b[0m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1304\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1305\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1306\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1307\u001b[0m         \u001b[0;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/lib/python3.7/site-packages/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    109\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mdeco\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    110\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 111\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    112\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    113\u001b[0m             \u001b[0mconverted\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mconvert_exception\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0me\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjava_exception\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/lib/python3.7/site-packages/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[0;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[1;32m    326\u001b[0m                 raise Py4JJavaError(\n\u001b[1;32m    327\u001b[0m                     \u001b[0;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 328\u001b[0;31m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[1;32m    329\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    330\u001b[0m                 raise Py4JError(\n",
      "\u001b[0;31mPy4JJavaError\u001b[0m: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 4 in stage 23.0 failed 1 times, most recent failure: Lost task 4.0 in stage 23.0 (TID 900) (192.168.0.142 executor driver): org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire 65536 bytes of memory, got 0\n\tat org.apache.spark.memory.MemoryConsumer.throwOom(MemoryConsumer.java:157)\n\tat org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:97)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.<init>(UnsafeInMemorySorter.java:128)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.<init>(UnsafeExternalSorter.java:164)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.create(UnsafeExternalSorter.java:131)\n\tat org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:91)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationMap.dumpToExternalSorter(ObjectAggregationMap.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.processInputs(ObjectAggregationIterator.scala:177)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.<init>(ObjectAggregationIterator.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2(ObjectHashAggregateExec.scala:107)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2$adapted(ObjectHashAggregateExec.scala:85)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2(RDD.scala:885)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2$adapted(RDD.scala:885)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.sql.execution.joins.UnsafeCartesianRDD.compute(CartesianProductExec.scala:47)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.CoalescedRDD.$anonfun$compute$1(CoalescedRDD.scala:99)\n\tat scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)\n\tat scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:490)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.agg_doAggregateWithKeys_0$(Unknown Source)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.processNext(Unknown Source)\n\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:755)\n\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:132)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:131)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:834)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2253)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2202)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2201)\n\tat scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)\n\tat scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)\n\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2201)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1078)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1078)\n\tat scala.Option.foreach(Option.scala:407)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1078)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2440)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2382)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2371)\n\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:868)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2202)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2223)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2242)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2267)\n\tat org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1030)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:414)\n\tat org.apache.spark.rdd.RDD.collect(RDD.scala:1029)\n\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:180)\n\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:566)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n\tat py4j.Gateway.invoke(Gateway.java:282)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n\tat java.base/java.lang.Thread.run(Thread.java:834)\nCaused by: org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire 65536 bytes of memory, got 0\n\tat org.apache.spark.memory.MemoryConsumer.throwOom(MemoryConsumer.java:157)\n\tat org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:97)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.<init>(UnsafeInMemorySorter.java:128)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.<init>(UnsafeExternalSorter.java:164)\n\tat org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.create(UnsafeExternalSorter.java:131)\n\tat org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:91)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationMap.dumpToExternalSorter(ObjectAggregationMap.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.processInputs(ObjectAggregationIterator.scala:177)\n\tat org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.<init>(ObjectAggregationIterator.scala:77)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2(ObjectHashAggregateExec.scala:107)\n\tat org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec.$anonfun$doExecute$2$adapted(ObjectHashAggregateExec.scala:85)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2(RDD.scala:885)\n\tat org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2$adapted(RDD.scala:885)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.sql.execution.joins.UnsafeCartesianRDD.compute(CartesianProductExec.scala:47)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:337)\n\tat org.apache.spark.rdd.CoalescedRDD.$anonfun$compute$1(CoalescedRDD.scala:99)\n\tat scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)\n\tat scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:490)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.agg_doAggregateWithKeys_0$(Unknown Source)\n\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.processNext(Unknown Source)\n\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:755)\n\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:132)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:131)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\t... 1 more\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "two_keyword_clusters = self_join.select(\"Keyword1\").distinct().rdd.flatMap(lambda x: x).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "two_keyword_clusters = self_join.select(\"Keyword1\").distinct().rdd.flatMap(lambda x: x).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Find a way to group by Keyword1, and apply find_url_intersection to each group"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "self_join = self_join.alias('Key1').join(df3.alias('Key2')).where('Key1.Keyword1 != Key1.Keyword2 != Key2.Keyword')\n",
    "self_join = self_join.withColumn(\"urlIntersect\", F.array_intersect('Key1.urlIntersect','Key2.urlIndices'))\\\n",
    ".where(F.size(F.col('urlIntersect'))>0)\\\n",
    ".drop('urlIndices')\n",
    "#self_join = self_join.toDF(\"Keyword1\", \"Keyword2\", \"urlIntersect\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "self_join.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "self_join = self_join.alias('IntersectKey').join(df3.alias('Key3')).where('IntersectKey.Keyword != Key3.Keyword')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "self_join = self_join.alias('Key1Key2').join(df3.alias('Key3')).where('Key1Key2.Key1.Keyword != Key3.Keyword')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [],
   "source": [
    "df4 = df3.select(df3.Keyword, F.explode(df3.urlIndices))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|             Keyword|  col|\n",
      "+--------------------+-----+\n",
      "|1/4 vs 1/3 sump pump|    7|\n",
      "|1/4 vs 1/3 sump pump|   17|\n",
      "|1/4 vs 1/3 sump pump|   42|\n",
      "|1/4 vs 1/3 sump pump|   64|\n",
      "|1/4 vs 1/3 sump pump|   84|\n",
      "|1/4 vs 1/3 sump pump|  339|\n",
      "|1/4 vs 1/3 sump pump|  704|\n",
      "|1/4 vs 1/3 sump pump| 1578|\n",
      "|1/4 vs 1/3 sump pump| 2130|\n",
      "|1/4 vs 1/3 sump pump| 2143|\n",
      "|2 stage dry sump ...| 1913|\n",
      "|2 stage dry sump ...| 2573|\n",
      "|2 stage dry sump ...| 3316|\n",
      "|2 stage dry sump ...| 5447|\n",
      "|2 stage dry sump ...|10975|\n",
      "|2 stage dry sump ...|13771|\n",
      "|2 stage dry sump ...|16509|\n",
      "|2 stage dry sump ...|28180|\n",
      "|2 stage dry sump ...|28958|\n",
      "|3 phase sump pump...| 2079|\n",
      "+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df4.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df5 = df4.groupBy(\"col\").agg(F.collect_list(\"Keyword\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "df5.show(5, truncate=100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df5.join(df5).join(df5).show(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "recursive_self_join = df5\n",
    "for i in range(1):\n",
    "    alias1 = str(i) + '_1'\n",
    "    alias2 = str(i) + '_2'\n",
    "    print(alias1)\n",
    "    print(alias2)\n",
    "    recursive_self_join = recursive_self_join.alias(alias1).join(df5.alias(alias2)).where(alias1+'.col != '+alias2+'.col')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "recursive_self_join.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "alias1 = '0_1'\n",
    "alias2 = '0_2'\n",
    "new_alias1 = '1_1'\n",
    "new_alias2 = '1_2'\n",
    "recursive_self_join = recursive_self_join.alias(new_alias1).join(df5.alias(new_alias2)).where(alias1+'.col != '+alias2+'.col')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tmp = df5.alias(alias1).join(df5.alias(alias2)).where(alias1+'.col != '+alias2+'.col')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pairs = df2.alias(\"Keyword1\").join(df2.alias(\"Keyword2\")).where('Keyword1.Keyword != Keyword2.Keyword')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df6 = df4.join(df5, on='col')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df6.show(200, truncate = 100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# number of urls two keywords have to share to be grouped together\n",
    "X = 3\n",
    "# minimum number of keywords per group\n",
    "Y = 4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@udf(VectorUDT())\n",
    "def intersect(v1, v2):\n",
    "    # Sparse vector will become dense\n",
    "    assert isinstance(v1, SparseVector) and isinstance(v2, SparseVector)\n",
    "    # Compute union of indices\n",
    "    indices = set(v1.indices).intersection(set(v2.indices))\n",
    "    return Vectors.sparse(34430, dict(zip(indices,[1.0]*len(indices))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pairs = df2.alias(\"Keyword1\").join(df2.alias(\"Keyword2\")).where('Keyword1.Keyword != Keyword2.Keyword')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pairs.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df3 = pairs.withColumn('urlIntersect', intersect('Keyword1.urlVec', 'Keyword2.urlVec'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df3 = df3.drop('urlVec')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df3.show(300)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}