pg_randomforest_remote.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "Py4JJavaError",
     "evalue": "An error occurred while calling None.org.apache.spark.api.java.JavaSparkContext.\n: java.lang.NullPointerException\r\n\tat org.apache.spark.SparkContext.<init>(SparkContext.scala:560)\r\n\tat org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58)\r\n\tat sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)\r\n\tat sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)\r\n\tat java.lang.reflect.Constructor.newInstance(Constructor.java:423)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:238)\r\n\tat py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)\r\n\tat py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Thread.java:748)\r\n",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-1-9e528dcaea6f>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m     32\u001b[0m         \u001b[1;33m.\u001b[0m\u001b[0mbuilder\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     33\u001b[0m         \u001b[1;33m.\u001b[0m\u001b[0mappName\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"RandomForestRegressorExample2\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 34\u001b[1;33m         \u001b[1;33m.\u001b[0m\u001b[0mmaster\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"spark://10.20.0.223:7077\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     35\u001b[0m         \u001b[1;33m.\u001b[0m\u001b[0mgetOrCreate\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     36\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\ProgramData\\Anaconda3\\lib\\site-packages\\pyspark\\sql\\session.py\u001b[0m in \u001b[0;36mgetOrCreate\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    171\u001b[0m                     \u001b[1;32mfor\u001b[0m \u001b[0mkey\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mvalue\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_options\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mitems\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    172\u001b[0m                         \u001b[0msparkConf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mset\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mkey\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mvalue\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 173\u001b[1;33m                     \u001b[0msc\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mSparkContext\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgetOrCreate\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0msparkConf\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    174\u001b[0m                     \u001b[1;31m# This SparkContext may be an existing one.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    175\u001b[0m                     \u001b[1;32mfor\u001b[0m \u001b[0mkey\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mvalue\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_options\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mitems\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\ProgramData\\Anaconda3\\lib\\site-packages\\pyspark\\context.py\u001b[0m in \u001b[0;36mgetOrCreate\u001b[1;34m(cls, conf)\u001b[0m\n\u001b[0;32m    365\u001b[0m         \u001b[1;32mwith\u001b[0m \u001b[0mSparkContext\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_lock\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    366\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0mSparkContext\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_active_spark_context\u001b[0m \u001b[1;32mis\u001b[0m \u001b[1;32mNone\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 367\u001b[1;33m                 \u001b[0mSparkContext\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mconf\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0mconf\u001b[0m \u001b[1;32mor\u001b[0m \u001b[0mSparkConf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    368\u001b[0m             \u001b[1;32mreturn\u001b[0m \u001b[0mSparkContext\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_active_spark_context\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    369\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\ProgramData\\Anaconda3\\lib\\site-packages\\pyspark\\context.py\u001b[0m in \u001b[0;36m__init__\u001b[1;34m(self, master, appName, sparkHome, pyFiles, environment, batchSize, serializer, conf, gateway, jsc, profiler_cls)\u001b[0m\n\u001b[0;32m    134\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    135\u001b[0m             self._do_init(master, appName, sparkHome, pyFiles, environment, batchSize, serializer,\n\u001b[1;32m--> 136\u001b[1;33m                           conf, jsc, profiler_cls)\n\u001b[0m\u001b[0;32m    137\u001b[0m         \u001b[1;32mexcept\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    138\u001b[0m             \u001b[1;31m# If an error occurs, clean up in order to allow future SparkContext creation:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\ProgramData\\Anaconda3\\lib\\site-packages\\pyspark\\context.py\u001b[0m in \u001b[0;36m_do_init\u001b[1;34m(self, master, appName, sparkHome, pyFiles, environment, batchSize, serializer, conf, jsc, profiler_cls)\u001b[0m\n\u001b[0;32m    196\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    197\u001b[0m         \u001b[1;31m# Create the Java SparkContext through Py4J\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 198\u001b[1;33m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jsc\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mjsc\u001b[0m \u001b[1;32mor\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_initialize_context\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_conf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jconf\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    199\u001b[0m         \u001b[1;31m# Reset the SparkConf to the one actually used by the SparkContext in JVM.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    200\u001b[0m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_conf\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mSparkConf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0m_jconf\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jsc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mconf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\ProgramData\\Anaconda3\\lib\\site-packages\\pyspark\\context.py\u001b[0m in \u001b[0;36m_initialize_context\u001b[1;34m(self, jconf)\u001b[0m\n\u001b[0;32m    304\u001b[0m         \u001b[0mInitialize\u001b[0m \u001b[0mSparkContext\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mfunction\u001b[0m \u001b[0mto\u001b[0m \u001b[0mallow\u001b[0m \u001b[0msubclass\u001b[0m \u001b[0mspecific\u001b[0m \u001b[0minitialization\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    305\u001b[0m         \"\"\"\n\u001b[1;32m--> 306\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mJavaSparkContext\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mjconf\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    307\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    308\u001b[0m     \u001b[1;33m@\u001b[0m\u001b[0mclassmethod\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\ProgramData\\Anaconda3\\lib\\site-packages\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1523\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_gateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1524\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1525\u001b[1;33m             answer, self._gateway_client, None, self._fqn)\n\u001b[0m\u001b[0;32m   1526\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1527\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\ProgramData\\Anaconda3\\lib\\site-packages\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    326\u001b[0m                 raise Py4JJavaError(\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    330\u001b[0m                 raise Py4JError(\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling None.org.apache.spark.api.java.JavaSparkContext.\n: java.lang.NullPointerException\r\n\tat org.apache.spark.SparkContext.<init>(SparkContext.scala:560)\r\n\tat org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58)\r\n\tat sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)\r\n\tat sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)\r\n\tat java.lang.reflect.Constructor.newInstance(Constructor.java:423)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:238)\r\n\tat py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)\r\n\tat py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Thread.java:748)\r\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Licensed to the Apache Software Foundation (ASF) under one or more\n",
    "# contributor license agreements.  See the NOTICE file distributed with\n",
    "# this work for additional information regarding copyright ownership.\n",
    "# The ASF licenses this file to You under the Apache License, Version 2.0\n",
    "# (the \"License\"); you may not use this file except in compliance with\n",
    "# the License.  You may obtain a copy of the License at\n",
    "#\n",
    "#    http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "#\n",
    "\n",
    "\"\"\"\n",
    "Random Forest Regressor Example - modified for distributed cluster.\n",
    "\"\"\"\n",
    "from __future__ import print_function\n",
    "\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.regression import RandomForestRegressor\n",
    "from pyspark.ml.feature import VectorIndexer\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "if __name__ == \"__main__\":    \n",
    "    spark = SparkSession\\\n",
    "        .builder\\\n",
    "        .appName(\"RandomForestRegressorExample2\")\\\n",
    "        .master(\"spark://10.20.0.223:7077\")\\\n",
    "        .getOrCreate()\n",
    "    \n",
    "    data = spark.read.format(\"libsvm\").load(\"file:////share//sample_libsvm_data.txt\")\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "    # Automatically identify categorical features, and index them.\n",
    "    # Set maxCategories so features with > 4 distinct values are treated as continuous.\n",
    "    featureIndexer =\\\n",
    "        VectorIndexer(inputCol=\"features\", outputCol=\"indexedFeatures\", maxCategories=4).fit(data)\n",
    "\n",
    "    # Split the data into training and test sets (30% held out for testing)\n",
    "    (trainingData, testData) = data.randomSplit([0.7, 0.3])\n",
    "\n",
    "    # Train a RandomForest model.\n",
    "    rf = RandomForestRegressor(featuresCol=\"indexedFeatures\")\n",
    "\n",
    "    # Chain indexer and forest in a Pipeline\n",
    "    pipeline = Pipeline(stages=[featureIndexer, rf])\n",
    "\n",
    "    # Train model.  This also runs the indexer.\n",
    "    model = pipeline.fit(trainingData)\n",
    "\n",
    "    # Make predictions.\n",
    "    predictions = model.transform(testData)\n",
    "\n",
    "    # Select example rows to display.\n",
    "    predictions.select(\"prediction\", \"label\", \"features\").show(5)\n",
    "\n",
    "    # Select (prediction, true label) and compute test error\n",
    "    evaluator = RegressionEvaluator(\n",
    "        labelCol=\"label\", predictionCol=\"prediction\", metricName=\"rmse\")\n",
    "    rmse = evaluator.evaluate(predictions)\n",
    "    print(\"Root Mean Squared Error (RMSE) on test data = %g\" % rmse)\n",
    "\n",
    "    rfModel = model.stages[1]\n",
    "    print(rfModel)  # summary only\n",
    "    # $example off$\n",
    "\n",
    "    spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "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.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}