old/5a-exploring-dataset-distributions-for-hours-diff-count_idalv_lineid.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Descrição\n",
    "# Este script faz uma análise exploratório do dataset processado com os registros dentro da \n",
    "# área de são paulo e que estão entre 6 e 22:59:59. Analisa diferença entre as horas, e para cada quarters %.\n",
    "# O programa também conta quanto há de eventos 64 e 0 distintamente para cada dia, \n",
    "# quantas linhas e id_avl disitndos há para cada dia\n",
    "#\n",
    "# Description\n",
    "# This script analyzes the processed dataset (register in Sao Paulo region between 6-23 hours), \n",
    "# the hours diff distribution by quarters\n",
    "# Also counts the how many lines_id and id_avl for each day."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Spark Config\n",
    "from pyspark import SparkConf\n",
    "from pyspark import SparkContext\n",
    "\n",
    "spark_conf = (SparkConf().set(\"spark.speculation\", \"false\"))\n",
    "sc = SparkContext.getOrCreate(conf = spark_conf)\n",
    "\n",
    "# spark = sparkSession\n",
    "hadoop_conf = spark._jsc.hadoopConfiguration()\n",
    "hadoop_conf.set(\"fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")\n",
    "hadoop_conf.set(\"mapreduce.fileoutputcommitter.algorithm.version\",\"2\")\n",
    "\n",
    "# installing necessary packages for notebook session\n",
    "sc.install_pypi_package(\"boto3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as F\n",
    "import boto3\n",
    "\n",
    "# new file header\n",
    "csv_out = \"day,line_id_distinct,id_avl_distinct,line-id_id-avl_distinct,event_0_count,event_64_count,id_point_distinct,hour_diff_mean,hour_diff_min,hour_diff_max,hour_diff_stddev,hour_diff_quantile_25,hour_diff_quantile_50,hour_diff_quantile_75\\n\"\n",
    "\n",
    "# from october 1 to october 31 \n",
    "for day in range(1,32):\n",
    "\n",
    "    # reading files\n",
    "    traces = spark.read.parquet(\"s3a://mobility-traces-sp/processed-data/using-server-hour/records-between-6-23-only-sp-server-hour/MO_1510\"+str(day) +  \"/\")\n",
    "\n",
    "    # counting distinct line_id\n",
    "    line_id_distinct = traces.agg(F.countDistinct(\"line_id\")).collect()[0][0]\n",
    "\n",
    "    # counting distinct id_avl\n",
    "    id_avl_distinct = traces.agg(F.countDistinct(\"id_avl\")).collect()[0][0]\n",
    "\n",
    "    # counting distinct pairs line_id,id_avl\n",
    "    lines_avl_distinct = traces.select('line_id','id_avl').distinct().count()\n",
    "\n",
    "    # counting the numbers of registers with 0 in field \"event\"\n",
    "    number_0 = traces.filter(\"event == 0\").count()\n",
    "\n",
    "    # counting the numbers of registers with 64 in field \"event\"\n",
    "    number_64 = traces.filter(\"event == 64\").count()\n",
    "\n",
    "    # counting the numbers of distinct id_point\n",
    "    id_point_distinct = traces.agg(F.countDistinct(\"id_point\")).collect()[0][0]\n",
    "\n",
    "    # getting hour_diff info (mean,min,max,stddev)\n",
    "    hour_diff_info = traces.agg(F.mean('hour_diff').alias('mean'),\n",
    "                       F.min('hour_diff').alias('min'),\n",
    "                       F.max('hour_diff').alias('max'),\n",
    "                       F.stddev('hour_diff').alias(\"stddev\")).collect()\n",
    "\n",
    "    # getting hour_diff quantiles (25%, 50%, 75%)\n",
    "    # 0.0001 is the precision\n",
    "    hour_diff_quantile = traces.approxQuantile(\"hour_diff\", [0.25,0.5,0.75], 0.0001)\n",
    "\n",
    "    csv_out += \"MO_1510\" + str(day) +\",\"+ str(line_id_distinct) + \",\"  + str(id_avl_distinct)+ \",\" + str(lines_avl_distinct) + \",\" + str(number_0) + \",\" + str(number_64) + \",\" + str(id_point_distinct)+ \",\" + str(hour_diff_info[0][\"mean\"]) + \",\" + str(hour_diff_info[0][\"min\"]) + \",\" + str(hour_diff_info[0][\"max\"]) + \",\" + str(hour_diff_info[0][\"stddev\"]) + \",\" + str(hour_diff_quantile[0]) + \",\" + str(hour_diff_quantile[1]) + \",\" + str(hour_diff_quantile[2]) + \"\\n\"\n",
    "s3 = boto3.client('s3')\n",
    "\n",
    "# writing results in S3\n",
    "s3.put_object(Body=bytes(csv_out,\"utf-8\"), Bucket='mobility-traces-sp', Key='statistics/exploring-data/1-general-statistics.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Another way to extract quantiles of data --> using spark sql approx percentile\n",
    "# from pyspark.sql import SQLContext\n",
    "# sqlContext = SQLContext(sc)\n",
    "# traces.registerTempTable(\"df\")\n",
    "# df = sqlContext.sql(\"SELECT approx_percentile(hour_diff, 0.25,1000000) FROM df\").collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# rounding data 2 fields after comma to fits in excel precision \n",
    "from pyspark.sql.types import *\n",
    "\n",
    "# schema of the raw data\n",
    "custom_schema = StructType([\n",
    "    StructField(\"day\", StringType()),\n",
    "    StructField(\"line_id_distinct\", IntegerType()),\n",
    "    StructField(\"id_avl_distinct\", IntegerType()),\n",
    "    StructField(\"line-id_id-avl_distinct\", IntegerType()),\n",
    "    StructField(\"event_0_count\", IntegerType()),\n",
    "    StructField(\"event_64_count\", IntegerType()),\n",
    "    StructField(\"hour_diff_mean\", DoubleType()),\n",
    "    StructField(\"hour_diff_min\", DoubleType()),\n",
    "    StructField(\"hour_diff_max\", DoubleType()),\n",
    "    StructField(\"hour_diff_stddev\", DoubleType()),\n",
    "    StructField(\"hour_diff_quantile_25\", DoubleType()),\n",
    "    StructField(\"hour_diff_quantile_50\", DoubleType()),\n",
    "    StructField(\"hour_diff_quantile_75\", DoubleType())\n",
    "])\n",
    "\n",
    "# reading file\n",
    "general_stats = spark.read.csv(\"s3a://mobility-traces-sp/statistics/exploring-data/1-general-statistics.csv\",header=\"true\",schema=custom_schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Adapting file with float with 2 fields, because of excel visualization support \n",
    "import numpy as np\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "# reading file\n",
    "lines_rdd = sc.textFile(\"s3a://mobility-traces-sp/statistics/exploring-data/1-general-statistics.csv\")\n",
    "\n",
    "#\n",
    "# people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))\n",
    "\n",
    "# # Infer the schema, and register the DataFrame as a table.\n",
    "# schemaPeople = spark.createDataFrame(people)\n",
    "# schemaPeople.createOrReplaceTempView(\"people\")\n",
    "\n",
    "# reading first line aka header\n",
    "header = lines_rdd.first()\n",
    "\n",
    "# eliminating header\n",
    "new_lines_rdd = lines_rdd.filter(lambda line: line != header)\n",
    "\n",
    "# splitting lines by ,\n",
    "splitted = new_lines_rdd.map(lambda l: l.split(\",\"))\n",
    "\n",
    "# transforming data with map lambda function\n",
    "transformed = splitted.map(lambda l: (l[0],l[1],l[2],l[3],l[4],l[5],l[6],float(np.round(float(l[7]),2)),float(np.round(float(l[8]),2)),float(np.round(float(l[9]),2)),float(np.round(float(l[10]),2)),float(np.round(float(l[11]),2)),float(np.round(float(l[12]),2)),float(np.round(float(l[13]),2))))\n",
    "\n",
    "# creating dataframe from the new transformed data\n",
    "new_df = spark.createDataFrame(transformed)\n",
    "\n",
    "# file header\n",
    "columns_names = [\"day\",\"line_id_distinct\",\"id_avl_distinct\",\"line-id_id-avl_distinct\",\"event_0_count\",\"event_64_count\",\"id_point_distinct\",\"hour_diff_mean\",\"hour_diff_min\",\"hour_diff_max\",\"hour_diff_stddev\",\"hour_diff_quantile_25\",\"hour_diff_quantile_50\",\"hour_diff_quantile_75\"]\n",
    "\n",
    "for i in range(0,len(columns_names)):\n",
    "    new_df = new_df.withColumnRenamed(\"_%s\" % (i + 1), columns_names[i])\n",
    "    \n",
    "# write file in S3\n",
    "new_df.repartition(1).write.option(\"header\",\"true\").csv(\"s3://mobility-traces-sp/statistics/exploring-data/1-general-statistics-excel-version\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#https://forums.wikitechy.com/question/how-to-find-median-and-quantiles-using-spark/\n",
    "# quantile splits data in equal sized groups\n",
    "# percentile = splits the data in 100 equally sized groups\n",
    "# https://towardsdatascience.com/exploratory-data-analysis-eda-with-pyspark-on-databricks-e8d6529626b1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Result\n",
    "# Saved file header/fields\n",
    "# day --> day of the month (october)\n",
    "# line_id_distinct --> number of distinct bus line in a day\n",
    "# id_avl_distinct --> number of distinct bus in a day\n",
    "# line-id_id-avl_distinct --> distinct line_id/bus_id pair\n",
    "# event_0_count --> number of event 0\n",
    "# event_64_count --> number of event 64\n",
    "# hour_diff_mean --> mean of difference between hour_server and hour_avl\n",
    "# hour_diff_min --> min value of difference between hour_server and hour_avl\n",
    "# hour_diff_max --> max value of difference between hour_server and hour_avl\n",
    "# hour_diff_stddev --> std value of difference between hour_server and hour_avl\n",
    "# hour_diff_quantile_25 --> hour_diff < x of 25% of registers in that day\n",
    "# hour_diff_quantile_50 --> hour_diff < x of 50% of registers in that day\n",
    "# hour_diff_quantile_75 --> hour_diff < x of 75% of registers in that day"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}