Sınavcalisma_9_10_2021_417.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Sınavcalisma.9.10.2021_417.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "authorship_tag": "ABX9TyPG68xRIixsiE4xBQ+TOdCB",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/busranurorhan/Complete-Python-3-Bootcamp/blob/master/S%C4%B1navcalisma_9_10_2021_417.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "0f1oYeHdaXk0",
        "outputId": "f7d4328c-380d-4995-e463-459253007f9d"
      },
      "source": [
        "!apt-get update\n",
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "!update-alternatives --set java /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java\n",
        "!java -version\n",
        "!pip install pyspark"
      ],
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\r0% [Working]\r            \rGet:1 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic InRelease [15.9 kB]\n",
            "\r0% [Waiting for headers] [Connecting to security.ubuntu.com (91.189.91.38)] [Co\r                                                                               \rHit:2 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
            "\r0% [Connecting to security.ubuntu.com (91.189.91.38)] [Connected to cloud.r-pro\r                                                                               \rGet:3 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
            "\r                                                                               \rHit:4 http://ppa.launchpad.net/cran/libgit2/ubuntu bionic InRelease\n",
            "\r0% [3 InRelease 60.5 kB/88.7 kB 68%] [Connecting to security.ubuntu.com (91.189\r0% [2 InRelease gpgv 242 kB] [3 InRelease 65.0 kB/88.7 kB 73%] [Connecting to s\r0% [2 InRelease gpgv 242 kB] [Waiting for headers] [Connecting to security.ubun\r                                                                               \rGet:5 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [74.6 kB]\n",
            "\r                                                                               \rGet:6 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic InRelease [15.9 kB]\n",
            "\r0% [2 InRelease gpgv 242 kB] [5 InRelease 20.0 kB/74.6 kB 27%] [Connecting to s\r0% [2 InRelease gpgv 242 kB] [Connecting to security.ubuntu.com (91.189.91.38)]\r0% [2 InRelease gpgv 242 kB] [Connecting to security.ubuntu.com (91.189.91.38)]\r                                                                               \rGet:7 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran40/ InRelease [3,626 B]\n",
            "Hit:8 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease\n",
            "Get:9 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
            "Ign:10 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease\n",
            "Ign:11 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
            "Hit:12 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Release\n",
            "Hit:13 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
            "Get:14 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main Sources [1,802 kB]\n",
            "Get:15 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu bionic/main amd64 Packages [922 kB]\n",
            "Get:16 http://archive.ubuntu.com/ubuntu bionic-updates/main amd64 Packages [2,801 kB]\n",
            "Get:17 http://archive.ubuntu.com/ubuntu bionic-updates/universe amd64 Packages [2,210 kB]\n",
            "Get:18 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu bionic/main amd64 Packages [40.8 kB]\n",
            "Get:20 http://security.ubuntu.com/ubuntu bionic-security/universe amd64 Packages [1,431 kB]\n",
            "Get:22 http://security.ubuntu.com/ubuntu bionic-security/main amd64 Packages [2,365 kB]\n",
            "Fetched 11.9 MB in 3s (3,607 kB/s)\n",
            "Reading package lists... Done\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java to provide /usr/bin/java (java) in manual mode\n",
            "openjdk version \"1.8.0_292\"\n",
            "OpenJDK Runtime Environment (build 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10)\n",
            "OpenJDK 64-Bit Server VM (build 25.292-b10, mixed mode)\n",
            "Collecting pyspark\n",
            "  Downloading pyspark-3.1.2.tar.gz (212.4 MB)\n",
            "\u001b[K     |████████████████████████████████| 212.4 MB 64 kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.9\n",
            "  Downloading py4j-0.10.9-py2.py3-none-any.whl (198 kB)\n",
            "\u001b[K     |████████████████████████████████| 198 kB 66.5 MB/s \n",
            "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.1.2-py2.py3-none-any.whl size=212880768 sha256=708ee46a410c23db1a84a5b0cc9bef429acd949f2437c44876b95d017f56f89c\n",
            "  Stored in directory: /root/.cache/pip/wheels/a5/0a/c1/9561f6fecb759579a7d863dcd846daaa95f598744e71b02c77\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.9 pyspark-3.1.2\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "m1S3-1uZgoE4",
        "outputId": "dfe3dc10-a15f-44dd-91a2-7ff637397e2a"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "pointsDF = spark.read.option('inferSchema','true').csv('points.txt')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c0','x') #Kolon isimlerini değiştiriyoruz\n",
        "pointsDF = pointsDF.withColumnRenamed('_c1','y') #Kolon isimlerini değiştiriyoruz\n",
        "pointsDF = pointsDF.withColumn(\"x\",pointsDF[\"x\"].cast(\"float\"))  #Kolonun veri tipi değiştiriyoruz\n",
        "pointsDF =  pointsDF.withColumn(\"2xy\", 2* pointsDF['x']+pointsDF['y']) #Yeni kolon ekliyoruz\n",
        "pointsDF.show()\n",
        "pointsDF.printSchema()"
      ],
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----+---+----+\n",
            "|   x|  y| 2xy|\n",
            "+----+---+----+\n",
            "| 5.0|  5|15.0|\n",
            "|15.0| 17|47.0|\n",
            "|15.0| 16|46.0|\n",
            "| 5.0|  6|16.0|\n",
            "| 6.0|  6|18.0|\n",
            "|16.0| 16|48.0|\n",
            "| 8.0|  8|24.0|\n",
            "| 7.0|  6|20.0|\n",
            "|17.0| 17|51.0|\n",
            "| 6.0|  7|19.0|\n",
            "| 7.0|  8|22.0|\n",
            "|18.0| 18|54.0|\n",
            "| 8.0|  6|22.0|\n",
            "|17.0| 16|50.0|\n",
            "| 5.0|  8|18.0|\n",
            "|17.0| 16|50.0|\n",
            "| 8.0|  8|24.0|\n",
            "|16.0| 17|49.0|\n",
            "|17.0| 18|52.0|\n",
            "|18.0| 16|52.0|\n",
            "+----+---+----+\n",
            "only showing top 20 rows\n",
            "\n",
            "root\n",
            " |-- x: float (nullable = true)\n",
            " |-- y: integer (nullable = true)\n",
            " |-- 2xy: float (nullable = true)\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "lI228ck7g-O3"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "pointsDF = spark.read.option('inferSchema','true').option('header','false').csv('points.txt')\n",
        "#pointsDF.printSchema()\n",
        "#pointsDF.show()\n",
        "\n",
        "#print( pointsDF.rdd.map(lambda x : tuple(x)).take(5) )\n",
        "\n",
        "\n",
        "pointsRDD = pointsDF.rdd\n",
        "\n",
        "\n",
        "\n",
        "df = pointsRDD.toDF()\n",
        "\n",
        "df.show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "gQT4fl-px9xJ",
        "outputId": "886af77b-8326-462a-ce75-0c3dc47157cf"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import FloatType\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "\n",
        "dollarDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','false')\\\n",
        "                        .option('delimiter','\\t').csv('DollarDataset.txt')\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "dollarDF = dollarDF.withColumnRenamed('_c0','id')\n",
        "dollarDF = dollarDF.withColumnRenamed('_c1','tarih')\n",
        "dollarDF = dollarDF.withColumnRenamed('_c2','fiyat')\n",
        "\n",
        "#dollarDF.show() \n",
        "\n",
        "#dollarDF.count() #17776\n",
        "\n",
        "\n",
        "dollarDF = dollarDF.filter(dollarDF['fiyat'].isNotNull())  #bu islemden sonra index bouldu\n",
        "\n",
        "\n",
        "#dollarDF.show(100) \n",
        "\n",
        "#dollarDF.count()  #17190 demekki 586 null values\n",
        "\n",
        "\n",
        "dollarDF = dollarDF.drop('id') #index column2ını drop ediyorum\n",
        "#dollarDF.show() \n",
        "\n",
        "rdd = dollarDF.rdd.zipWithIndex() #index'i en son'a ekledi\n",
        "\n",
        "#rdd.collect()\n",
        "\n",
        "\n",
        "dollarDF = rdd.map(lambda x : (x[1],x[0][0],x[0][1])).toDF()\n",
        "\n",
        "\n",
        "#dollarDF.show() #baslıklar bozudlu\n",
        "\n",
        "# data frame çevirince _1 _2 otomatik atanıyor\n",
        "dollarDF = dollarDF.withColumnRenamed('_1','id')\n",
        "dollarDF = dollarDF.withColumnRenamed('_2','tarih')\n",
        "dollarDF = dollarDF.withColumnRenamed('_3','fiyat')\n",
        "\n",
        "#dollarDF.show()\n",
        "#dollarDF.printSchema()\n",
        "\n",
        "def stringtoFloat(f):\n",
        "  f=f.replace('.','')\n",
        "  f=f.replace(',','.')\n",
        "  return float(f)\n",
        "\n",
        "stringtoFloatUDF = UserDefinedFunction(stringtoFloat,FloatType())  #eger yeni bir kolon ekleyeeksek ve bu mevcut kolanla işlemler yapılıp elde edilefcekse bu uderdefinedfunction yapmamız lazım\n",
        "\n",
        "dollarDF = dollarDF.withColumn(\"fiyat\",stringtoFloatUDF('fiyat'))\n",
        "\n",
        "#dollarDF.show()\n",
        "\n",
        "dollarDF2 = dollarDF\n",
        "dollarDF2 = dollarDF2.withColumn('id',dollarDF2['id']+1)\n",
        "dollarDF2 = dollarDF2.withColumnRenamed('tarih','tarih2')\n",
        "dollarDF2 = dollarDF2.withColumnRenamed('fiyat','fiyat2')\n",
        "\n",
        "#dollarDF2.show()\n",
        "\n",
        "joinedDF = dollarDF2.join(dollarDF,'id')\n",
        "\n",
        "joinedDF = joinedDF.sort('id')\n",
        "\n",
        "joinedDF = joinedDF.withColumn('oran',((joinedDF[\"fiyat\"]-joinedDF[\"fiyat2\"])*100)/joinedDF['fiyat2'])\n",
        "\n",
        "joinedDF = joinedDF.sort('oran',ascending=False)\n",
        "joinedDF.show()"
      ],
      "execution_count": 53,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----+----------+---------+----------+---------+------------------+\n",
            "|   id|    tarih2|   fiyat2|     tarih|    fiyat|              oran|\n",
            "+-----+----------+---------+----------+---------+------------------+\n",
            "| 2683|19-08-1960|      2.8|22-08-1960|      9.0| 221.4285751994776|\n",
            "| 7589|24-01-1980|     35.0|25-01-1980|     70.0|             100.0|\n",
            "| 5201|07-08-1970|      9.0|10-08-1970|    14.85| 65.00000678168402|\n",
            "|12912|22-02-2001| 685391.0|23-02-2001| 957879.0| 39.75657690281898|\n",
            "|11172|05-04-1994| 23031.84|06-04-1994| 31988.89| 38.88986153848502|\n",
            "| 7433|11-06-1979|     26.5|12-06-1979|     35.0|32.075471698113205|\n",
            "| 7108|28-02-1978|    19.25|01-03-1978|     25.0| 29.87012987012987|\n",
            "|11173|06-04-1994| 31988.89|07-04-1994| 39853.13| 24.58427901483376|\n",
            "|11125|26-01-1994| 15155.63|27-01-1994|  17215.5|13.591451969845416|\n",
            "|12913|23-02-2001| 957879.0|26-02-2001|1072988.0|12.017071049683729|\n",
            "|11186|25-04-1994| 30658.56|26-04-1994| 33925.01|10.654287715190682|\n",
            "|12936|03-04-2001|1114738.0|04-04-2001|1232604.0|10.573426222125738|\n",
            "| 6997|20-09-1977|     17.5|21-09-1977|    19.25|              10.0|\n",
            "|12935|02-04-2001|1020560.0|03-04-2001|1114738.0| 9.228070863055578|\n",
            "|13010|17-07-2001|1366647.0|18-07-2001|1482571.0| 8.482365965754141|\n",
            "|11211|02-06-1994| 30623.63|03-06-1994|  33181.5| 8.352598926775999|\n",
            "|11270|24-08-1994|  32583.7|25-08-1994| 35291.28| 8.309620883812805|\n",
            "|14843|22-10-2008|     1.52|23-10-2008|     1.63| 7.236843450908203|\n",
            "|12933|29-03-2001| 992354.0|30-03-2001|1056544.0| 6.468457828557148|\n",
            "|12922|14-03-2001| 941519.0|15-03-2001| 999788.0| 6.188828903081085|\n",
            "+-----+----------+---------+----------+---------+------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "8PxbLwzpEEst",
        "outputId": "f62b01e5-7e7c-4deb-8807-e93e244772ce"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import FloatType\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "\n",
        "depremDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter','\\t').csv('EarthquakeDataset.txt')\n",
        "\n",
        "\n",
        "depremDF = depremDF.select(\"Latitude\",\"Longitude\",\"xM\",\"Location\")\n",
        "depremDF.show()\n"
      ],
      "execution_count": 55,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+---------+---+--------------------+\n",
            "|Latitude|Longitude| xM|            Location|\n",
            "+--------+---------+---+--------------------+\n",
            "| 37.8435|  26.7775|3.8|          EGE DENIZI|\n",
            "| 36.4918|  28.7092|3.9|             AKDENIZ|\n",
            "|  37.796|  26.4165|4.4|ONIKI ADALAR (AKD...|\n",
            "| 38.4172|  39.1482|3.9|KAVAKKOY-SIVRICE ...|\n",
            "|  36.475|  28.7852|4.2|             AKDENIZ|\n",
            "|  38.505|   39.218|5.6|KAVAKTEPE- (ELAZI...|\n",
            "| 38.5305|  39.2102|4.1|KAVAKTEPE- (ELAZI...|\n",
            "| 38.0033|  42.9642|3.9|KORULU-CATAK (VAN...|\n",
            "| 36.5213|   28.695|3.7|             AKDENIZ|\n",
            "| 37.8888|  27.6075|3.7|DAGKARAAGAC-GERME...|\n",
            "| 35.7737|   36.447|3.9|              SURIYE|\n",
            "| 37.8967|  43.8062|4.5|AYDEMIR-BASKALE (...|\n",
            "|  37.126|  28.5738|4.0|ARICILAR-ULA (MUG...|\n",
            "| 37.1175|  31.0832|3.7|ETLER-SERIK (ANTA...|\n",
            "| 35.7012|  26.2608|3.6|             AKDENIZ|\n",
            "| 35.7032|  26.2382|4.6|             AKDENIZ|\n",
            "|  35.704|  26.2952|3.8|             AKDENIZ|\n",
            "|  35.691|  26.2818|3.9|             AKDENIZ|\n",
            "| 38.8573|  43.5082|4.7|ERMISLER- (VAN) [...|\n",
            "|  37.683|  27.0502|3.5|MILLI PARK-KUSADA...|\n",
            "+--------+---------+---+--------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "l2t5f_VZFo-Y",
        "outputId": "128936ac-3da7-4175-e5fe-a37cf9b3ea0a"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import FloatType,StringType\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "\n",
        "depremDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter','\\t').csv('EarthquakeDataset.txt')\n",
        "\n",
        "\n",
        "depremDF = depremDF.select(\"Latitude\",\"Longitude\",\"xM\",\"Location\")\n",
        "\n",
        "\n",
        "def bolgeAta(enlem,boylam):\n",
        "    bolge = \"\"\n",
        "    if (enlem >= 39):\n",
        "        if (boylam <=31):\n",
        "            bolge = \"Bölge 1\"\n",
        "        elif (boylam <= 36):\n",
        "            bolge = \"Bölge 2\"\n",
        "        elif (boylam <= 41):\n",
        "            bolge = \"Bölge 3\"\n",
        "        else:\n",
        "            bolge = \"Bölge 4\"\n",
        "    else:\n",
        "        if (boylam <=31):\n",
        "            bolge = \"Bölge 5\"\n",
        "        elif (boylam <= 36):\n",
        "            bolge = \"Bölge 6\"\n",
        "        elif (boylam <= 41):\n",
        "            bolge = \"Bölge 7\"\n",
        "        else:\n",
        "            bolge = \"Bölge 8\"\n",
        "    return bolge\n",
        "\n",
        "bolgeAtaUDF = UserDefinedFunction(bolgeAta,StringType())\n",
        "depremDF = depremDF.withColumn('Bolge',bolgeAtaUDF('Latitude','Longitude'))\n",
        "depremDF = depremDF.groupBy('Bolge').count()\n",
        "depremDF = depremDF.sort('count',ascending=False)\n",
        "depremDF.show(9)\n"
      ],
      "execution_count": 68,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----+\n",
            "|  Bolge|count|\n",
            "+-------+-----+\n",
            "|Bölge 5| 7946|\n",
            "|Bölge 1| 2827|\n",
            "|Bölge 8| 1547|\n",
            "|Bölge 7| 1335|\n",
            "|Bölge 6| 1277|\n",
            "|Bölge 2|  994|\n",
            "|Bölge 3|  982|\n",
            "|Bölge 4|  752|\n",
            "+-------+-----+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6t5gs7wUNyMj",
        "outputId": "03208876-0b58-4ce7-b5f1-52c6be8a775e"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "worldDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','false')\\\n",
        "                        .option('delimiter',',').csv('world.txt')\n",
        "\n",
        "\n",
        "worldDF = worldDF.withColumnRenamed('_c0','ulke')\n",
        "worldDF = worldDF.withColumnRenamed('_c1','kita')\n",
        "worldDF = worldDF.withColumnRenamed('_c2','baskent')\n",
        "worldDF = worldDF.withColumnRenamed('_c3','nufus')\n",
        "\n",
        "#worldDF.show()\n",
        "\n",
        "#Dünyada hangi kıtada kaç ülke vardır ?\n",
        "cozum1DF = worldDF.groupBy('kita').count()\n",
        "cozum1DF = cozum1DF.sort('count',ascending=False)\n",
        "cozum1DF.show()\n",
        "#Kıtalardaki ülkelerinin ortalama nüfusu (kıta bazında) nedir ?\n",
        "cozum2DF = worldDF.groupBy('kita').mean()\n",
        "cozum2DF = cozum2DF.sort('avg(nufus)',ascending=False)\n",
        "cozum2DF = cozum2DF.withColumn('ortalama nufus',cozum2DF['avg(nufus)'].cast('int'))\n",
        "cozum2DF.show()\n",
        "\n",
        "cozum3DF = worldDF.groupBy('kita').sum('nufus')\n",
        "cozum3DF = cozum3DF.sort('sum(nufus)',ascending=False)\n",
        "cozum3DF.show()\n",
        "\n",
        "#Kıtaların toplam nüfusu"
      ],
      "execution_count": 78,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------------+-----+\n",
            "|         kita|count|\n",
            "+-------------+-----+\n",
            "|       Africa|   55|\n",
            "|         Asia|   49|\n",
            "|       Europe|   46|\n",
            "|North America|   23|\n",
            "|       Oceana|   13|\n",
            "|South America|   12|\n",
            "+-------------+-----+\n",
            "\n",
            "+-------------+--------------------+--------------+\n",
            "|         kita|          avg(nufus)|ortalama nufus|\n",
            "+-------------+--------------------+--------------+\n",
            "|         Asia| 8.127662059183674E7|      81276620|\n",
            "|South America|       3.124975725E7|      31249757|\n",
            "|North America| 2.222895156521739E7|      22228951|\n",
            "|       Africa|1.6516486363636363E7|      16516486|\n",
            "|       Europe|1.4397671369565217E7|      14397671|\n",
            "|       Oceana|  2462989.3076923075|       2462989|\n",
            "+-------------+--------------------+--------------+\n",
            "\n",
            "+-------------+----------+\n",
            "|         kita|sum(nufus)|\n",
            "+-------------+----------+\n",
            "|         Asia|3982554409|\n",
            "|       Africa| 908406750|\n",
            "|       Europe| 662292883|\n",
            "|North America| 511265886|\n",
            "|South America| 374997087|\n",
            "|       Oceana|  32018861|\n",
            "+-------------+----------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "0K5R9-fsUL44",
        "outputId": "f1edc35e-50ce-4ddc-92d1-934706783c3c"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "worldDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','false')\\\n",
        "                        .option('delimiter',',').csv('world.txt')\n",
        "\n",
        "worldDF = worldDF.withColumnRenamed('_c0','ulke')\n",
        "worldDF = worldDF.withColumnRenamed('_c1','kita')\n",
        "worldDF = worldDF.withColumnRenamed('_c2','baskent')\n",
        "worldDF = worldDF.withColumnRenamed('_c3','nufus')\n",
        "\n",
        "worldDF.createOrReplaceTempView('world')\n",
        "\n",
        "cozum1DF = spark.sql('select kita,count(*) as count from world group by kita order by count desc')\n",
        "cozum1DF.show()"
      ],
      "execution_count": 80,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------------+-----+\n",
            "|         kita|count|\n",
            "+-------------+-----+\n",
            "|       Africa|   55|\n",
            "|         Asia|   49|\n",
            "|       Europe|   46|\n",
            "|North America|   23|\n",
            "|       Oceana|   13|\n",
            "|South America|   12|\n",
            "+-------------+-----+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "o1KNNT-e3kgF",
        "outputId": "164cd4e8-35f3-465d-88c4-0ff93d6ba970"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "actorsDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',',').csv('join-actors_BNO.txt')\n",
        "\n",
        "seriesDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',',').csv('join-series_BNO.txt')\n",
        "seriesDF.show()\n",
        "actorsDF.show(25)\n",
        "\n",
        "'''\n",
        "#Without SQL\n",
        "joinedDF = actorsDF.join(seriesDF,'movieid','outer')\n",
        "joinedDF1 = actorsDF.join(seriesDF,'movieid','right')\n",
        "joinedDF2 = actorsDF.join(seriesDF,'movieid','left') #left , right ,inner ,outer\n",
        "joinedDF3 = actorsDF.join(seriesDF,'movieid','inner')\n",
        "\n",
        "#joinedDF = seriesDF.join(actorsDF,'movieid','outer')\n",
        "#joinedDF1 = seriesDF.join(actorsDF,'movieid','right')\n",
        "#joinedDF2 = seriesDF.join(actorsDF,'movieid','left') #left , right ,inner ,outer\n",
        "#joinedDF3 = seriesDF.join(actorsDF,'movieid','inner')\n",
        "\n",
        "\n",
        "joinedDF= joinedDF.sort(\"movieid\",ascending=True) \n",
        "\n",
        "joinedDF1= joinedDF1.sort(\"movieid\",ascending=True) \n",
        "joinedDF2= joinedDF2.sort(\"movieid\",ascending=True) \n",
        "joinedDF3= joinedDF3.sort(\"movieid\",ascending=True) \n",
        "\n",
        "joinedDF.show(26)\n",
        "joinedDF1.show(26)\n",
        "joinedDF2.show(26)\n",
        "joinedDF3.show(26)\n",
        "\n",
        "'''\n",
        "\n",
        "#with SQl\n",
        "\n",
        "actorsDF.createOrReplaceTempView('actors')\n",
        "seriesDF.createOrReplaceTempView('series')\n",
        "#sqlDF = spark.sql('select a.*,s.* from actors a,series s where s.movieid=a.movieid')\n",
        "sqlDF = spark.sql('select a.*,s.* from actors a full outer join series s on s.movieid=a.movieid')\n",
        "sqlDF.show(40)"
      ],
      "execution_count": 21,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+------------------+\n",
            "|movieid|         moviename|\n",
            "+-------+------------------+\n",
            "|      1|     Lethal Weapon|\n",
            "|      2|     The Last Ship|\n",
            "|      3|Person of Interest|\n",
            "|      4|   Game of Thrones|\n",
            "|      5|      Flashforward|\n",
            "|      6|        Revolution|\n",
            "|      7|              Lost|\n",
            "|      8|  Leyle ile Mecnun|\n",
            "|      9|               BNO|\n",
            "+-------+------------------+\n",
            "\n",
            "+---+--------------------+-------+\n",
            "| id|                name|movieid|\n",
            "+---+--------------------+-------+\n",
            "|  1|           Eric Dane|      2|\n",
            "|  2|        Adam Baldwin|      2|\n",
            "|  3|     Charles Parnell|      2|\n",
            "|  4|         Billy Burke|      6|\n",
            "|  5|    Tracy Spiridakos|      6|\n",
            "|  6|  Giancarlo Esposito|      6|\n",
            "|  7|        Damon Wayans|      1|\n",
            "|  8|     Clayne Crawford|      1|\n",
            "|  9|      Joseph Fiennes|      5|\n",
            "| 10|      Jack Davenport|      5|\n",
            "| 11|    Zachary Knighton|      5|\n",
            "| 12|        Jim Caviezel|      3|\n",
            "| 13|     Michael Emerson|      3|\n",
            "| 14|    Taraji P. Henson|      3|\n",
            "| 15|      Peter Dinklage|      4|\n",
            "| 16|       Emilia Clarke|      4|\n",
            "| 17|       Kit Harington|      4|\n",
            "| 18|Nikolaj Coster-Wa...|      4|\n",
            "| 19|     Ahmet Demirelli|   null|\n",
            "| 20|       Josh Holloway|      7|\n",
            "| 21|    Evangeline Lilly|      7|\n",
            "| 22|         Matthew Fox|      7|\n",
            "| 23|                null|      9|\n",
            "+---+--------------------+-------+\n",
            "\n",
            "+----+--------------------+-------+-------+------------------+\n",
            "|  id|                name|movieid|movieid|         moviename|\n",
            "+----+--------------------+-------+-------+------------------+\n",
            "|  19|     Ahmet Demirelli|   null|   null|              null|\n",
            "|   7|        Damon Wayans|      1|      1|     Lethal Weapon|\n",
            "|   8|     Clayne Crawford|      1|      1|     Lethal Weapon|\n",
            "|   4|         Billy Burke|      6|      6|        Revolution|\n",
            "|   5|    Tracy Spiridakos|      6|      6|        Revolution|\n",
            "|   6|  Giancarlo Esposito|      6|      6|        Revolution|\n",
            "|  12|        Jim Caviezel|      3|      3|Person of Interest|\n",
            "|  13|     Michael Emerson|      3|      3|Person of Interest|\n",
            "|  14|    Taraji P. Henson|      3|      3|Person of Interest|\n",
            "|   9|      Joseph Fiennes|      5|      5|      Flashforward|\n",
            "|  10|      Jack Davenport|      5|      5|      Flashforward|\n",
            "|  11|    Zachary Knighton|      5|      5|      Flashforward|\n",
            "|  23|                null|      9|      9|               BNO|\n",
            "|  15|      Peter Dinklage|      4|      4|   Game of Thrones|\n",
            "|  16|       Emilia Clarke|      4|      4|   Game of Thrones|\n",
            "|  17|       Kit Harington|      4|      4|   Game of Thrones|\n",
            "|  18|Nikolaj Coster-Wa...|      4|      4|   Game of Thrones|\n",
            "|null|                null|   null|      8|  Leyle ile Mecnun|\n",
            "|  20|       Josh Holloway|      7|      7|              Lost|\n",
            "|  21|    Evangeline Lilly|      7|      7|              Lost|\n",
            "|  22|         Matthew Fox|      7|      7|              Lost|\n",
            "|   1|           Eric Dane|      2|      2|     The Last Ship|\n",
            "|   2|        Adam Baldwin|      2|      2|     The Last Ship|\n",
            "|   3|     Charles Parnell|      2|      2|     The Last Ship|\n",
            "+----+--------------------+-------+-------+------------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Icx0rmZEC1in",
        "outputId": "6b566117-d244-4724-8227-a8a3efca1642"
      },
      "source": [
        "\n",
        "#points-null-vakues.txt dosyasında;\n",
        " # --- ilk kolondaki null değerleri 1 ile \n",
        " # --- ikinci kolondaki null değerleri 0 ile değiştiriniz!! (UserDefinedFunction)\n",
        " # --- kısayol\n",
        " # --- ilk kolondaki null değerleri ortalama ile\n",
        " # --- ikinci kolondaki medyan ile değişitiriniz\n",
        "\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import IntegerType\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "pointsDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','false')\\\n",
        "                        .option('delimiter',',').csv('points-null-values.txt')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c0','x')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c1','y')\n",
        "\n",
        "#pointsDF.show(25)\n",
        "\n",
        "\n",
        "def replaceNullColumn1(x):\n",
        "  if x is None:\n",
        "    x=999\n",
        "  return x\n",
        "\n",
        "def replaceNullColumn2(x):\n",
        "  if x is None:\n",
        "    x=777\n",
        "  return x\n",
        "\n",
        "replaceUDF1=UserDefinedFunction(replaceNullColumn1,IntegerType())\n",
        "replaceUDF2=UserDefinedFunction(replaceNullColumn2,IntegerType())\n",
        "\n",
        "pointsDF = pointsDF.withColumn('X_',replaceUDF1('x'))\n",
        "pointsDF = pointsDF.withColumn('Y_',replaceUDF2('y'))\n",
        "\n",
        "pointsDF.show(50)"
      ],
      "execution_count": 26,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+--------+--------+--------+\n",
            "|       x|       y|      X_|      Y_|\n",
            "+--------+--------+--------+--------+\n",
            "|       5|       5|       5|       5|\n",
            "|     105|     107|     105|     107|\n",
            "|     105|     106|     105|     106|\n",
            "|    null|       6|     999|       6|\n",
            "|       6|       6|       6|       6|\n",
            "|     106|    null|     106|     777|\n",
            "|       7|       7|       7|       7|\n",
            "|    null|       6|     999|       6|\n",
            "|     107|     107|     107|     107|\n",
            "|       6|       7|       6|       7|\n",
            "|       7|    null|       7|     777|\n",
            "|     108|     108|     108|     108|\n",
            "|       8|       6|       8|       6|\n",
            "|    null|     108|     999|     108|\n",
            "|       5|       8|       5|       8|\n",
            "|     107|     106|     107|     106|\n",
            "|       8|       8|       8|       8|\n",
            "|    null|     107|     999|     107|\n",
            "|     107|     108|     107|     108|\n",
            "|     108|     106|     108|     106|\n",
            "|10000008|10000006|10000008|10000006|\n",
            "+--------+--------+--------+--------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "JII_l1DSEns3",
        "outputId": "99bc03f4-da90-46d3-b2b6-8c4768674590"
      },
      "source": [
        "#Kısa Yol (Sabit sayı ile değiştirme)\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import IntegerType\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "pointsDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','false')\\\n",
        "                        .option('delimiter',',').csv('points-null-values.txt')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c0','x')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c1','y')\n",
        "\n",
        "pointsDF = pointsDF.fillna(9999,'x')\n",
        "pointsDF = pointsDF.fillna(8888,'y')\n",
        "pointsDF.show()"
      ],
      "execution_count": 27,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----+----+\n",
            "|   x|   y|\n",
            "+----+----+\n",
            "|   5|   5|\n",
            "| 105| 107|\n",
            "| 105| 106|\n",
            "|9999|   6|\n",
            "|   6|   6|\n",
            "| 106|8888|\n",
            "|   7|   7|\n",
            "|9999|   6|\n",
            "| 107| 107|\n",
            "|   6|   7|\n",
            "|   7|8888|\n",
            "| 108| 108|\n",
            "|   8|   6|\n",
            "|9999| 108|\n",
            "|   5|   8|\n",
            "| 107| 106|\n",
            "|   8|   8|\n",
            "|9999| 107|\n",
            "| 107| 108|\n",
            "| 108| 106|\n",
            "+----+----+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "IKiP-kJLE4TX",
        "outputId": "78c0dc14-dd46-4eac-9bed-f52dd2325e4c"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import IntegerType\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "pointsDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','false')\\\n",
        "                        .option('delimiter',',').csv('points-null-values.txt')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c0','x')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c1','y')\n",
        "pointsDF.show(40)\n",
        "pointsDF.describe().show()  #aşağıdaki cıktıda standart sapma ortalama mean değerden cok fazla demekki aykırı degerler var o zaman medıan kullanmak daha iyidir"
      ],
      "execution_count": 30,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+--------+\n",
            "|       x|       y|\n",
            "+--------+--------+\n",
            "|       5|       5|\n",
            "|     105|     107|\n",
            "|     105|     106|\n",
            "|    null|       6|\n",
            "|       6|       6|\n",
            "|     106|    null|\n",
            "|       7|       7|\n",
            "|    null|       6|\n",
            "|     107|     107|\n",
            "|       6|       7|\n",
            "|       7|    null|\n",
            "|     108|     108|\n",
            "|       8|       6|\n",
            "|    null|     108|\n",
            "|       5|       8|\n",
            "|     107|     106|\n",
            "|       8|       8|\n",
            "|    null|     107|\n",
            "|     107|     108|\n",
            "|     108|     106|\n",
            "|10000008|10000006|\n",
            "+--------+--------+\n",
            "\n",
            "+-------+-----------------+-----------------+\n",
            "|summary|                x|                y|\n",
            "+-------+-----------------+-----------------+\n",
            "|  count|               17|               19|\n",
            "|   mean|         588289.0|526369.8947368421|\n",
            "| stddev|2425344.472743975|2294145.690034356|\n",
            "|    min|                5|                5|\n",
            "|    max|         10000008|         10000006|\n",
            "+-------+-----------------+-----------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "iinmq1S_Eh-Y",
        "outputId": "72ad6264-a5f5-4763-c88b-0d79b4b88fb6"
      },
      "source": [
        "#Kısa Yol (Ortalama ile değiştirme)\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import Imputer\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "pointsDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','false')\\\n",
        "                        .option('delimiter',',').csv('points-null-values.txt')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c0','x')\n",
        "pointsDF = pointsDF.withColumnRenamed('_c1','y')\n",
        "pointsDF.show(26)\n",
        "\n",
        "#Imputer kullanırken öncelikle outlier var mı diye bakarız\n",
        "#Eğer outlier yok ise ortalama (Mean) ile değiştirmek mantıklıdır\n",
        "#Eğer outlier varsa ve silmek istemiyorsak medyan (median) ile değiştirmek mantıklıdır\n",
        "\n",
        "meanImputer = Imputer(inputCol='x',outputCol='imputedXMean',strategy='mean')\n",
        "pointsDF = meanImputer.fit(pointsDF).transform(pointsDF)\n",
        "\n",
        "medianImputer = Imputer(inputCol='y',outputCol='imputedYMedian',strategy='median')\n",
        "pointsDF = medianImputer.fit(pointsDF).transform(pointsDF)\n",
        "\n",
        "pointsDF.show(100)\n"
      ],
      "execution_count": 33,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+--------+\n",
            "|       x|       y|\n",
            "+--------+--------+\n",
            "|       5|       5|\n",
            "|     105|     107|\n",
            "|     105|     106|\n",
            "|    null|       6|\n",
            "|       6|       6|\n",
            "|     106|    null|\n",
            "|       7|       7|\n",
            "|    null|       6|\n",
            "|     107|     107|\n",
            "|       6|       7|\n",
            "|       7|    null|\n",
            "|     108|     108|\n",
            "|       8|       6|\n",
            "|    null|     108|\n",
            "|       5|       8|\n",
            "|     107|     106|\n",
            "|       8|       8|\n",
            "|    null|     107|\n",
            "|     107|     108|\n",
            "|     108|     106|\n",
            "|10000008|10000006|\n",
            "+--------+--------+\n",
            "\n",
            "+--------+--------+------------+--------------+\n",
            "|       x|       y|imputedXMean|imputedYMedian|\n",
            "+--------+--------+------------+--------------+\n",
            "|       5|       5|           5|             5|\n",
            "|     105|     107|         105|           107|\n",
            "|     105|     106|         105|           106|\n",
            "|    null|       6|      588289|             6|\n",
            "|       6|       6|           6|             6|\n",
            "|     106|    null|         106|           106|\n",
            "|       7|       7|           7|             7|\n",
            "|    null|       6|      588289|             6|\n",
            "|     107|     107|         107|           107|\n",
            "|       6|       7|           6|             7|\n",
            "|       7|    null|           7|           106|\n",
            "|     108|     108|         108|           108|\n",
            "|       8|       6|           8|             6|\n",
            "|    null|     108|      588289|           108|\n",
            "|       5|       8|           5|             8|\n",
            "|     107|     106|         107|           106|\n",
            "|       8|       8|           8|             8|\n",
            "|    null|     107|      588289|           107|\n",
            "|     107|     108|         107|           108|\n",
            "|     108|     106|         108|           106|\n",
            "|10000008|10000006|    10000008|      10000006|\n",
            "+--------+--------+------------+--------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "mgoSPVq_MYYc"
      },
      "source": [
        "#Homework 2 hint: DataFrame pivot, Recommendation Engine , Collabrative Filtering"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "B6U2xsb5Mfbc"
      },
      "source": [
        "## Categorical Values IMPUTER##"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "3N_qI4yaMnPc",
        "outputId": "9e048210-6c17-4545-ea48-6682cca3b5bf"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import IntegerType\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "evlerDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',';').csv('evler.csv')\n",
        "\n",
        "def lengthFunction(text):\n",
        "  return len(text)\n",
        "\n",
        "#pointsDF = pointsDF.withColumn('sehirX',pointsDF['Sehir']) #bu şekilde çalışıyor ama len almak isteidğimizde hata veriyor. Ç;ünkü tüm kolonu obje olarak\n",
        "# almaya çalışıyor \n",
        "#çarpım dolduğunda bunu satır satır hepsiyle yapıyor ama len denildiğinde bunu anlamıyor lenght metodunu burada çalışmaz UDF yazmamız lazım\n",
        "# yeni eklenecek kolan bu değer İzmir'i\n",
        "\n",
        "IUDF = UserDefinedFunction(lengthFunction,IntegerType())\n",
        "\n",
        "\n",
        "\n",
        "evlerDF = evlerDF.withColumn('sehirX',IUDF('Sehir'))\n",
        "\n",
        "evlerDF.show()\n"
      ],
      "execution_count": 49,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+---+-----+------+------+\n",
            "|   Sehir|Oda|salon| fiyat|sehirX|\n",
            "+--------+---+-----+------+------+\n",
            "|   izmir|  3|    1|500000|     5|\n",
            "|istanbul|  3|    1|800000|     8|\n",
            "|  ankara|  3|    1|600000|     6|\n",
            "|   izmir|  2|    1|300000|     5|\n",
            "|istanbul|  2|    1|450000|     8|\n",
            "|istanbul|  1|    1|250000|     8|\n",
            "|  ankara|  2|    1|380000|     6|\n",
            "|   izmir|  1|    1|220000|     5|\n",
            "|istanbul|  1|    1|245000|     8|\n",
            "+--------+---+-----+------+------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uT9DlToAPMx6"
      },
      "source": [
        "## Categorical Values&One Encoding ##"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "p9U5pmoAPU34",
        "outputId": "01672f10-bd31-4369-cd38-29fb40bacec8"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer,OneHotEncoder,VectorAssembler # böle bir indexleme var şehire göre modelimi girdisi etkilensin istiyoruz f(a,b,c)= ax +by +e bu x'i etkilicek\n",
        "from pyspark.ml.regression import LinearRegression\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "evlerDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',';').csv('evler.csv')\n",
        "\n",
        "evlerDF.show()   \n",
        "\n",
        "#machine learning kosturmak için datamız tümuyle sayısal olmalı, bu sebeple sehir isimlerini değiştircez\n",
        "\n",
        "indexer = StringIndexer(inputCol='Sehir',outputCol='indexedSehir') ##Linear modellerde bu kulanılmamalı (Tree-based lerde kötü sonuç vermeyebilir)\n",
        "\n",
        "evlerDF = indexer.fit(evlerDF).transform(evlerDF)\n",
        "\n",
        "evlerDF.show()   \n",
        "#Index işleminden sonra Encode işlemi yaparsak en doğrusu olur (Hem linear hemde tree-based algoritmalar için)\n",
        "\n",
        "encoder = OneHotEncoder(inputCol='indexedSehir',outputCol='encodedSehir')\n",
        "\n",
        "evlerDF = encoder.fit(evlerDF).transform(evlerDF)\n",
        "\n",
        "evlerDF.show()  "
      ],
      "execution_count": 53,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+---+-----+------+\n",
            "|   Sehir|Oda|salon| fiyat|\n",
            "+--------+---+-----+------+\n",
            "|   izmir|  3|    1|500000|\n",
            "|istanbul|  3|    1|800000|\n",
            "|  ankara|  3|    1|600000|\n",
            "|   izmir|  2|    1|300000|\n",
            "|istanbul|  2|    1|450000|\n",
            "|istanbul|  1|    1|250000|\n",
            "|  ankara|  2|    1|380000|\n",
            "|   izmir|  1|    1|220000|\n",
            "|istanbul|  1|    1|245000|\n",
            "+--------+---+-----+------+\n",
            "\n",
            "+--------+---+-----+------+------------+\n",
            "|   Sehir|Oda|salon| fiyat|indexedSehir|\n",
            "+--------+---+-----+------+------------+\n",
            "|   izmir|  3|    1|500000|         1.0|\n",
            "|istanbul|  3|    1|800000|         0.0|\n",
            "|  ankara|  3|    1|600000|         2.0|\n",
            "|   izmir|  2|    1|300000|         1.0|\n",
            "|istanbul|  2|    1|450000|         0.0|\n",
            "|istanbul|  1|    1|250000|         0.0|\n",
            "|  ankara|  2|    1|380000|         2.0|\n",
            "|   izmir|  1|    1|220000|         1.0|\n",
            "|istanbul|  1|    1|245000|         0.0|\n",
            "+--------+---+-----+------+------------+\n",
            "\n",
            "+--------+---+-----+------+------------+-------------+\n",
            "|   Sehir|Oda|salon| fiyat|indexedSehir| encodedSehir|\n",
            "+--------+---+-----+------+------------+-------------+\n",
            "|   izmir|  3|    1|500000|         1.0|(2,[1],[1.0])|\n",
            "|istanbul|  3|    1|800000|         0.0|(2,[0],[1.0])|\n",
            "|  ankara|  3|    1|600000|         2.0|    (2,[],[])|\n",
            "|   izmir|  2|    1|300000|         1.0|(2,[1],[1.0])|\n",
            "|istanbul|  2|    1|450000|         0.0|(2,[0],[1.0])|\n",
            "|istanbul|  1|    1|250000|         0.0|(2,[0],[1.0])|\n",
            "|  ankara|  2|    1|380000|         2.0|    (2,[],[])|\n",
            "|   izmir|  1|    1|220000|         1.0|(2,[1],[1.0])|\n",
            "|istanbul|  1|    1|245000|         0.0|(2,[0],[1.0])|\n",
            "+--------+---+-----+------+------------+-------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "wLAGEMhIYFWI"
      },
      "source": [
        "#Dummyfication  bu kullanılan yönteme göre yeni column'lar ekleyip diğer column'ı iptal ediyoruz3 şehir varsa son şehir eklenmez bunun için hoca'nın yüklediği excel'e bak\n",
        "#OneHotEncoding\n",
        "\n",
        "\n",
        "# Bu veriyi büyütür çünkü her yeni sehirler için yeni column ortaya çıkarıyoruz, kategorisel bir kalonumuz varsa, plaka kodu da olabilir,şehir ismide\n",
        "# cinsiyet,yaş bunlar\n",
        "\n",
        "#Sparsevector , densevector bunlar distinct eleman sayısı listenin yarısından az ise işe yarıyor buna bakarsın\n",
        "\n",
        "#distinct sayısına nasıl bakılır öğrenmem lazım"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "2MiVL6IdYUb9",
        "outputId": "31e8f318-859e-4dc5-e350-9a1309622379"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer,OneHotEncoder,VectorAssembler # böle bir indexleme var şehire göre modelimi girdisi etkilensin istiyoruz f(a,b,c)= ax +by +e bu x'i etkilicek\n",
        "from pyspark.ml.regression import LinearRegression\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "\n",
        "evlerDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',';').csv('evler.csv')\n",
        "\n",
        "\n",
        "indexer = StringIndexer(inputCol='Sehir',outputCol='indexedSehir')  ##linear modellerde bu kullanılması\n",
        "indexModel=indexer.fit(evlerDF)\n",
        "evlerDF = indexModel.transform(evlerDF)\n",
        "\n",
        "encoder = OneHotEncoder(inputCol='indexedSehir',outputCol='encodedSehir')\n",
        "#encodeModel=encoder.fit(evlerDF)\n",
        "#evlerDF = encodeModel.transform(evlerDF)\n",
        "\n",
        "evlerDF= encoder.fit(evlerDF).transform(evlerDF)\n",
        "\n",
        "evlerDF = evlerDF.withColumnRenamed('fiyat','label') #label atama\n",
        "evlerDF = evlerDF.select('encodedSehir','Oda','salon','label')\n",
        "\n",
        "vec =VectorAssembler(inputCols=['encodedSehir','Oda','salon'],outputCol='features')\n",
        "evlerDF =vec.transform(evlerDF)\n",
        "#evlerDF.show()\n",
        "\n",
        "evlerDF = evlerDF.select('features','label')\n",
        "#evlerDF.show()\n",
        "\n",
        "lr = LinearRegression()\n",
        "\n",
        "model = lr.fit(evlerDF) ##Makine Öğrenmesinin yapıldığı satır\n",
        "print('Bitti')\n",
        "\n",
        "print(\"Çarpanlar:  \",model.coefficients)\n",
        "print(\"Sabit sayi \", model.intercept)"
      ],
      "execution_count": 63,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Bitti\n",
            "Çarpanlar:   [107857.14398708245,-42261.90458170463,215476.19109859318,0.0]\n",
            "Sabit sayi  -48690.477997543654\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "6CElb45AayxY"
      },
      "source": [
        "\n",
        "#SparseVector  --> DenseVector\n",
        "#(2,[],[])     --> [0,0]\n",
        "#(2,[1],[1.0]) --> [0,1]\n",
        "#(2,[0],[1.0]  --> [1,0]\n",
        "#[1,2,3,4] --> 4 *  32bit\n",
        "#[11,0,0,0,34,0,23,0,0,0,0,0,....]  --> 1000000 * 32bit --> 4 MB #DenseVector\n",
        "#(1000000, [0,4,6],[11,34,23] )  --> 7 * 32bit #SparseVector"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "xnb6WX7WbF6H",
        "outputId": "fb03438c-9603-49dc-d645-dccf5c15c6af"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer,OneHotEncoder,VectorAssembler # böle bir indexleme var şehire göre modelimi girdisi etkilensin istiyoruz f(a,b,c)= ax +by +e bu x'i etkilicek\n",
        "from pyspark.ml.regression import LinearRegression\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "\n",
        "evlerDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',';').csv('evler.csv')\n",
        "\n",
        "\n",
        "indexer = StringIndexer(inputCol='Sehir',outputCol='indexedSehir')  ##linear modellerde bu kullanılması\n",
        "indexModel=indexer.fit(evlerDF)\n",
        "evlerDF = indexModel.transform(evlerDF)\n",
        "\n",
        "encoder = OneHotEncoder(inputCol='indexedSehir',outputCol='encodedSehir')\n",
        "encodeModel=encoder.fit(evlerDF)\n",
        "\n",
        "#evlerDF = encodeModel.transform(evlerDF)\n",
        "\n",
        "evlerDF= encoder.fit(evlerDF).transform(evlerDF)\n",
        "\n",
        "evlerDF = evlerDF.withColumnRenamed('fiyat','label') #label atama\n",
        "evlerDF = evlerDF.select('encodedSehir','Oda','salon','label')\n",
        "\n",
        "vec =VectorAssembler(inputCols=['encodedSehir','Oda','salon'],outputCol='features')\n",
        "evlerDF =vec.transform(evlerDF)\n",
        "#evlerDF.show()\n",
        "\n",
        "evlerDF = evlerDF.select('features','label')\n",
        "#evlerDF.show()\n",
        "\n",
        "lr = LinearRegression()\n",
        "\n",
        "model = lr.fit(evlerDF) ##Makine Öğrenmesinin yapıldığı satır\n",
        "#print('Bitti')\n",
        "\n",
        "print(\"Çarpanlar:  \",model.coefficients)\n",
        "print(\"Sabit sayi \", model.intercept)\n",
        "\n",
        "\n",
        "\n",
        "#fiyatı bilinmeyen evler\n",
        "\n",
        "yenilerDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',';').csv('evler-fiyati-bilinmeyen.csv')\n",
        "\n",
        "yenilerDF = indexModel.transform(yenilerDF)\n",
        "yenilerDF = encodeModel.transform(yenilerDF)\n",
        "yenilerDF = vec.transform(yenilerDF)\n",
        "#yenilerDF.show()\n",
        "\n",
        "\n",
        "\n",
        "yenilerDF = model.transform(yenilerDF) #Prediction yapıyoruz\n",
        "yenilerDF.show()"
      ],
      "execution_count": 67,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Çarpanlar:   [107857.14398708245,-42261.90458170463,215476.19109859318,0.0]\n",
            "Sabit sayi  -48690.477997543654\n",
            "+--------+---+-----+------------+-------------+-----------------+-----------------+\n",
            "|   Sehir|Oda|salon|indexedSehir| encodedSehir|         features|       prediction|\n",
            "+--------+---+-----+------------+-------------+-----------------+-----------------+\n",
            "|   izmir|  3|    1|         1.0|(2,[1],[1.0])|[0.0,1.0,3.0,1.0]|555476.1907165312|\n",
            "|istanbul|  3|    1|         0.0|(2,[0],[1.0])|[1.0,0.0,3.0,1.0]|705595.2392853183|\n",
            "|  ankara|  3|    1|         2.0|    (2,[],[])|[0.0,0.0,3.0,1.0]|597738.0952982359|\n",
            "+--------+---+-----+------------+-------------+-----------------+-----------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "IFj0NQLXdOqy",
        "outputId": "1099884c-c099-4687-9cd9-57c410b78e3f"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
        "from pyspark.ml.classification import DecisionTreeClassifier\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "irisDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',',').csv('iris-dataset.txt')\n",
        "#irisDF.show()\n",
        "\n",
        "#irisDF.describe().show()\n",
        "#irisDF.count()\n",
        "#irisDF.printSchema() #string'mi degilmi bakıyoruz\n",
        "\n",
        "#Label da dahil olmak üzere makine öğrenmesinde kullanılacak her kolon tipi sayısal olmalıdır.\n",
        "#Burada class (label) indexliyoruz\n",
        "\n",
        "indexer = StringIndexer(inputCol='class',outputCol='label')\n",
        "indexerModel = indexer.fit(irisDF) #Yeni kolonu eklemek için gerekli olan bilgi (kaç çiçek türü var) \n",
        "                                    #hesaplamak için tüm kolondaki saatırları tek tek işler\n",
        "\n",
        "irisDF = indexerModel.transform(irisDF)\n",
        "\n",
        "#irisDF.show()\n",
        "\n",
        "vec =VectorAssembler(inputCols=['sepal-length', 'sepal-width', 'petal-length', 'petal-width'],outputCol='features')\n",
        "\n",
        "irisDF = vec.transform(irisDF)\n",
        "#irisDF.show()\n",
        "\n",
        "irisDF = irisDF.select('features','label')\n",
        "irisDF.show(5)\n",
        "\n",
        "classifier = DecisionTreeClassifier()\n",
        "model = classifier.fit(irisDF) #Makine Öğrenmesini yapıldığı satır. Buna Model Training adı verilir.\n",
        "print(\"Makine öğrendi\")\n",
        "\n",
        "print(model.toDebugString)"
      ],
      "execution_count": 79,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------------+-----+\n",
            "|         features|label|\n",
            "+-----------------+-----+\n",
            "|[5.1,3.5,1.4,0.2]|  0.0|\n",
            "|[4.9,3.0,1.4,0.2]|  0.0|\n",
            "|[4.7,3.2,1.3,0.2]|  0.0|\n",
            "|[4.6,3.1,1.5,0.2]|  0.0|\n",
            "|[5.0,3.6,1.4,0.2]|  0.0|\n",
            "+-----------------+-----+\n",
            "only showing top 5 rows\n",
            "\n",
            "Makine öğrendi\n",
            "DecisionTreeClassificationModel: uid=DecisionTreeClassifier_a22ef284dbde, depth=5, numNodes=17, numClasses=3, numFeatures=4\n",
            "  If (feature 2 <= 2.45)\n",
            "   Predict: 0.0\n",
            "  Else (feature 2 > 2.45)\n",
            "   If (feature 3 <= 1.75)\n",
            "    If (feature 2 <= 4.95)\n",
            "     If (feature 3 <= 1.65)\n",
            "      Predict: 1.0\n",
            "     Else (feature 3 > 1.65)\n",
            "      Predict: 2.0\n",
            "    Else (feature 2 > 4.95)\n",
            "     If (feature 3 <= 1.55)\n",
            "      Predict: 2.0\n",
            "     Else (feature 3 > 1.55)\n",
            "      If (feature 0 <= 6.75)\n",
            "       Predict: 1.0\n",
            "      Else (feature 0 > 6.75)\n",
            "       Predict: 2.0\n",
            "   Else (feature 3 > 1.75)\n",
            "    If (feature 2 <= 4.85)\n",
            "     If (feature 0 <= 5.95)\n",
            "      Predict: 1.0\n",
            "     Else (feature 0 > 5.95)\n",
            "      Predict: 2.0\n",
            "    Else (feature 2 > 4.85)\n",
            "     Predict: 2.0\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "-BbLU35HfeM8",
        "outputId": "0c78f818-7419-450a-ef75-8794c3649a16"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
        "from pyspark.ml.classification import RandomForestClassifier\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "irisDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',',').csv('iris-dataset.txt')\n",
        "#irisDF.show()\n",
        "\n",
        "#irisDF.describe().show()\n",
        "#irisDF.count()\n",
        "#irisDF.printSchema() #string'mi degilmi bakıyoruz\n",
        "\n",
        "indexer = StringIndexer(inputCol='class',outputCol='label')\n",
        "indexerModel = indexer.fit(irisDF) #Yeni kolonu eklemek için gerekli olan bilgi (kaç çiçek türü var) \n",
        "                                    #hesaplamak için tüm kolondaki saatırları tek tek işler\n",
        "\n",
        "irisDF = indexerModel.transform(irisDF)\n",
        "\n",
        "#irisDF.show()\n",
        "\n",
        "vec =VectorAssembler(inputCols=['sepal-length', 'sepal-width', 'petal-length', 'petal-width'],outputCol='features')\n",
        "\n",
        "irisDF = vec.transform(irisDF)\n",
        "#irisDF.show()\n",
        "\n",
        "irisDF = irisDF.select('features','label')\n",
        "irisDF.show(5)\n",
        "\n",
        "classifier = RandomForestClassifier()\n",
        "model = classifier.fit(irisDF) #Makine Öğrenmesini yapıldığı satır. Buna Model Training adı verilir.\n",
        "print(\"Makine öğrendi\")\n",
        "print(model.featureImportances)\n"
      ],
      "execution_count": 80,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------------+-----+\n",
            "|         features|label|\n",
            "+-----------------+-----+\n",
            "|[5.1,3.5,1.4,0.2]|  0.0|\n",
            "|[4.9,3.0,1.4,0.2]|  0.0|\n",
            "|[4.7,3.2,1.3,0.2]|  0.0|\n",
            "|[4.6,3.1,1.5,0.2]|  0.0|\n",
            "|[5.0,3.6,1.4,0.2]|  0.0|\n",
            "+-----------------+-----+\n",
            "only showing top 5 rows\n",
            "\n",
            "Makine öğrendi\n",
            "(4,[0,1,2,3],[0.0824681606506779,0.013281914233677052,0.34766202244616545,0.5565879026694797])\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "DmFZPYNVgySz",
        "outputId": "ef2b23ab-45f9-4006-8839-cb2e8e55ad60"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
        "from pyspark.ml.classification import DecisionTreeClassifier\n",
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "irisDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',',').csv('iris-dataset.txt')\n",
        "\n",
        "indexer = StringIndexer(inputCol='class',outputCol='label')\n",
        "indexerModel = indexer.fit(irisDF) #Yeni kolonu eklemek için gerekli olan bilgi (kaç çiçek türü var) \n",
        "                                    #hesaplamak için tüm kolondaki saatırları tek tek işler\n",
        "\n",
        "irisDF = indexerModel.transform(irisDF)\n",
        "\n",
        "vec =VectorAssembler(inputCols=['sepal-length', 'sepal-width', 'petal-length', 'petal-width'],outputCol='features')\n",
        "irisDF = vec.transform(irisDF)\n",
        "irisDF = irisDF.select('features','label')\n",
        "#irisDF.show()\n",
        "\n",
        "\n",
        "trainDF, testDF = irisDF.randomSplit([0.75,0.25])\n",
        "classifier = DecisionTreeClassifier()\n",
        "model = classifier.fit(trainDF) #Makine Öğrenmesini yapıldığı satır. Buna Model Training adı verilir.\n",
        "print(\"Makine öğrendi\")\n",
        "\n",
        "#testDF.show()\n",
        "resultDF = model.transform(testDF) #Prediction\n",
        "resultDF.show()\n",
        "\n",
        "eva = MulticlassClassificationEvaluator(metricName='accuracy')\n",
        "result = eva.evaluate(resultDF)\n",
        "print(\"Accuracy : \",result)"
      ],
      "execution_count": 83,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Makine öğrendi\n",
            "+-----------------+-----+--------------+-------------+----------+\n",
            "|         features|label| rawPrediction|  probability|prediction|\n",
            "+-----------------+-----+--------------+-------------+----------+\n",
            "|[4.5,2.3,1.3,0.3]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[4.8,3.4,1.9,0.2]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[4.9,3.1,1.5,0.1]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.0,3.4,1.5,0.2]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.0,3.4,1.6,0.4]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.0,3.5,1.6,0.6]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.1,3.7,1.5,0.4]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.2,4.1,1.5,0.1]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.4,3.4,1.5,0.4]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.5,2.3,4.0,1.3]|  1.0|[0.0,39.0,0.0]|[0.0,1.0,0.0]|       1.0|\n",
            "|[5.5,4.2,1.4,0.2]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.6,2.9,3.6,1.3]|  1.0|[0.0,39.0,0.0]|[0.0,1.0,0.0]|       1.0|\n",
            "|[5.7,3.8,1.7,0.3]|  0.0|[39.0,0.0,0.0]|[1.0,0.0,0.0]|       0.0|\n",
            "|[5.8,2.7,3.9,1.2]|  1.0|[0.0,39.0,0.0]|[0.0,1.0,0.0]|       1.0|\n",
            "|[5.8,2.7,4.1,1.0]|  1.0|[0.0,39.0,0.0]|[0.0,1.0,0.0]|       1.0|\n",
            "|[5.9,3.0,4.2,1.5]|  1.0|[0.0,39.0,0.0]|[0.0,1.0,0.0]|       1.0|\n",
            "|[6.1,2.9,4.7,1.4]|  1.0|[0.0,39.0,0.0]|[0.0,1.0,0.0]|       1.0|\n",
            "|[6.3,2.5,5.0,1.9]|  2.0|[0.0,0.0,32.0]|[0.0,0.0,1.0]|       2.0|\n",
            "|[6.3,2.7,4.9,1.8]|  2.0|[0.0,0.0,32.0]|[0.0,0.0,1.0]|       2.0|\n",
            "|[6.3,3.3,6.0,2.5]|  2.0|[0.0,0.0,32.0]|[0.0,0.0,1.0]|       2.0|\n",
            "+-----------------+-----+--------------+-------------+----------+\n",
            "only showing top 20 rows\n",
            "\n",
            "Accuracy :  1.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "64UUUYV3nS5k",
        "outputId": "3eb8d6fe-f9fc-4388-c59f-a9d576dd8f3a"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
        "from pyspark.ml.classification import RandomForestClassifier\n",
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
        "\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
        "from pyspark.ml.classification import RandomForestClassifier\n",
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
        "\n",
        "spark = SparkSession.Builder().appName('Spark DataFrame Introduction').getOrCreate()\n",
        "irisDF = spark.read.option('inferSchema','true')\\\n",
        "                        .option('header','true')\\\n",
        "                        .option('delimiter',',').csv('iris-dataset.txt')\n",
        "\n",
        "indexer = StringIndexer(inputCol='class',outputCol='label')\n",
        "indexerModel = indexer.fit(irisDF) \n",
        "irisDF = indexerModel.transform(irisDF)\n",
        "\n",
        "vec =VectorAssembler(inputCols=['sepal-length', 'sepal-width', 'petal-length', 'petal-width'],outputCol='features')\n",
        "irisDF = vec.transform(irisDF)\n",
        "irisDF = irisDF.select('features','label')\n",
        "\n",
        "\n",
        "trainDF, testDF = irisDF.randomSplit([0.75,0.25],seed=123)\n",
        "\n",
        "\n",
        "classifier = RandomForestClassifier()\n",
        "model = classifier.fit(trainDF) #Makine Öğrenmesini yapıldığı satır. Buna Model Training adı verilir.\n",
        "print(\"Makine öğrendi\")\n",
        "\n",
        "testDF.show()\n",
        "resultDF = model.transform(testDF) #Prediction\n",
        "resultDF.show()\n",
        "\n",
        "eva = MulticlassClassificationEvaluator(metricName='accuracy')\n",
        "result = eva.evaluate(resultDF)\n",
        "print(\"Accuracy : \",result)"
      ],
      "execution_count": 89,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Makine öğrendi\n",
            "+-----------------+-----+\n",
            "|         features|label|\n",
            "+-----------------+-----+\n",
            "|[4.4,3.0,1.3,0.2]|  0.0|\n",
            "|[4.6,3.2,1.4,0.2]|  0.0|\n",
            "|[4.8,3.0,1.4,0.3]|  0.0|\n",
            "|[4.8,3.1,1.6,0.2]|  0.0|\n",
            "|[4.9,3.0,1.4,0.2]|  0.0|\n",
            "|[5.0,2.3,3.3,1.0]|  1.0|\n",
            "|[5.0,3.5,1.3,0.3]|  0.0|\n",
            "|[5.0,3.5,1.6,0.6]|  0.0|\n",
            "|[5.1,3.3,1.7,0.5]|  0.0|\n",
            "|[5.1,3.4,1.5,0.2]|  0.0|\n",
            "|[5.1,3.5,1.4,0.2]|  0.0|\n",
            "|[5.1,3.8,1.5,0.3]|  0.0|\n",
            "|[5.1,3.8,1.9,0.4]|  0.0|\n",
            "|[5.4,3.4,1.5,0.4]|  0.0|\n",
            "|[5.5,2.5,4.0,1.3]|  1.0|\n",
            "|[5.5,4.2,1.4,0.2]|  0.0|\n",
            "|[5.6,2.7,4.2,1.3]|  1.0|\n",
            "|[5.6,2.8,4.9,2.0]|  2.0|\n",
            "|[5.8,2.6,4.0,1.2]|  1.0|\n",
            "|[5.8,2.7,5.1,1.9]|  2.0|\n",
            "+-----------------+-----+\n",
            "only showing top 20 rows\n",
            "\n",
            "+-----------------+-----+--------------------+--------------------+----------+\n",
            "|         features|label|       rawPrediction|         probability|prediction|\n",
            "+-----------------+-----+--------------------+--------------------+----------+\n",
            "|[4.4,3.0,1.3,0.2]|  0.0|      [19.0,1.0,0.0]|     [0.95,0.05,0.0]|       0.0|\n",
            "|[4.6,3.2,1.4,0.2]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[4.8,3.0,1.4,0.3]|  0.0|      [19.0,1.0,0.0]|     [0.95,0.05,0.0]|       0.0|\n",
            "|[4.8,3.1,1.6,0.2]|  0.0|      [19.0,1.0,0.0]|     [0.95,0.05,0.0]|       0.0|\n",
            "|[4.9,3.0,1.4,0.2]|  0.0|      [19.0,1.0,0.0]|     [0.95,0.05,0.0]|       0.0|\n",
            "|[5.0,2.3,3.3,1.0]|  1.0|      [0.0,20.0,0.0]|       [0.0,1.0,0.0]|       1.0|\n",
            "|[5.0,3.5,1.3,0.3]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.0,3.5,1.6,0.6]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.1,3.3,1.7,0.5]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.1,3.4,1.5,0.2]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.1,3.5,1.4,0.2]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.1,3.8,1.5,0.3]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.1,3.8,1.9,0.4]|  0.0|      [17.0,3.0,0.0]|     [0.85,0.15,0.0]|       0.0|\n",
            "|[5.4,3.4,1.5,0.4]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.5,2.5,4.0,1.3]|  1.0|[0.0,19.856821589...|[0.0,0.9928410794...|       1.0|\n",
            "|[5.5,4.2,1.4,0.2]|  0.0|      [20.0,0.0,0.0]|       [1.0,0.0,0.0]|       0.0|\n",
            "|[5.6,2.7,4.2,1.3]|  1.0|[0.0,19.856821589...|[0.0,0.9928410794...|       1.0|\n",
            "|[5.6,2.8,4.9,2.0]|  2.0|[0.0,0.0256410256...|[0.0,0.0012820512...|       2.0|\n",
            "|[5.8,2.6,4.0,1.2]|  1.0|[0.0,19.856821589...|[0.0,0.9928410794...|       1.0|\n",
            "|[5.8,2.7,5.1,1.9]|  2.0|[0.0,0.0256410256...|[0.0,0.0012820512...|       2.0|\n",
            "+-----------------+-----+--------------------+--------------------+----------+\n",
            "only showing top 20 rows\n",
            "\n",
            "Accuracy :  0.9696969696969697\n"
          ]
        }
      ]
    }
  ]
}