10_graphframes.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "7503c9a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# add imports for python and spark and configure environment\n",
    "\n",
    "from graphframes import *\n",
    "from pyspark import *\n",
    "from pyspark.sql import *\n",
    "\n",
    "import pandas as pd\n",
    "import os\n",
    "import sys\n",
    "import pyarrow as pa\n",
    "import pyarrow.parquet as pq\n",
    "os.environ['PYSPARK_PYTHON'] = sys.executable\n",
    "os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = \"\"\"--name job_name --master local --conf spark.dynamicAllocation.enabled=true pyspark-shell\"\"\"\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages graphframes:graphframes:0.8.1-spark3.0-s_2.12 pyspark-shell'\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages graphframes:graphframes:0.8.1-spark3.0-s_2.12,neo4j-contrib:neo4j-connector-apache-spark_2.12:4.0.1_for_spark_3 pyspark-shell'\n",
    "\n",
    "spark = SparkSession.builder.appName(\"DataFrame\").getOrCreate()\n",
    "conf = SparkConf().setMaster(\"local\").setAppName(\"Assignment 10\")\n",
    "sc = SparkContext.getOrCreate(conf=conf)\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "2cbd3280",
   "metadata": {},
   "outputs": [],
   "source": [
    "# read bike data and create dataframes for stations and trip data\n",
    "bikeStations = spark.read.option(\"header\",\"true\")\\\n",
    ".csv(\"201508_station_data.csv\")\n",
    "tripData = spark.read.option(\"header\",\"true\")\\\n",
    ".csv(\"201508_trip_data.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "8ef1fe51",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- station_id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- lat: string (nullable = true)\n",
      " |-- long: string (nullable = true)\n",
      " |-- dockcount: string (nullable = true)\n",
      " |-- landmark: string (nullable = true)\n",
      " |-- installation: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#print schema from bikeStations\n",
    "bikeStations.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "5e81a5dd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Trip ID: string (nullable = true)\n",
      " |-- Duration: string (nullable = true)\n",
      " |-- Start Date: string (nullable = true)\n",
      " |-- Start Station: string (nullable = true)\n",
      " |-- Start Terminal: string (nullable = true)\n",
      " |-- End Date: string (nullable = true)\n",
      " |-- End Station: string (nullable = true)\n",
      " |-- End Terminal: string (nullable = true)\n",
      " |-- Bike #: string (nullable = true)\n",
      " |-- Subscriber Type: string (nullable = true)\n",
      " |-- Zip Code: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#print schema from tripData\n",
    "tripData.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "dd460c0d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# rename columns to create graphframes\n",
    "stationVertices = bikeStations.withColumnRenamed(\"name\", \"id\").distinct()\n",
    "tripEdges = tripData\\\n",
    ".withColumnRenamed(\"Start Station\", \"src\")\\\n",
    ".withColumnRenamed(\"End Station\", \"dst\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "d1dd3585",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/16 15:31:43 WARN CacheManager: Asked to cache already cached data.\n",
      "22/04/16 15:31:43 WARN CacheManager: Asked to cache already cached data.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "GraphFrame(v:[id: string, station_id: string ... 5 more fields], e:[src: string, dst: string ... 9 more fields])"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# create station graphframe\n",
    "from graphframes import GraphFrame\n",
    "stationGraph = GraphFrame(stationVertices, tripEdges)\n",
    "stationGraph.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "e7d8e734",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[key: string, value: string]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# import expr func and add configuration for timestamps\n",
    "from pyspark.sql.functions import expr\n",
    "spark.sql(\"set spark.sql.legacy.timeParserPolicy=LEGACY\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "cf64b51c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create ring of 2 length\n",
    "motifs1 = stationGraph.find(\"(a)-[ab]->(b); (b)-[ba]->(a)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "b63dbd3a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 9:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-----------------+---------------+---------------+\n",
      "|id               |id               |Start Date     |End Date       |\n",
      "+-----------------+-----------------+---------------+---------------+\n",
      "|Townsend at 7th  |2nd at South Park|7/15/2015 16:57|7/15/2015 17:12|\n",
      "|Market at Sansome|2nd at South Park|6/22/2015 17:12|6/22/2015 17:26|\n",
      "+-----------------+-----------------+---------------+---------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# ring of 2 length query\n",
    "motifs1.selectExpr(\"*\",\n",
    "\"to_timestamp(ab.`Start Date`, 'MM/dd/yyyy HH:mm') as abStart\",\n",
    "\"to_timestamp(ba.`Start Date`, 'MM/dd/yyyy HH:mm') as baStart\")\\\n",
    ".where(\"ab.`Bike #` = ba.`Bike #`\")\\\n",
    ".where(\"a.id != b.id\")\\\n",
    ".where(\"abStart < baStart\")\\\n",
    ".orderBy(expr(\"cast(baStart as long) - cast(abStart as long)\"))\\\n",
    ".selectExpr(\"a.id\", \"b.id\", \"ab.`Start Date`\", \"ba.`End Date`\")\\\n",
    ".limit(2).show(2, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "0088e6bb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create ring of 3 length\n",
    "motifs2 = stationGraph.find(\"(a)-[ab]->(b); (b)-[bc]->(c); (c)-[ca]->(a)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "0c5a479c",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR:root:Exception while sending command.                         (0 + 1) / 4]\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/f_dev/.local/lib/python3.9/site-packages/py4j/clientserver.py\", line 480, in send_command\n",
      "    raise Py4JNetworkError(\"Answer from Java side is empty\")\n",
      "py4j.protocol.Py4JNetworkError: Answer from Java side is empty\n",
      "\n",
      "During handling of the above exception, another exception occurred:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/f_dev/.local/lib/python3.9/site-packages/py4j/java_gateway.py\", line 1038, in send_command\n",
      "    response = connection.send_command(command)\n",
      "  File \"/home/f_dev/.local/lib/python3.9/site-packages/py4j/clientserver.py\", line 503, in send_command\n",
      "    raise Py4JNetworkError(\n",
      "py4j.protocol.Py4JNetworkError: Error while sending or receiving\n"
     ]
    },
    {
     "ename": "Py4JError",
     "evalue": "An error occurred while calling o96.showString",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mPy4JError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m/tmp/ipykernel_5147/2710495116.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m motifs2.selectExpr(\"*\",\n\u001b[0m\u001b[1;32m      2\u001b[0m \u001b[0;34m\"to_timestamp(ab.`Start Date`, 'MM/dd/yyyy HH:mm') as abStart\"\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0;34m\"to_timestamp(bc.`Start Date`, 'MM/dd/yyyy HH:mm') as bcStart\"\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      4\u001b[0m \"to_timestamp(ca.`Start Date`, 'MM/dd/yyyy HH:mm') as caStart\")\\\n\u001b[1;32m      5\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0mwhere\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"ca.`Bike #` = bc.`Bike #`\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwhere\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"ab.`Bike #` = bc.`Bike #`\"\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[0;32m~/.local/lib/python3.9/site-packages/pyspark/sql/dataframe.py\u001b[0m in \u001b[0;36mshow\u001b[0;34m(self, n, truncate, vertical)\u001b[0m\n\u001b[1;32m    500\u001b[0m                     \"Parameter 'truncate={}' should be either bool or int.\".format(truncate))\n\u001b[1;32m    501\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 502\u001b[0;31m             \u001b[0mprint\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[0mshowString\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mn\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mint_truncate\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mvertical\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    503\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    504\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m__repr__\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\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~/.local/lib/python3.9/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1319\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1320\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[0;32m-> 1321\u001b[0;31m         return_value = get_return_value(\n\u001b[0m\u001b[1;32m   1322\u001b[0m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[1;32m   1323\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.9/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~/.local/lib/python3.9/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    332\u001b[0m                     format(target_id, \".\", name, value))\n\u001b[1;32m    333\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[0;32m--> 334\u001b[0;31m             raise Py4JError(\n\u001b[0m\u001b[1;32m    335\u001b[0m                 \u001b[0;34m\"An error occurred while calling {0}{1}{2}\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    336\u001b[0m                 format(target_id, \".\", name))\n",
      "\u001b[0;31mPy4JError\u001b[0m: An error occurred while calling o96.showString"
     ]
    }
   ],
   "source": [
    "# ring of 3 length query\n",
    "motifs2.selectExpr(\"*\",\n",
    "\"to_timestamp(ab.`Start Date`, 'MM/dd/yyyy HH:mm') as abStart\",\n",
    "\"to_timestamp(bc.`Start Date`, 'MM/dd/yyyy HH:mm') as bcStart\",\n",
    "\"to_timestamp(ca.`Start Date`, 'MM/dd/yyyy HH:mm') as caStart\")\\\n",
    ".where(\"ca.`Bike #` = bc.`Bike #`\").where(\"ab.`Bike #` = bc.`Bike #`\")\\\n",
    ".where(\"a.id != b.id\").where(\"b.id != c.id\")\\\n",
    ".where(\"abStart < bcStart\").where(\"bcStart < caStart\")\\\n",
    ".orderBy(expr(\"cast(caStart as long) - cast(abStart as long)\"))\\\n",
    ".selectExpr(\"a.id\", \"b.id\", \"c.id\", \"ab.`Start Date`\", \"ca.`End Date`\")\\\n",
    ".limit(3).show(3, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "2d97b413",
   "metadata": {},
   "outputs": [
    {
     "ename": "ConnectionRefusedError",
     "evalue": "[Errno 111] Connection refused",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mConnectionRefusedError\u001b[0m                    Traceback (most recent call last)",
      "\u001b[0;32m/tmp/ipykernel_5147/4183373879.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mmotifs3\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mstationGraph\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfind\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cb]->(b); (b)-[ba]->(a)\"\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~/.local/lib/python3.9/site-packages/graphframes/graphframe.py\u001b[0m in \u001b[0;36mfind\u001b[0;34m(self, pattern)\u001b[0m\n\u001b[1;32m    194\u001b[0m         \u001b[0;34m:\u001b[0m\u001b[0;32mreturn\u001b[0m\u001b[0;34m:\u001b[0m  \u001b[0mDataFrame\u001b[0m \u001b[0;32mwith\u001b[0m \u001b[0mone\u001b[0m \u001b[0mRow\u001b[0m \u001b[0;32mfor\u001b[0m \u001b[0meach\u001b[0m \u001b[0minstance\u001b[0m \u001b[0mof\u001b[0m \u001b[0mthe\u001b[0m \u001b[0mmotif\u001b[0m \u001b[0mfound\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    195\u001b[0m         \"\"\"\n\u001b[0;32m--> 196\u001b[0;31m         \u001b[0mjdf\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jvm_graph\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfind\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mpattern\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    197\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mjdf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_sqlContext\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    198\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.9/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1318\u001b[0m             \u001b[0mproto\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mEND_COMMAND_PART\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1319\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1320\u001b[0;31m         \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[0m\u001b[1;32m   1321\u001b[0m         return_value = get_return_value(\n\u001b[1;32m   1322\u001b[0m             answer, self.gateway_client, self.target_id, self.name)\n",
      "\u001b[0;32m~/.local/lib/python3.9/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36msend_command\u001b[0;34m(self, command, retry, binary)\u001b[0m\n\u001b[1;32m   1034\u001b[0m          \u001b[0;32mif\u001b[0m\u001b[0;31m \u001b[0m\u001b[0;31m`\u001b[0m\u001b[0mbinary\u001b[0m\u001b[0;31m`\u001b[0m \u001b[0;32mis\u001b[0m\u001b[0;31m \u001b[0m\u001b[0;31m`\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;31m`\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1035\u001b[0m         \"\"\"\n\u001b[0;32m-> 1036\u001b[0;31m         \u001b[0mconnection\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_get_connection\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   1037\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[1;32m   1038\u001b[0m             \u001b[0mresponse\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mconnection\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[0;32m~/.local/lib/python3.9/site-packages/py4j/clientserver.py\u001b[0m in \u001b[0;36m_get_connection\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    279\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    280\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mconnection\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mNone\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0mconnection\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msocket\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 281\u001b[0;31m             \u001b[0mconnection\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_create_new_connection\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    282\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mconnection\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    283\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.9/site-packages/py4j/clientserver.py\u001b[0m in \u001b[0;36m_create_new_connection\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    286\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjava_parameters\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mpython_parameters\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    287\u001b[0m             self.gateway_property, self)\n\u001b[0;32m--> 288\u001b[0;31m         \u001b[0mconnection\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mconnect_to_java_server\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    289\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mset_thread_connection\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mconnection\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    290\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mconnection\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.9/site-packages/py4j/clientserver.py\u001b[0m in \u001b[0;36mconnect_to_java_server\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    400\u001b[0m                 self.socket = self.ssl_context.wrap_socket(\n\u001b[1;32m    401\u001b[0m                     self.socket, server_hostname=self.java_address)\n\u001b[0;32m--> 402\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msocket\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mconnect\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjava_address\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjava_port\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    403\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstream\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msocket\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmakefile\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"rb\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    404\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mis_connected\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mTrue\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mConnectionRefusedError\u001b[0m: [Errno 111] Connection refused"
     ]
    }
   ],
   "source": [
    "# create ring of 4 length\n",
    "motifs3 = stationGraph.find(\"(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cb]->(b); (b)-[ba]->(a)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a74ca4ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ring of 4 length query\n",
    "motifs3.selectExpr(\"*\",\n",
    "\"to_timestamp(ab.`Start Date`, 'MM/dd/yyyy HH:mm') as abStart\",\n",
    "\"to_timestamp(bc.`Start Date`, 'MM/dd/yyyy HH:mm') as bcStart\",\n",
    "\"to_timestamp(cb.`Start Date`, 'MM/dd/yyyy HH:mm') as cbStart\",\n",
    "\"to_timestamp(ba.`Start Date`, 'MM/dd/yyyy HH:mm') as baStart\")\\\n",
    ".where(\"ab.`Bike #` = bc.`Bike #`\").where(\"bc.`Bike #` = cb.`Bike #`\").where(\"cb.`Bike #` = ba.`Bike #`\")\\\n",
    ".where(\"a.id != b.id\").where(\"b.id != c.id\")\\\n",
    ".where(\"abStart < bcStart\").where(\"bcStart < cbStart\").where(\"cbStart < baStart\")\\\n",
    ".orderBy(expr(\"cast(baStart as long) - cast(abStart as long)\"))\\\n",
    ".selectExpr(\"a.id\", \"b.id\", \"c.id\", \"ab.`Start Date`\", \"ba.`End Date`\")\\\n",
    ".limit(4).show(4, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "f25486d5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------+---------------+--------------------+--------------+---------------+--------------------+------------+------+---------------+--------+\n",
      "|Trip ID|Duration|     Start Date|                 src|Start Terminal|       End Date|                 dst|End Terminal|Bike #|Subscriber Type|Zip Code|\n",
      "+-------+--------+---------------+--------------------+--------------+---------------+--------------------+------------+------+---------------+--------+\n",
      "| 913460|     765|8/31/2015 23:26|Harry Bridges Pla...|            50|8/31/2015 23:39|San Francisco Cal...|          70|   288|     Subscriber|    2139|\n",
      "| 913459|    1036|8/31/2015 23:11|San Antonio Shopp...|            31|8/31/2015 23:28|Mountain View Cit...|          27|    35|     Subscriber|   95032|\n",
      "| 913455|     307|8/31/2015 23:13|      Post at Kearny|            47|8/31/2015 23:18|   2nd at South Park|          64|   468|     Subscriber|   94107|\n",
      "| 913454|     409|8/31/2015 23:10|  San Jose City Hall|            10|8/31/2015 23:17| San Salvador at 1st|           8|    68|     Subscriber|   95113|\n",
      "| 913453|     789|8/31/2015 23:09|Embarcadero at Fo...|            51|8/31/2015 23:22|Embarcadero at Sa...|          60|   487|       Customer|    9069|\n",
      "+-------+--------+---------------+--------------------+--------------+---------------+--------------------+------------+------+---------------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#display edges for stations\n",
    "stationGraph.edges.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "078d74b8",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 32:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------+\n",
      "|                  id|inDegree|\n",
      "+--------------------+--------+\n",
      "|San Francisco Cal...|   34810|\n",
      "|San Francisco Cal...|   22523|\n",
      "|Harry Bridges Pla...|   17810|\n",
      "|     2nd at Townsend|   15463|\n",
      "|     Townsend at 7th|   15422|\n",
      "+--------------------+--------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# order stations by largest number of other stations\n",
    "stationGraph.inDegrees.orderBy(\"inDegree\").sort('inDegree', ascending=False).limit(5).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "66cfd3f8",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 35:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+-----+\n",
      "|                 src|                 dst|count|\n",
      "+--------------------+--------------------+-----+\n",
      "|San Francisco Cal...|     Townsend at 7th| 3748|\n",
      "|Harry Bridges Pla...|Embarcadero at Sa...| 3145|\n",
      "|     2nd at Townsend|Harry Bridges Pla...| 2973|\n",
      "|     Townsend at 7th|San Francisco Cal...| 2734|\n",
      "|Harry Bridges Pla...|     2nd at Townsend| 2640|\n",
      "|Embarcadero at Fo...|San Francisco Cal...| 2439|\n",
      "|   Steuart at Market|     2nd at Townsend| 2356|\n",
      "|Embarcadero at Sa...|   Steuart at Market| 2330|\n",
      "|     Townsend at 7th|San Francisco Cal...| 2192|\n",
      "|Temporary Transba...|San Francisco Cal...| 2184|\n",
      "+--------------------+--------------------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# get count for stations groups\n",
    "from pyspark.sql.functions import col, desc, asc\n",
    "stationGraph.edges.groupBy(\"src\", \n",
    "\"dst\").count().orderBy(col(\"count\").desc()).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "1c464ee8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|                 src|count|\n",
      "+--------------------+-----+\n",
      "|San Francisco Cal...|26304|\n",
      "|San Francisco Cal...|21758|\n",
      "|Harry Bridges Pla...|17255|\n",
      "|Temporary Transba...|14436|\n",
      "|Embarcadero at Sa...|14158|\n",
      "|     2nd at Townsend|14026|\n",
      "|     Townsend at 7th|13752|\n",
      "|   Steuart at Market|13687|\n",
      "|      Market at 10th|11885|\n",
      "|   Market at Sansome|11431|\n",
      "|       Market at 4th| 9894|\n",
      "|  Powell Street BART| 9695|\n",
      "|   2nd at South Park| 9469|\n",
      "|     Beale at Market| 8359|\n",
      "|Grant Avenue at C...| 8337|\n",
      "|       2nd at Folsom| 7999|\n",
      "|Civic Center BART...| 7760|\n",
      "|       5th at Howard| 7708|\n",
      "|Broadway St at Ba...| 7676|\n",
      "|Embarcadero at Fo...| 7596|\n",
      "+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "stationGraph.edges.groupBy(\"src\").count().orderBy(col(\"count\").desc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "fdf20954",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----------+--------------------+---------+-----------+---------+-------------+------------+\n",
      "|count|station_id|                  id|      lat|       long|dockcount|     landmark|installation|\n",
      "+-----+----------+--------------------+---------+-----------+---------+-------------+------------+\n",
      "|  496|        51|Embarcadero at Fo...|37.791464|-122.391034|       19|San Francisco|   8/20/2013|\n",
      "|  496|        58|San Francisco Cit...| 37.77865|-122.418235|       19|San Francisco|   8/21/2013|\n",
      "|  496|        60|Embarcadero at Sa...| 37.80477|-122.403234|       15|San Francisco|   8/21/2013|\n",
      "|  496|        65|     Townsend at 7th|37.771058|-122.402717|       15|San Francisco|   8/22/2013|\n",
      "|  496|        63|       Howard at 2nd|37.786978|-122.398108|       19|San Francisco|   8/22/2013|\n",
      "|   39|        33|Rengstorff Avenue...|37.400241|-122.099076|       15|Mountain View|   8/16/2013|\n",
      "|   32|        25|Stanford in Redwo...| 37.48537|-122.203288|       15| Redwood City|   8/12/2013|\n",
      "|  496|        71|Powell at Post (U...|37.788446|-122.408499|       19|San Francisco|   8/23/2013|\n",
      "|  496|        77|   Market at Sansome|37.789625|-122.400811|       27|San Francisco|   8/25/2013|\n",
      "|  105|        84|         Ryland Park|37.342725|-121.895617|       15|     San Jose|    4/9/2014|\n",
      "|  496|        72|Civic Center BART...|37.781039|-122.411748|       23|San Francisco|   8/23/2013|\n",
      "|  105|        12|SJSU 4th at San C...|37.332808|-121.883891|       19|     San Jose|    8/7/2013|\n",
      "|  496|        39|  Powell Street BART|37.783871|-122.408433|       19|San Francisco|   8/25/2013|\n",
      "|  105|        16|SJSU - San Salvad...|37.333955|-121.877349|       15|     San Jose|    8/7/2013|\n",
      "|  496|        62|       2nd at Folsom|37.785299|-122.396236|       19|San Francisco|   8/22/2013|\n",
      "|   51|        36|California Ave Ca...|37.429082|-122.142805|       15|    Palo Alto|   8/14/2013|\n",
      "|  496|        74|   Steuart at Market|37.794139|-122.394434|       23|San Francisco|   8/25/2013|\n",
      "|  105|        80|Santa Clara Count...|37.352601|-121.905733|       15|     San Jose|  12/31/2013|\n",
      "|    0|        47|     Post at Kearney|37.788975|-122.403452|       19|San Francisco|   8/19/2013|\n",
      "|  496|        69|San Francisco Cal...|  37.7766| -122.39547|       23|San Francisco|   8/23/2013|\n",
      "+-----+----------+--------------------+---------+-----------+---------+-------------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# count number of triangles passing through\n",
    "stationGraph.triangleCount().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "b5d990a1",
   "metadata": {},
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'pyspark.tests'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "\u001b[0;32m/tmp/ipykernel_5620/2355964009.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0mgraphframes\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexamples\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mGraphs\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      2\u001b[0m \u001b[0mg\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mGraphs\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msqlContext\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfriends\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m  \u001b[0;31m# Get example graph\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.9/site-packages/graphframes/examples/__init__.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0mbelief_propagation\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mBeliefPropagation\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;32mfrom\u001b[0m \u001b[0;34m.\u001b[0m\u001b[0mgraphs\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mGraphs\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[0m\n\u001b[1;32m      5\u001b[0m \u001b[0m__all__\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m[\u001b[0m\u001b[0;34m'BeliefPropagation'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m'Graphs'\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.9/site-packages/graphframes/examples/belief_propagation.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     20\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyspark\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mSparkConf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mSparkContext\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     21\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msql\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mSQLContext\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfunctions\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0msqlfunctions\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtypes\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 22\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0mpyspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtests\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mQuietTest\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0mSuppressSparkLogs\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     23\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     24\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mgraphframes\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mGraphFrame\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mModuleNotFoundError\u001b[0m: No module named 'pyspark.tests'"
     ]
    }
   ],
   "source": [
    "from graphframes.examples import Graphs\n",
    "g = Graphs(sqlContext).friends()  # Get example graph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "32c2baf5",
   "metadata": {},
   "outputs": [],
   "source": [
    "from neo4j import GraphDatabase\n",
    "\n",
    "neo_conn = GraphDatabase.driver(uri = 'bolt://localhost:7687', auth = ('neo4j', 'test'))\n",
    "neo_session = neo_conn.session()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "08dfb0bb",
   "metadata": {},
   "outputs": [
    {
     "ename": "Py4JJavaError",
     "evalue": "An error occurred while calling o101.load.\n: org.neo4j.driver.exceptions.AuthenticationException: The client is unauthorized due to authentication failure.\n\tat org.neo4j.driver.internal.util.Futures.blockingGet(Futures.java:144)\n\tat org.neo4j.driver.internal.InternalSession.run(InternalSession.java:69)\n\tat org.neo4j.driver.internal.InternalSession.run(InternalSession.java:51)\n\tat org.neo4j.driver.internal.AbstractQueryRunner.run(AbstractQueryRunner.java:37)\n\tat org.neo4j.driver.internal.AbstractQueryRunner.run(AbstractQueryRunner.java:55)\n\tat org.neo4j.spark.util.ValidateConnection.validate(Validations.scala:69)\n\tat org.neo4j.spark.util.Validations$.$anonfun$validate$1(Validations.scala:12)\n\tat org.neo4j.spark.util.Validations$.$anonfun$validate$1$adapted(Validations.scala:12)\n\tat scala.collection.immutable.Set$Set1.foreach(Set.scala:141)\n\tat org.neo4j.spark.util.Validations$.validate(Validations.scala:12)\n\tat org.neo4j.spark.DataSource.inferSchema(DataSource.scala:28)\n\tat org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.getTableFromProvider(DataSourceV2Utils.scala:81)\n\tat org.apache.spark.sql.DataFrameReader.$anonfun$load$1(DataFrameReader.scala:233)\n\tat scala.Option.map(Option.scala:230)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:210)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:174)\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.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)\n\tat py4j.ClientServerConnection.run(ClientServerConnection.java:106)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\n\tSuppressed: org.neo4j.driver.exceptions.ServiceUnavailableException: Connection to the database terminated. Please ensure that your database is listening on the correct host and port and that you have compatible encryption settings both on Neo4j server and driver. Note that the default encryption setting has changed in Neo4j 4.0.\n\t\tat org.neo4j.driver.internal.util.ErrorUtil.newConnectionTerminatedError(ErrorUtil.java:56)\n\t\tat org.neo4j.driver.internal.util.ErrorUtil.newConnectionTerminatedError(ErrorUtil.java:49)\n\t\tat org.neo4j.driver.internal.async.inbound.ChannelErrorHandler.channelInactive(ChannelErrorHandler.java:74)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:392)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:392)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:813)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n\t\t... 1 more\n\tSuppressed: org.neo4j.driver.internal.util.ErrorUtil$InternalExceptionCause\n\t\tat org.neo4j.driver.internal.util.ErrorUtil.newNeo4jError(ErrorUtil.java:76)\n\t\tat org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher.handleFailureMessage(InboundMessageDispatcher.java:122)\n\t\tat org.neo4j.driver.internal.messaging.common.CommonMessageReader.unpackFailureMessage(CommonMessageReader.java:83)\n\t\tat org.neo4j.driver.internal.messaging.common.CommonMessageReader.read(CommonMessageReader.java:59)\n\t\tat org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:83)\n\t\tat org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:35)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)\n\t\tat org.neo4j.driver.internal.async.inbound.MessageDecoder.channelRead(MessageDecoder.java:47)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n\t\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/tmp/ipykernel_5620/3191652624.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mread\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"org.neo4j.spark.DataSource\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m \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      2\u001b[0m   \u001b[0;34m.\u001b[0m\u001b[0moption\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"url\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"bolt://localhost:7687\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m \u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m   \u001b[0;34m.\u001b[0m\u001b[0moption\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"labels\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\":Movie\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;31m \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[0mload\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~/.local/lib/python3.9/site-packages/pyspark/sql/readwriter.py\u001b[0m in \u001b[0;36mload\u001b[0;34m(self, path, format, schema, **options)\u001b[0m\n\u001b[1;32m    162\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_df\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jreader\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mload\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_spark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPythonUtils\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtoSeq\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mpath\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[1;32m    163\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[0;32m--> 164\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_df\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jreader\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mload\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    165\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    166\u001b[0m     def json(self, path, schema=None, primitivesAsString=None, prefersDecimal=None,\n",
      "\u001b[0;32m~/.local/lib/python3.9/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1319\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1320\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[0;32m-> 1321\u001b[0;31m         return_value = get_return_value(\n\u001b[0m\u001b[1;32m   1322\u001b[0m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[1;32m   1323\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.9/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~/.local/lib/python3.9/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    324\u001b[0m             \u001b[0mvalue\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mOUTPUT_CONVERTER\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0mtype\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0manswer\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m2\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mgateway_client\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    325\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0manswer\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m1\u001b[0m\u001b[0;34m]\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0mREFERENCE_TYPE\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 326\u001b[0;31m                 raise Py4JJavaError(\n\u001b[0m\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[1;32m    328\u001b[0m                     format(target_id, \".\", name), value)\n",
      "\u001b[0;31mPy4JJavaError\u001b[0m: An error occurred while calling o101.load.\n: org.neo4j.driver.exceptions.AuthenticationException: The client is unauthorized due to authentication failure.\n\tat org.neo4j.driver.internal.util.Futures.blockingGet(Futures.java:144)\n\tat org.neo4j.driver.internal.InternalSession.run(InternalSession.java:69)\n\tat org.neo4j.driver.internal.InternalSession.run(InternalSession.java:51)\n\tat org.neo4j.driver.internal.AbstractQueryRunner.run(AbstractQueryRunner.java:37)\n\tat org.neo4j.driver.internal.AbstractQueryRunner.run(AbstractQueryRunner.java:55)\n\tat org.neo4j.spark.util.ValidateConnection.validate(Validations.scala:69)\n\tat org.neo4j.spark.util.Validations$.$anonfun$validate$1(Validations.scala:12)\n\tat org.neo4j.spark.util.Validations$.$anonfun$validate$1$adapted(Validations.scala:12)\n\tat scala.collection.immutable.Set$Set1.foreach(Set.scala:141)\n\tat org.neo4j.spark.util.Validations$.validate(Validations.scala:12)\n\tat org.neo4j.spark.DataSource.inferSchema(DataSource.scala:28)\n\tat org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.getTableFromProvider(DataSourceV2Utils.scala:81)\n\tat org.apache.spark.sql.DataFrameReader.$anonfun$load$1(DataFrameReader.scala:233)\n\tat scala.Option.map(Option.scala:230)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:210)\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:174)\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.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)\n\tat py4j.ClientServerConnection.run(ClientServerConnection.java:106)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\n\tSuppressed: org.neo4j.driver.exceptions.ServiceUnavailableException: Connection to the database terminated. Please ensure that your database is listening on the correct host and port and that you have compatible encryption settings both on Neo4j server and driver. Note that the default encryption setting has changed in Neo4j 4.0.\n\t\tat org.neo4j.driver.internal.util.ErrorUtil.newConnectionTerminatedError(ErrorUtil.java:56)\n\t\tat org.neo4j.driver.internal.util.ErrorUtil.newConnectionTerminatedError(ErrorUtil.java:49)\n\t\tat org.neo4j.driver.internal.async.inbound.ChannelErrorHandler.channelInactive(ChannelErrorHandler.java:74)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:392)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:392)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:813)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n\t\t... 1 more\n\tSuppressed: org.neo4j.driver.internal.util.ErrorUtil$InternalExceptionCause\n\t\tat org.neo4j.driver.internal.util.ErrorUtil.newNeo4jError(ErrorUtil.java:76)\n\t\tat org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher.handleFailureMessage(InboundMessageDispatcher.java:122)\n\t\tat org.neo4j.driver.internal.messaging.common.CommonMessageReader.unpackFailureMessage(CommonMessageReader.java:83)\n\t\tat org.neo4j.driver.internal.messaging.common.CommonMessageReader.read(CommonMessageReader.java:59)\n\t\tat org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:83)\n\t\tat org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:35)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)\n\t\tat org.neo4j.driver.internal.async.inbound.MessageDecoder.channelRead(MessageDecoder.java:47)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)\n\t\tat org.neo4j.driver.internal.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n\t\t... 1 more\n"
     ]
    }
   ],
   "source": [
    "# read from Movie db in neo 4j and import into spark\n",
    "spark.read.format(\"org.neo4j.spark.DataSource\") \\\n",
    "  .option(\"url\", \"bolt://localhost:7687\") \\\n",
    "  .option(\"labels\", \":Movie\") \\\n",
    "  .load()"
   ]
  }
 ],
 "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}