notebook/analyze_exported.ipynb

﻿{
  "metadata": {
    "name": "analyze",
    "kernelspec": {
      "language": "scala",
      "name": "spark2-scala"
    },
    "language_info": {
      "codemirror_mode": "text/x-scala",
      "file_extension": ".scala",
      "mimetype": "text/x-scala",
      "name": "scala",
      "pygments_lexer": "scala"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2,
  "cells": [
    {
      "cell_type": "code",
      "execution_count": 0,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%pyspark\nsc.version"
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%pyspark\n\ndf_session_ts \u003d spark.read.format(\"jdbc\")\\\n    .option(\"dbtable\", \"session_timestamps\") \\\n    .load()\n\n#df로 import하기\ndf_session_tx \u003d spark.read.format(\"jdbc\")\\\n    .option(\"dbtable\", \"session_transactions\") \\\n    .load()\n    \ndf_channel \u003d spark.read.format(\"jdbc\")\\\n    .option(\"dbtable\", \"channels\") \\\n    .load()\n    \ndf_usc \u003d spark.read.format(\"jdbc\")\\\n    .option(\"dbtable\", \"user_session_channels\") \\\n    .load()\n\n\n# 전체 dataframe 출력\n# df_session_ts.show()|\n\n# explain으로 실행계획 확인하기\n# df_session_ts.select(\"ts\").explain()\n\ndf_usc.limit(1).show()\ndf_session_ts.limit(1).show()\ndf_session_tx.limit(1).show()\ndf_channel.limit(1).show()\n\n#SQL에 사용할 TempView를 만들기\ndf_usc.createOrReplaceTempView(\"usc\")\ndf_session_ts.createOrReplaceTempView(\"st\")\ndf_session_tx.createOrReplaceTempView(\"tx\")\ndf_channel.createOrReplaceTempView(\"ch\")\n\n\ndf_session_ts.show()"
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%pyspark\nfrom pyspark.sql.functions import count\n\nz.show(df_session_ts.\\\n    alias(\"st\").\\\n    join(df_usc.alias(\"usc\"), df_session_ts.sessionid \u003d\u003d df_usc.sessionid, how\u003d\u0027left\u0027).\\\n    groupby(\u0027st.timestamp\u0027).\\\n    agg(count(\u0027usc.userid\u0027))\n)\n\n"
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%pyspark\n\nz.show(spark.sql(\"\"\"\nSELECT usc.userid as uid, SUM(IF(tx.refunded \u003d false, tx.amount, 0)) as total\nFROM tx\nLEFT JOIN usc\nON tx.sessionid \u003d usc.sessionid\nGROUP BY 1\nORDER BY 2 DESC\nLIMIT 10\n\"\"\"))\n\n"
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%pyspark\n\nz.show(spark.sql(\"\"\"\nSELECT ch.channelname, COUNT(usc.userid) as count\nFROM ch\nLEFT JOIN usc\nON ch.channelname \u003d usc.channel\nGROUP BY ch.channelname\nORDER BY 2 desc\n\"\"\"))\n"
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%sql\n\nSELECT \n    DATE_FORMAT(st.timestamp,\"yyyy-MM\") as month, \n    channel, \n    count(distinct usc.userid) as uniqueUsers,\n    count(amount) as paidUsers,\n    CONCAT(IF(count(amount) \u003d 0, 0,round((count(amount) / count(distinct usc.userid)) * 100,1)),\u0027%\u0027) as conversionRate,\n    SUM(amount) as growthRevenue,\n    SUM(IF(refunded \u003d FALSE ,amount, 0)) as netRevenue\nFROM ch\nLEFT JOIN usc\non ch.channelname \u003d usc.channel\nLEFT JOIN st\non st.sessionid \u003d usc.sessionid\nLEFT JOIN tx\non tx.sessionid \u003d usc.sessionid\nWHERE channel IS NOT NULL\nGROUP BY 1, 2\nORDER BY 1, 2\n"
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%pyspark\nfrom pyspark.sql.functions import months_between, countDistinct, col\n\ncohorts \u003d spark.sql(\"\"\"\nSELECT \n    cohort_month,\n    months_between(visit_month,cohort_month) as month_diff,\n    count(distinct cohort.userid) as unique_users\nFROM (\n    SELECT userid, MIN(DATE_FORMAT(timestamp,\u0027yyyy-MM\u0027)) as cohort_month\n    FROM usc\n    JOIN st ON st.sessionid \u003d usc.sessionid\n    GROUP BY 1\n) as cohort \nLEFT JOIN (\n    SELECT distinct userid, DATE_FORMAT(timestamp,\u0027yyyy-MM\u0027) as visit_month\n    FROM usc\n    JOIN st ON st.sessionid \u003d usc.sessionid    \n) as visit\nON cohort.cohort_month \u003c\u003d visit.visit_month AND cohort.userid \u003d visit.userid\nGROUP BY 1,2\nORDER BY 1,2\n\"\"\")\n# cohorts.show()\nz.show(cohorts.groupBy(\"cohort_month\").pivot(\"month_diff\").agg({\"unique_users\":\"sum\"}).orderBy(\"cohort_month\"))"
    },
    {
      "cell_type": "raw",
      "metadata": {
        "format": "text/plain"
      },
      "source": "%pyspark\n"
    }
  ]
}