11주차_PySpark_기본_2일차.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "PySpark 기본: 2일차",
      "provenance": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/learn-programmers/programmers_kdt_II/blob/main/9%EC%A3%BC%EC%B0%A8_PySpark_%EA%B8%B0%EB%B3%B8_2%EC%9D%BC%EC%B0%A8.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xAqhTDfuWrcM"
      },
      "source": [
        "PySpark을 로컬머신에 설치하고 노트북을 사용하기 보다는 머신러닝 관련 다양한 라이브러리가 이미 설치되었고 좋은 하드웨어를 제공해주는 Google Colab을 통해 실습을 진행한다."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XIA23YgbXKJd"
      },
      "source": [
        "이를 위해 pyspark과 Py4J 패키지를 설치한다. Py4J 패키지는 파이썬 프로그램이 자바가상머신상의 오브젝트들을 접근할 수 있게 해준다. Local Standalone Spark을 사용한다."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "NbT0rpGfVdiq",
        "outputId": "32661d29-d279-4542-b6b4-a294c676da5c"
      },
      "source": [
        "!pip install pyspark==3.0.1 py4j==0.10.9 "
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Requirement already satisfied: pyspark==3.0.1 in /usr/local/lib/python3.6/dist-packages (3.0.1)\n",
            "Requirement already satisfied: py4j==0.10.9 in /usr/local/lib/python3.6/dist-packages (0.10.9)\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "tQIB8vLPBMeu",
        "outputId": "058ae550-0d54-4825-ebf9-06f761184805"
      },
      "source": [
        "!ls -tl"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "total 4\n",
            "drwxr-xr-x 1 root root 4096 Jan 20 17:27 sample_data\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "NHXY3xYctlCe",
        "outputId": "0441f11c-dc75-43e6-c66e-512bedfdf008"
      },
      "source": [
        "!ls -tl sample_data"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "total 55504\n",
            "-rw-r--r-- 1 root root 18289443 Jan 20 17:27 mnist_test.csv\n",
            "-rw-r--r-- 1 root root 36523880 Jan 20 17:27 mnist_train_small.csv\n",
            "-rw-r--r-- 1 root root   301141 Jan 20 17:27 california_housing_test.csv\n",
            "-rw-r--r-- 1 root root  1706430 Jan 20 17:27 california_housing_train.csv\n",
            "-rwxr-xr-x 1 root root     1697 Jan  1  2000 anscombe.json\n",
            "-rwxr-xr-x 1 root root      930 Jan  1  2000 README.md\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Ew_eTGrvXlDw"
      },
      "source": [
        "**Spark Session:** SparkSession은 Spark 2.0부터 엔트리 포인트로 사용된다. 그 이전에는 SparkContext가 사용되었다. SparkSession을 이용해 RDD, 데이터 프레임등을 만든다. SparkSession은 SparkSession.builder를 호출하여 생성하며 다양한 함수들을 통해 세부 설정이 가능하다"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3vm6tgcPXdnR"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "\n",
        "spark = SparkSession.builder\\\n",
        "        .master(\"local[*]\")\\\n",
        "        .appName('PySpark_Tutorial')\\\n",
        "        .getOrCreate()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 214
        },
        "id": "LSs_1PYaYWxI",
        "outputId": "0f2ac99a-c9f5-4524-d9cb-8ea24e94f627"
      },
      "source": [
        "spark"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/html": [
              "\n",
              "            <div>\n",
              "                <p><b>SparkSession - in-memory</b></p>\n",
              "                \n",
              "        <div>\n",
              "            <p><b>SparkContext</b></p>\n",
              "\n",
              "            <p><a href=\"http://6f561d346705:4040\">Spark UI</a></p>\n",
              "\n",
              "            <dl>\n",
              "              <dt>Version</dt>\n",
              "                <dd><code>v3.0.1</code></dd>\n",
              "              <dt>Master</dt>\n",
              "                <dd><code>local[*]</code></dd>\n",
              "              <dt>AppName</dt>\n",
              "                <dd><code>PySpark_Tutorial</code></dd>\n",
              "            </dl>\n",
              "        </div>\n",
              "        \n",
              "            </div>\n",
              "        "
            ],
            "text/plain": [
              "<pyspark.sql.session.SparkSession at 0x7ff2ef8585c0>"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 22
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VNHAAJuCZCqE"
      },
      "source": [
        "**Python 객체를 RDD로 변환해보기**\n",
        "\n",
        "**1> Python 리스트 생성**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "HhijTd1UY1i9"
      },
      "source": [
        "name_list_json = [ '{\"name\": \"keeyong\"}', '{\"name\": \"benjamin\"}', '{\"name\": \"claire\"}' ]"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "hXqf1lC_Zdxf",
        "outputId": "553d372c-ea45-4b8a-f5ed-61c8032e9308"
      },
      "source": [
        "for n in name_list_json:\n",
        "  print(n)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "{\"name\": \"keeyong\"}\n",
            "{\"name\": \"benjamin\"}\n",
            "{\"name\": \"claire\"}\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "f1_A6f6Nbr57",
        "outputId": "d4339f51-e7e6-483b-c13d-5091023918ae"
      },
      "source": [
        "import json\n",
        "\n",
        "for n in name_list_json:\n",
        "  jn = json.loads(n)\n",
        "  print(jn[\"name\"])"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "keeyong\n",
            "benjamin\n",
            "claire\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YKlanTznb75N"
      },
      "source": [
        "**2> 파이썬 리스트를 RDD로 변환. RDD로 변환되는 순간 Spark 클러스터의 서버들에 데이터가 나눠 저장됨 (파티션). 또한 Lazy Execution이 된다는 점 기억**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "NvTOpLsrZ_I8"
      },
      "source": [
        "rdd = spark.sparkContext.parallelize(name_list_json)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "KIMTLkPdk4ul",
        "outputId": "1bea447d-f88b-43f3-b262-77c8263c615a"
      },
      "source": [
        "rdd"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "ParallelCollectionRDD[4] at readRDDFromFile at PythonRDD.scala:262"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 27
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "KmszVYtzaL3Q",
        "outputId": "2d9d3f51-142f-4b95-f931-a60d4da466f1"
      },
      "source": [
        "rdd.count()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "3"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 28
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "gMLop2SQbMnZ"
      },
      "source": [
        "parsed_rdd = rdd.map(lambda el:json.loads(el))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "RAgNR7EWcl2t",
        "outputId": "a7bc7b08-9c96-46ab-fe9c-d3674dafc398"
      },
      "source": [
        "parsed_rdd"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "PythonRDD[6] at RDD at PythonRDD.scala:53"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 30
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "qt5SMf4IcoIZ",
        "outputId": "230bea4b-cdc5-41c5-eda1-b8fe87cba814"
      },
      "source": [
        "parsed_rdd.collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[{'name': 'keeyong'}, {'name': 'benjamin'}, {'name': 'claire'}]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 31
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "T1Kmn_qycqQl"
      },
      "source": [
        "parsed_name_rdd = rdd.map(lambda el:json.loads(el)[\"name\"])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "4ZWxTKgJnDm8",
        "outputId": "f10bca80-db64-4055-bd1f-bbab948147b2"
      },
      "source": [
        "parsed_name_rdd.collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "['keeyong', 'benjamin', 'claire']"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 34
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "McGKfLtWnsTB"
      },
      "source": [
        "**파이썬 리스트를 데이터프레임으로 변환하기**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_WlhiRsKo7j7"
      },
      "source": [
        "from pyspark.sql.types import StringType\n",
        "\n",
        "df = spark.createDataFrame(name_list_json, StringType())"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "S2-pyiKGpLZe",
        "outputId": "04fe945a-2395-4f6f-ed38-a4266a04f584"
      },
      "source": [
        "df.count()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "3"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 36
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "9Sql4KPhppgr",
        "outputId": "d9c659e8-193a-4a64-e13b-cfe7d6494b96"
      },
      "source": [
        "df.printSchema()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- value: string (nullable = true)\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "U2XzTjdEpuZ4",
        "outputId": "5145ec97-4851-4e72-ac8f-4f2921d65064"
      },
      "source": [
        "df.select('*').collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[Row(value='{\"name\": \"keeyong\"}'),\n",
              " Row(value='{\"name\": \"benjamin\"}'),\n",
              " Row(value='{\"name\": \"claire\"}')]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 38
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "SmLv4Guwny00",
        "outputId": "5072cc74-6e70-4651-c6aa-451b6a3c86c4"
      },
      "source": [
        "df.select('value').collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[Row(value='{\"name\": \"keeyong\"}'),\n",
              " Row(value='{\"name\": \"benjamin\"}'),\n",
              " Row(value='{\"name\": \"claire\"}')]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 39
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "93TLUFxgqLrm"
      },
      "source": [
        "from pyspark.sql import Row\n",
        "\n",
        "row = Row(\"name\") # Or some other column name\n",
        "df_name = parsed_name_rdd.map(row).toDF()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "LtFEYRG61-L8",
        "outputId": "ceea17ca-9524-459b-f35a-e6805230d2da"
      },
      "source": [
        "df_name.printSchema()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- name: string (nullable = true)\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "MptOIByKsLeG",
        "outputId": "476cbc99-a4e8-4b85-88d4-a346def6bd53"
      },
      "source": [
        "df_name.select('name').collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[Row(name='keeyong'), Row(name='benjamin'), Row(name='claire')]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 41
        }
      ]
    }
  ]
}