9주차_PySpark_기본_3일차_2.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "PySpark 기본: 3일차-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_3%EC%9D%BC%EC%B0%A8_2.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": "90f1ae80-8594-48a7-db73-08134632ae77"
      },
      "source": [
        "!pip install pyspark==3.0.1 py4j==0.10.9 "
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Collecting pyspark==3.0.1\n",
            "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/f0/26/198fc8c0b98580f617cb03cb298c6056587b8f0447e20fa40c5b634ced77/pyspark-3.0.1.tar.gz (204.2MB)\n",
            "\u001b[K     |████████████████████████████████| 204.2MB 69kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.9\n",
            "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/9e/b6/6a4fb90cd235dc8e265a6a2067f2a2c99f0d91787f06aca4bcf7c23f3f80/py4j-0.10.9-py2.py3-none-any.whl (198kB)\n",
            "\u001b[K     |████████████████████████████████| 204kB 66.5MB/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.0.1-py2.py3-none-any.whl size=204612242 sha256=fa4d5c38d2f852cc0734fcbc4d189db86c682487c1c76ddb360c9610579bddbb\n",
            "  Stored in directory: /root/.cache/pip/wheels/5e/bd/07/031766ca628adec8435bb40f0bd83bb676ce65ff4007f8e73f\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.9 pyspark-3.0.1\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1iHGbrRlA01u"
      },
      "source": [
        "지금부터 실습은 Redshift에 있는 데이터를 가지고 해볼 예정이고 그래서 Redshift 관련 JAR 파일을 설치해야함"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "OdTTZZqVA0Xa",
        "outputId": "a9f0438c-dcb1-43af-fa9c-43259978d549"
      },
      "source": [
        "!cd /usr/local/lib/python3.6/dist-packages/pyspark/jars && wget https://s3.amazonaws.com/redshift-downloads/drivers/jdbc/1.2.20.1043/RedshiftJDBC42-no-awssdk-1.2.20.1043.jar"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "--2021-01-31 18:49:38--  https://s3.amazonaws.com/redshift-downloads/drivers/jdbc/1.2.20.1043/RedshiftJDBC42-no-awssdk-1.2.20.1043.jar\n",
            "Resolving s3.amazonaws.com (s3.amazonaws.com)... 52.216.24.70\n",
            "Connecting to s3.amazonaws.com (s3.amazonaws.com)|52.216.24.70|:443... connected.\n",
            "HTTP request sent, awaiting response... 200 OK\n",
            "Length: 2413910 (2.3M) [application/java-archive]\n",
            "Saving to: ‘RedshiftJDBC42-no-awssdk-1.2.20.1043.jar’\n",
            "\n",
            "RedshiftJDBC42-no-a 100%[===================>]   2.30M  3.85MB/s    in 0.6s    \n",
            "\n",
            "2021-01-31 18:49:40 (3.85 MB/s) - ‘RedshiftJDBC42-no-awssdk-1.2.20.1043.jar’ saved [2413910/2413910]\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Ew_eTGrvXlDw"
      },
      "source": [
        "**Spark Session:** 이번 SparkSession은 spark.jars를 통해 앞서 다운로드받은 Redshift 연결을 위한 JDBC 드라이버를 사용함 (.config(\"spark.jars\", ...)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3vm6tgcPXdnR"
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "\n",
        "spark = SparkSession \\\n",
        "    .builder \\\n",
        "    .appName(\"Python Spark SQL basic example\") \\\n",
        "    .config(\"spark.jars\", \"/usr/local/lib/python3.6/dist-packages/pyspark/jars/RedshiftJDBC42-no-awssdk-1.2.20.1043.jar\") \\\n",
        "    .getOrCreate()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 217
        },
        "id": "LSs_1PYaYWxI",
        "outputId": "409bf417-b320-4c0c-dcad-5899be79e759"
      },
      "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://cf78351d33b3: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>Python Spark SQL basic example</code></dd>\n",
              "            </dl>\n",
              "        </div>\n",
              "        \n",
              "            </div>\n",
              "        "
            ],
            "text/plain": [
              "<pyspark.sql.session.SparkSession at 0x7fcf2340de10>"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 4
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "H7Wl1fgP--_n"
      },
      "source": [
        "# **SparkSQL 맛보기**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XSqSK6-T3SlS"
      },
      "source": [
        "판다스로 일단 CSV 파일 하나 로드하기"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MlppT68QCorD"
      },
      "source": [
        "import pandas as pd\n",
        "\n",
        "namegender_pd = pd.read_csv(\"https://s3-geospatial.s3-us-west-2.amazonaws.com/name_gender.csv\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 203
        },
        "id": "21mud2c_CpoT",
        "outputId": "161240c7-c3ef-4426-d5f1-c0f674158d7a"
      },
      "source": [
        "namegender_pd.head()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/html": [
              "<div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>name</th>\n",
              "      <th>gender</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>Adaleigh</td>\n",
              "      <td>F</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>Amryn</td>\n",
              "      <td>Unisex</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>Apurva</td>\n",
              "      <td>Unisex</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>Aryion</td>\n",
              "      <td>M</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>Alixia</td>\n",
              "      <td>F</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "       name  gender\n",
              "0  Adaleigh       F\n",
              "1     Amryn  Unisex\n",
              "2    Apurva  Unisex\n",
              "3    Aryion       M\n",
              "4    Alixia       F"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 6
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 172
        },
        "id": "hGF0RDpI1nFk",
        "outputId": "c5e5deba-df0d-4ba6-d50a-70b14ad8e3db"
      },
      "source": [
        "namegender_pd.groupby([\"gender\"]).count()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/html": [
              "<div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>name</th>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>gender</th>\n",
              "      <th></th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>F</th>\n",
              "      <td>65</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>M</th>\n",
              "      <td>28</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>Unisex</th>\n",
              "      <td>7</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ],
            "text/plain": [
              "        name\n",
              "gender      \n",
              "F         65\n",
              "M         28\n",
              "Unisex     7"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 7
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6YpBBzy11wwV"
      },
      "source": [
        "판다스 데이터프레임을 Spark 데이터프레임으로 변환하기"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eBXgab2KCsUV"
      },
      "source": [
        "namegender_df = spark.createDataFrame(namegender_pd)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "oUlABaYtCvKC",
        "outputId": "f82e52fc-e91f-42f9-88c9-f8794e9bc741"
      },
      "source": [
        "namegender_df.printSchema()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- name: string (nullable = true)\n",
            " |-- gender: string (nullable = true)\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "U9oPMiJRCy4R",
        "outputId": "cbd93fda-9be8-484e-a8dc-c086d54fc5ec"
      },
      "source": [
        "namegender_df.show()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+----------+------+\n",
            "|      name|gender|\n",
            "+----------+------+\n",
            "|  Adaleigh|     F|\n",
            "|     Amryn|Unisex|\n",
            "|    Apurva|Unisex|\n",
            "|    Aryion|     M|\n",
            "|    Alixia|     F|\n",
            "|Alyssarose|     F|\n",
            "|    Arvell|     M|\n",
            "|     Aibel|     M|\n",
            "|   Atiyyah|     F|\n",
            "|     Adlie|     F|\n",
            "|    Anyely|     F|\n",
            "|    Aamoni|     F|\n",
            "|     Ahman|     M|\n",
            "|    Arlane|     F|\n",
            "|   Armoney|     F|\n",
            "|   Atzhiry|     F|\n",
            "| Antonette|     F|\n",
            "|   Akeelah|     F|\n",
            "| Abdikadir|     M|\n",
            "|    Arinze|     M|\n",
            "+----------+------+\n",
            "only showing top 20 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "bt432nZD13md",
        "outputId": "a22be05a-ed1c-412c-e769-a55829fcc339"
      },
      "source": [
        "namegender_df.groupBy([\"gender\"]).count().collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[Row(gender='F', count=65),\n",
              " Row(gender='M', count=28),\n",
              " Row(gender='Unisex', count=7)]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 12
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Dj4z2JTS-ykt"
      },
      "source": [
        "# https://towardsdatascience.com/pyspark-and-sparksql-basics-6cb4bf967e53"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PJmaTT5619bl"
      },
      "source": [
        "데이터프레임을 테이블뷰로 만들어서 SparkSQL로 처리해보기"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "wCHeqrAW6n-E"
      },
      "source": [
        "namegender_df.createOrReplaceTempView(\"namegender\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ghlHnLJv-Xie"
      },
      "source": [
        "namegender_group_df = spark.sql(\"SELECT gender, count(1) FROM namegender GROUP BY 1\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "LbbfGZHP-y4w",
        "outputId": "20818259-f879-45c3-a168-97f226867453"
      },
      "source": [
        "namegender_group_df.collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[Row(gender='F', count(1)=65),\n",
              " Row(gender='M', count(1)=28),\n",
              " Row(gender='Unisex', count(1)=7)]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 28
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_vqm-DsuH_hO"
      },
      "source": [
        "Redshift와 연결해서 테이블들을 데이터프레임으로 로딩하기"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "EEaNuN3xKNXm"
      },
      "source": [
        "df_user_session_channel = spark.read \\\n",
        "    .format(\"jdbc\") \\\n",
        "    .option(\"driver\", \"com.amazon.redshift.jdbc42.Driver\") \\\n",
        "    .option(\"url\", \"jdbc:redshift://learnde.cduaw970ssvt.ap-northeast-2.redshift.amazonaws.com:5439/prod?user=guest&password=Guest1!*\") \\\n",
        "    .option(\"dbtable\", \"raw_data.user_session_channel\") \\\n",
        "    .load()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IQlBPcO_INv9"
      },
      "source": [
        "df_session_timestamp = spark.read \\\n",
        "    .format(\"jdbc\") \\\n",
        "    .option(\"driver\", \"com.amazon.redshift.jdbc42.Driver\") \\\n",
        "    .option(\"url\", \"jdbc:redshift://learnde.cduaw970ssvt.ap-northeast-2.redshift.amazonaws.com:5439/prod?user=guest&password=Guest1!*\") \\\n",
        "    .option(\"dbtable\", \"raw_data.session_timestamp\") \\\n",
        "    .load()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cb87SiYQ3lX8"
      },
      "source": [
        "df_user_session_channel.createOrReplaceTempView(\"user_session_channel\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "LFMFrRpP3sP6"
      },
      "source": [
        "df_session_timestamp.createOrReplaceTempView(\"session_timestamp\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GdiSxs1030G0"
      },
      "source": [
        "channel_count_df = spark.sql(\"\"\"\n",
        "    SELECT channel, count(distinct userId) uniqueUsers\n",
        "    FROM session_timestamp st\n",
        "    JOIN user_session_channel usc ON st.sessionID = usc.sessionID\n",
        "    GROUP BY 1\n",
        "    ORDER BY 1\n",
        "\"\"\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "z1C5HKcd61_j",
        "outputId": "096f5a6c-0a15-41e5-e6d4-6959947f33d2"
      },
      "source": [
        "channel_count_df"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "DataFrame[channel: string, uniqueUsers: bigint]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 24
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "whIte26D63hG",
        "outputId": "1cb90192-3cd0-4016-fa31-b4d6d8325780"
      },
      "source": [
        "channel_count_df.show()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+---------+-----------+\n",
            "|  channel|uniqueUsers|\n",
            "+---------+-----------+\n",
            "| Facebook|        889|\n",
            "|   Google|        893|\n",
            "|Instagram|        895|\n",
            "|    Naver|        882|\n",
            "|  Organic|        895|\n",
            "|  Youtube|        889|\n",
            "+---------+-----------+\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "vE8iL4vy6705"
      },
      "source": [
        "channel_with_o_count_df = spark.sql(\"\"\"\n",
        "    SELECT COUNT(1)\n",
        "    FROM user_session_channel\n",
        "    WHERE channel like '%o%'\n",
        "\"\"\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "QU44opcvDE54",
        "outputId": "941915df-9020-4f70-d768-10b5c607b693"
      },
      "source": [
        "channel_with_o_count_df.collect()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[Row(count(1)=50864)]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 27
        }
      ]
    }
  ]
}