StreamingNotebook/MLnotebook.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[<pyspark.sql.streaming.StreamingQuery at 0x7f6d9bbea6a0>]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.streams.active"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Using TensorFlow backend.\n"
     ]
    }
   ],
   "source": [
    "from keras import optimizers\n",
    "from keras.models import Sequential\n",
    "from keras.layers import Dense, LSTM, Dropout, GRU\n",
    "from pyspark.ml.evaluation import RegressionEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "userSchema = StructType() \\\n",
    "        .add(\"window\",StructType()\\\n",
    "             .add(\"start\",TimestampType())\\\n",
    "             .add(\"end\",TimestampType()))\\\n",
    "        .add(\"system\", StringType())\\\n",
    "        .add(\"count\", LongType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- window: struct (nullable = true)\n",
      " |    |-- start: timestamp (nullable = true)\n",
      " |    |-- end: timestamp (nullable = true)\n",
      " |-- system: string (nullable = true)\n",
      " |-- count: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "raw_data = spark\\\n",
    ".readStream.format(\"parquet\")\\\n",
    ".schema(userSchema)\\\n",
    ".load(\"/cms/users/carizapo/ming/groupdata_cmsweb_logs\");\n",
    "raw_data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data_match = raw_data.withColumn('date',col(\"window.start\")).drop(col(\"window\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data_flow = raw_data.writeStream.queryName(\"hdfs\").outputMode(\"Append\").format(\"memory\").start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'id': '62892391-6575-4b53-8cec-b83aa6b414b6',\n",
       " 'runId': '786e9e90-b5d6-4af9-81bc-58aad3a22370',\n",
       " 'name': 'hdfs',\n",
       " 'timestamp': '2019-07-11T13:44:36.798Z',\n",
       " 'batchId': 1,\n",
       " 'numInputRows': 0,\n",
       " 'inputRowsPerSecond': 0.0,\n",
       " 'processedRowsPerSecond': 0.0,\n",
       " 'durationMs': {'getOffset': 788, 'triggerExecution': 788},\n",
       " 'stateOperators': [],\n",
       " 'sources': [{'description': 'FileStreamSource[hdfs://analytix/cms/users/carizapo/ming/groupdata_cmsweb_logs]',\n",
       "   'startOffset': {'logOffset': 0},\n",
       "   'endOffset': {'logOffset': 0},\n",
       "   'numInputRows': 0,\n",
       "   'inputRowsPerSecond': 0.0,\n",
       "   'processedRowsPerSecond': 0.0}],\n",
       " 'sink': {'description': 'MemorySink'}}"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "raw_data_flow.lastProgress"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data_flow.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "alerts = spark.sql(\"select * from hdfs\")\n",
    "# alerts.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<li>Must renamed column in joining dataframe.</li>\n",
    "<li>Generate unique id to every system. In order to do that we must apply window spec(work for static dataframe only) to the dataframe</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.window import Window as W\n",
    "raw_data_id = alerts.select(\"system\").distinct()\n",
    "windowSpec = W.orderBy(\"system\")\n",
    "raw_data_id=raw_data_id.withColumn(\"systemID\", row_number().over(windowSpec)).withColumnRenamed(\"system\",\"tempSystem\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data_id.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "alerts_df=alerts\\\n",
    ".join(raw_data_id, raw_data_id.tempSystem == alerts.system)\\\n",
    ".withColumn('date',col(\"window.start\")).drop(col(\"window\"))\\\n",
    ".drop(raw_data_id.tempSystem)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "alerts_df.show()\n",
    "# raw_data_id.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install --user --upgrade dist-keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.feature import MinMaxScaler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "%run StoreItemDemand/custom_transformers.ipynb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_data, test_data = alerts_df.randomSplit([0.8,0.2], seed=1234)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_train = train_data.withColumn('set', lit(0))\n",
    "df_train = df_train.withColumn('id', lit(-1))\n",
    "df_test = test_data.withColumn('set', lit(1))\n",
    "\n",
    "df_test = df_test.withColumn('count', lit(-1))\n",
    "joined = df_test.union(df_train.select(*df_test.columns))\n",
    "\n",
    "train_data = joined.filter('set == 0')\n",
    "test_data = joined.filter('set == 1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, validation = train_data.randomSplit([0.8,0.2], seed=1234)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "SerieMaker?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Feature extraction\n",
    "dc = DateConverter(inputCol='date', outputCol='dateFormated')\n",
    "dex = DayExtractor(inputCol='dateFormated')\n",
    "mex = MonthExtractor(inputCol='dateFormated')\n",
    "yex = YearExtractor(inputCol='dateFormated')\n",
    "# Data process\n",
    "#tentar fazer 'day', 'month', 'year', 'weekday', 'weekend' (as colunas derivadas) ficarem de forma dinâmica, no lugar delas ficar a saída de seu respectivo transformer\n",
    "va = VectorAssembler(inputCols=['systemID','count', 'day', 'month', 'year'], outputCol=\"features\")\n",
    "# scaler = StandardScaler(inputCol=\"features\", outputCol=\"scaledFeatures\", withStd=True, withMean=True)\n",
    "scaler = MinMaxScaler(inputCol=\"features\", outputCol=\"scaledFeatures\")\n",
    "\n",
    "# Serialize data\n",
    "sm = SerieMaker(inputCol='scaledFeatures', dateCol='date', idCol=['systemID'], serieSize=30)\n",
    "\n",
    "pipeline = Pipeline(stages=[dc, dex, mex, yex, va, scaler, sm])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipiline_model = pipeline.fit(train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_transformed = pipiline_model.transform(train)\n",
    "validation_transformed = pipiline_model.transform(validation)\n",
    "test_transformed = pipiline_model.transform(test_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# validation_stream=raw_data_match.join(validation_transformed, [\"system\",\"date\",\"count\"], \"inner\")\n",
    "# test_stream=raw_data_match.join(test_transformed, [\"system\",\"date\",\"count\"], \"inner\")\n",
    "# train_stream=raw_data_match.join(train_transformed, [\"system\",\"date\",\"count\"], \"inner\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# validation_stream_flow.stop()\n",
    "# train_stream_flow.stop()\n",
    "# test_stream_flow.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Train raw: %s' % train.count())\n",
    "print('Validation raw: %s' % validation.count())\n",
    "print('Test raw: %s' % test_data.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Train transformed: %s' % train_transformed.count())\n",
    "print('Validation transformed: %s' % validation_transformed.count())\n",
    "print('Test transformed: %s' % test_transformed.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "trainPd=train_transformed.select('serie', 'count').dropna().toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'trainPd' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-1-0032f75901d0>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mnumpy\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0mnp\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      2\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mml\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlinalg\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mDenseVector\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0mVectors\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 3\u001b[0;31m \u001b[0mtrain_try_1\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mtrainPd\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      4\u001b[0m \u001b[0mtrain_try_2\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mtrainPd\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      5\u001b[0m \u001b[0ma\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mVectors\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdense\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mnp\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0marray\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mNameError\u001b[0m: name 'trainPd' is not defined"
     ]
    }
   ],
   "source": [
    "import numpy as np\n",
    "from pyspark.ml.linalg import DenseVector,Vectors\n",
    "train_try_1=trainPd\n",
    "train_try_2=trainPd\n",
    "a=Vectors.dense(np.array([0,0,0,0,0]))\n",
    "# print(train_try_1['serie'].map(lambda item: item.map(lambda val: a if val is None else val)).head(20))\n",
    "type(train_try_1['series'])\n",
    "train_try_1['series'].head(20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "validationNp=validation_transformed.select('serie','count').dropna().toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_x = trainPd['serie'].values.tolist()\n",
    "train_y = trainPd['count'].values.tolist()\n",
    "validation_x = validationPd['serie'].values.tolist()\n",
    "validation_y = validationPd['count'].values.tolist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_x_np = np.array(train_x)\n",
    "train_y_np = np.array(train_y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_y.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_x.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_label = 1\n",
    "serie_size = len(train_x_np[0])\n",
    "n_features = len(train_x_np[0][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "[ if v is None else v for v in train_x_np]\n",
    "train_x[10]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_y= [i if i[0] is not None else (0, i[1]) for i in train_y_np]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from keras.layers import Input, Flatten, Reshape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# hyperparameters\n",
    "epochs = 80\n",
    "batch = 512\n",
    "lr = 0.001\n",
    "\n",
    "# design network\n",
    "model = Sequential()\n",
    "model.add(GRU(40, input_shape=(serie_size, n_features)))\n",
    "\n",
    "model.add(Flatten())\n",
    "model.add(Dense(10, kernel_initializer='glorot_normal', activation='relu'))\n",
    "model.add(Dense(n_label))\n",
    "\n",
    "model.summary()\n",
    "\n",
    "adam = optimizers.Adam(lr)\n",
    "model.compile(loss='mae', optimizer=adam, metrics=['mse', 'msle'])\n",
    "\n",
    "history = model.fit(train_x, train_y, epochs=epochs, batch_size=batch, validation_data=(validation_x, validation_y), verbose=2, shuffle=False)"
   ]
  },
  {
   "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.6.5"
  },
  "sparkconnect": {
   "bundled_options": [],
   "list_of_options": [
    {
     "name": "spark.jars",
     "value": "/eos/home-y/ysunthor/SWAN_projects/StreamingNotebook"
    },
    {
     "name": "spark.jars.packages",
     "value": "org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.1"
    },
    {
     "name": "spark.executor.instances",
     "value": "10"
    },
    {
     "name": "spark.driver.memory",
     "value": "10g"
    },
    {
     "name": "spark.executor.memory",
     "value": "10g"
    },
    {
     "name": "spark.rpc.message.maxSize",
     "value": "1000"
    },
    {
     "name": "spark.executor.memoryOverhead",
     "value": "5g"
    }
   ]
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}