{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceType":"datasetVersion","sourceId":2965819,"datasetId":1818367,"databundleVersionId":3013484},{"sourceType":"datasetVersion","sourceId":3316532,"datasetId":10100,"databundleVersionId":3367434},{"sourceType":"kernelVersion","sourceId":279669691}],"dockerImageVersionId":31286,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!pip install pyspark","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:19.956125Z","iopub.execute_input":"2026-03-06T08:47:19.956888Z","iopub.status.idle":"2026-03-06T08:47:24.139966Z","shell.execute_reply.started":"2026-03-06T08:47:19.956853Z","shell.execute_reply":"2026-03-06T08:47:24.139193Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql import SparkSession\n\nspark = SparkSession.builder \\\n    .appName(\"NYC Taxi Analysis\") \\\n    .getOrCreate()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:24.971815Z","iopub.execute_input":"2026-03-06T08:47:24.972691Z","iopub.status.idle":"2026-03-06T08:47:34.036111Z","shell.execute_reply.started":"2026-03-06T08:47:24.972652Z","shell.execute_reply":"2026-03-06T08:47:34.035206Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"spark","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:34.037538Z","iopub.execute_input":"2026-03-06T08:47:34.037903Z","iopub.status.idle":"2026-03-06T08:47:34.499945Z","shell.execute_reply.started":"2026-03-06T08:47:34.037879Z","shell.execute_reply":"2026-03-06T08:47:34.49914Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df = spark.read.csv(\"/kaggle/input/datasets/yasserh/nyc-taxi-trip-duration/NYC.csv\", header=True, inferSchema=True)\ndf.printSchema()\ndf.show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:34.501056Z","iopub.execute_input":"2026-03-06T08:47:34.501296Z","iopub.status.idle":"2026-03-06T08:47:42.631191Z","shell.execute_reply.started":"2026-03-06T08:47:34.501275Z","shell.execute_reply":"2026-03-06T08:47:42.63043Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:42.632887Z","iopub.execute_input":"2026-03-06T08:47:42.63313Z","iopub.status.idle":"2026-03-06T08:47:42.707135Z","shell.execute_reply.started":"2026-03-06T08:47:42.63311Z","shell.execute_reply":"2026-03-06T08:47:42.706431Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"descriptive ","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.functions import avg\n\ndf.select(avg(\"trip_duration\")).show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:46.374683Z","iopub.execute_input":"2026-03-06T08:47:46.375561Z","iopub.status.idle":"2026-03-06T08:47:48.3846Z","shell.execute_reply.started":"2026-03-06T08:47:46.375527Z","shell.execute_reply":"2026-03-06T08:47:48.383842Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df.groupBy(\"passenger_count\").count().show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:48.38604Z","iopub.execute_input":"2026-03-06T08:47:48.386441Z","iopub.status.idle":"2026-03-06T08:47:50.196843Z","shell.execute_reply.started":"2026-03-06T08:47:48.386407Z","shell.execute_reply":"2026-03-06T08:47:50.196102Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"predictive","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.functions import col, sum\n\ndf.select(\n    sum(col(\"passenger_count\").isNull().cast(\"int\")).alias(\"null_count\")\n).show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:54.984979Z","iopub.execute_input":"2026-03-06T08:47:54.985254Z","iopub.status.idle":"2026-03-06T08:47:56.321357Z","shell.execute_reply.started":"2026-03-06T08:47:54.985234Z","shell.execute_reply":"2026-03-06T08:47:56.319928Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df_ml = df.select(\"passenger_count\", \"trip_duration\")\nfrom pyspark.ml.feature import VectorAssembler\n\nassembler = VectorAssembler(\n    inputCols=[\"passenger_count\"],\n    outputCol=\"features\"\n)\n\ndata = assembler.transform(df_ml)\n\ndata.select(\"passenger_count\", \"features\").show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:56.324075Z","iopub.execute_input":"2026-03-06T08:47:56.324273Z","iopub.status.idle":"2026-03-06T08:47:57.031959Z","shell.execute_reply.started":"2026-03-06T08:47:56.324255Z","shell.execute_reply":"2026-03-06T08:47:57.031207Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"data.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:47:57.033016Z","iopub.execute_input":"2026-03-06T08:47:57.033392Z","iopub.status.idle":"2026-03-06T08:47:57.1974Z","shell.execute_reply.started":"2026-03-06T08:47:57.033353Z","shell.execute_reply":"2026-03-06T08:47:57.196401Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Train Prediction Model","metadata":{}},{"cell_type":"code","source":"from pyspark.ml.regression import LinearRegression\n\nlr = LinearRegression(featuresCol=\"features\", labelCol=\"trip_duration\")\n\nmodel = lr.fit(data)\n\npredictions = model.transform(data)\n\npredictions.select(\"trip_duration\", \"prediction\").show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:48:01.220555Z","iopub.execute_input":"2026-03-06T08:48:01.221087Z","iopub.status.idle":"2026-03-06T08:48:07.305986Z","shell.execute_reply.started":"2026-03-06T08:48:01.22106Z","shell.execute_reply":"2026-03-06T08:48:07.303779Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Evaluate Model","metadata":{}},{"cell_type":"code","source":"from pyspark.ml.evaluation import RegressionEvaluator\n\n# RMSE\nrmse_evaluator = RegressionEvaluator(\n    labelCol=\"trip_duration\",\n    predictionCol=\"prediction\",\n    metricName=\"rmse\"\n)\n\nrmse = rmse_evaluator.evaluate(predictions)\nprint(\"RMSE:\", rmse)\n\n\n# MAE\nmae_evaluator = RegressionEvaluator(\n    labelCol=\"trip_duration\",\n    predictionCol=\"prediction\",\n    metricName=\"mae\"\n)\n\nmae = mae_evaluator.evaluate(predictions)\nprint(\"MAE:\", mae)\n\n\n# R2 Score\nr2_evaluator = RegressionEvaluator(\n    labelCol=\"trip_duration\",\n    predictionCol=\"prediction\",\n    metricName=\"r2\"\n)\n\nr2 = r2_evaluator.evaluate(predictions)\nprint(\"R2 Score:\", r2)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:48:11.873906Z","iopub.execute_input":"2026-03-06T08:48:11.874181Z","iopub.status.idle":"2026-03-06T08:48:17.875188Z","shell.execute_reply.started":"2026-03-06T08:48:11.874159Z","shell.execute_reply":"2026-03-06T08:48:17.874222Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Streaming Analytics for Structured Data","metadata":{}},{"cell_type":"code","source":"stream_df = spark.readStream \\\n    .schema(df.schema) \\\n    .csv(\"/kaggle/input/datasets/yasserh/nyc-taxi-trip-duration/\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:48:18.023813Z","iopub.execute_input":"2026-03-06T08:48:18.024298Z","iopub.status.idle":"2026-03-06T08:48:18.064266Z","shell.execute_reply.started":"2026-03-06T08:48:18.024273Z","shell.execute_reply":"2026-03-06T08:48:18.063433Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import count\n\ntrip_counts = stream_df.groupBy(\"passenger_count\") \\\n                       .agg(count(\"*\").alias(\"total_trips\"))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:48:19.041458Z","iopub.execute_input":"2026-03-06T08:48:19.042252Z","iopub.status.idle":"2026-03-06T08:48:19.063785Z","shell.execute_reply.started":"2026-03-06T08:48:19.042221Z","shell.execute_reply":"2026-03-06T08:48:19.063017Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"query = trip_counts.writeStream \\\n    .outputMode(\"complete\") \\\n    .format(\"console\") \\\n    .start()\n\nquery.awaitTermination(20)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:48:20.984989Z","iopub.execute_input":"2026-03-06T08:48:20.985665Z","iopub.status.idle":"2026-03-06T08:48:41.280816Z","shell.execute_reply.started":"2026-03-06T08:48:20.985635Z","shell.execute_reply":"2026-03-06T08:48:41.280163Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"**SEMI STRUCTURED DATA**","metadata":{}},{"cell_type":"code","source":"import os\n\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        if \"yelp\" in dirname.lower() or \"yelp\" in filename.lower():\n            print(os.path.join(dirname, filename))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:52:30.766711Z","iopub.execute_input":"2026-03-06T08:52:30.767516Z","iopub.status.idle":"2026-03-06T08:52:30.786133Z","shell.execute_reply.started":"2026-03-06T08:52:30.767485Z","shell.execute_reply":"2026-03-06T08:52:30.78522Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"reviews_df = spark.read.json(\"/kaggle/input/datasets/organizations/yelp-dataset/yelp-dataset\")\n\nreviews_df.show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:55:03.146517Z","iopub.execute_input":"2026-03-06T08:55:03.147581Z","iopub.status.idle":"2026-03-06T08:55:51.075069Z","shell.execute_reply.started":"2026-03-06T08:55:03.147545Z","shell.execute_reply":"2026-03-06T08:55:51.074181Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Descriptive Analytics","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.functions import avg\n\nreviews_df.select(avg(\"stars\")).show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:56:47.602915Z","iopub.execute_input":"2026-03-06T08:56:47.603825Z","iopub.status.idle":"2026-03-06T08:57:07.525697Z","shell.execute_reply.started":"2026-03-06T08:56:47.603796Z","shell.execute_reply":"2026-03-06T08:57:07.522899Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"reviews_df.groupBy(\"stars\").count().show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:58:17.264936Z","iopub.execute_input":"2026-03-06T08:58:17.265399Z","iopub.status.idle":"2026-03-06T08:58:37.639195Z","shell.execute_reply.started":"2026-03-06T08:58:17.265345Z","shell.execute_reply":"2026-03-06T08:58:37.638054Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"reviews_df.groupBy(\"user_id\") \\\n.count() \\\n.orderBy(\"count\", ascending=False) \\\n.show(10)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T08:59:07.64794Z","iopub.execute_input":"2026-03-06T08:59:07.648474Z","iopub.status.idle":"2026-03-06T08:59:35.099065Z","shell.execute_reply.started":"2026-03-06T08:59:07.648444Z","shell.execute_reply":"2026-03-06T08:59:35.098071Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"MapReduce","metadata":{}},{"cell_type":"code","source":"reviews_rdd = reviews_df.select(\"stars\").rdd\n\nmapped = reviews_rdd.map(lambda x: (x.stars, 1))\n\nreduced = mapped.reduceByKey(lambda a, b: a + b)\n\nprint(reduced.collect())","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:00:16.663786Z","iopub.execute_input":"2026-03-06T09:00:16.664111Z","iopub.status.idle":"2026-03-06T09:01:08.130094Z","shell.execute_reply.started":"2026-03-06T09:00:16.664086Z","shell.execute_reply":"2026-03-06T09:01:08.129404Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Predictive Analytics","metadata":{}},{"cell_type":"markdown","source":"Create feature:","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.functions import length\n\nreviews_ml = reviews_df.select(\"text\",\"stars\")\n\nreviews_ml = reviews_ml.withColumn(\"review_length\", length(\"text\"))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:02:19.445299Z","iopub.execute_input":"2026-03-06T09:02:19.445874Z","iopub.status.idle":"2026-03-06T09:02:19.460571Z","shell.execute_reply.started":"2026-03-06T09:02:19.445845Z","shell.execute_reply":"2026-03-06T09:02:19.459887Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Create feature vector:","metadata":{}},{"cell_type":"code","source":"from pyspark.ml.feature import VectorAssembler\n\nassembler = VectorAssembler(\ninputCols=[\"review_length\"],\noutputCol=\"features\"\n)\n\ndata = assembler.transform(reviews_ml)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:02:28.861626Z","iopub.execute_input":"2026-03-06T09:02:28.862191Z","iopub.status.idle":"2026-03-06T09:02:28.907317Z","shell.execute_reply.started":"2026-03-06T09:02:28.86216Z","shell.execute_reply":"2026-03-06T09:02:28.906399Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Train regression model:","metadata":{}},{"cell_type":"code","source":"reviews_ml = reviews_ml.dropna()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:04:41.069133Z","iopub.execute_input":"2026-03-06T09:04:41.06983Z","iopub.status.idle":"2026-03-06T09:04:41.083415Z","shell.execute_reply.started":"2026-03-06T09:04:41.069801Z","shell.execute_reply":"2026-03-06T09:04:41.082371Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import length\n\n# select needed columns\nreviews_ml = reviews_df.select(\"text\", \"stars\")\n\n# create review length feature\nreviews_ml = reviews_ml.withColumn(\"review_length\", length(\"text\"))\n\n# remove null rows\nreviews_ml = reviews_ml.dropna()\n\nfrom pyspark.ml.feature import VectorAssembler\n\nassembler = VectorAssembler(\n    inputCols=[\"review_length\"],\n    outputCol=\"features\"\n)\n\ndata = assembler.transform(reviews_ml)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:05:53.151876Z","iopub.execute_input":"2026-03-06T09:05:53.152171Z","iopub.status.idle":"2026-03-06T09:05:53.207669Z","shell.execute_reply.started":"2026-03-06T09:05:53.15215Z","shell.execute_reply":"2026-03-06T09:05:53.206941Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.regression import LinearRegression\n\nlr = LinearRegression(featuresCol=\"features\", labelCol=\"stars\")\n\nmodel = lr.fit(data)\n\npredictions = model.transform(data)\n\npredictions.select(\"stars\",\"prediction\").show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:06:06.051881Z","iopub.execute_input":"2026-03-06T09:06:06.052559Z","iopub.status.idle":"2026-03-06T09:07:12.582752Z","shell.execute_reply.started":"2026-03-06T09:06:06.052532Z","shell.execute_reply":"2026-03-06T09:07:12.5815Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"assembler = VectorAssembler(\n    inputCols=[\"review_length\"],\n    outputCol=\"features\",\n    handleInvalid=\"skip\"\n)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:10:05.473323Z","iopub.execute_input":"2026-03-06T09:10:05.473767Z","iopub.status.idle":"2026-03-06T09:10:05.481267Z","shell.execute_reply.started":"2026-03-06T09:10:05.473735Z","shell.execute_reply":"2026-03-06T09:10:05.480586Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"MapReduce Analytics","metadata":{}},{"cell_type":"code","source":"reviews_rdd = reviews_df.select(\"stars\").rdd","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:12:25.045489Z","iopub.execute_input":"2026-03-06T09:12:25.046333Z","iopub.status.idle":"2026-03-06T09:12:25.086001Z","shell.execute_reply.started":"2026-03-06T09:12:25.046292Z","shell.execute_reply":"2026-03-06T09:12:25.08512Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"mapped = reviews_rdd.map(lambda x: (x.stars, 1))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:12:39.196786Z","iopub.execute_input":"2026-03-06T09:12:39.197103Z","iopub.status.idle":"2026-03-06T09:12:39.201486Z","shell.execute_reply.started":"2026-03-06T09:12:39.197078Z","shell.execute_reply":"2026-03-06T09:12:39.200773Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"reduced = mapped.reduceByKey(lambda a, b: a + b)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:12:49.308994Z","iopub.execute_input":"2026-03-06T09:12:49.309747Z","iopub.status.idle":"2026-03-06T09:12:49.32549Z","shell.execute_reply.started":"2026-03-06T09:12:49.309717Z","shell.execute_reply":"2026-03-06T09:12:49.324863Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print(reduced.collect())","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:12:59.14188Z","iopub.execute_input":"2026-03-06T09:12:59.142182Z","iopub.status.idle":"2026-03-06T09:13:47.630079Z","shell.execute_reply.started":"2026-03-06T09:12:59.142158Z","shell.execute_reply":"2026-03-06T09:13:47.628867Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Streaming Analytics","metadata":{}},{"cell_type":"code","source":"stream_reviews = spark.readStream \\\n    .schema(reviews_df.schema) \\\n    .json(\"/kaggle/input\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:14:52.190635Z","iopub.execute_input":"2026-03-06T09:14:52.191304Z","iopub.status.idle":"2026-03-06T09:14:52.226374Z","shell.execute_reply.started":"2026-03-06T09:14:52.191276Z","shell.execute_reply":"2026-03-06T09:14:52.225589Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import count\n\nrating_stream = stream_reviews.groupBy(\"stars\") \\\n                              .agg(count(\"*\").alias(\"review_count\"))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:15:14.028299Z","iopub.execute_input":"2026-03-06T09:15:14.02897Z","iopub.status.idle":"2026-03-06T09:15:14.050353Z","shell.execute_reply.started":"2026-03-06T09:15:14.028942Z","shell.execute_reply":"2026-03-06T09:15:14.049536Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"query = rating_stream.writeStream \\\n    .outputMode(\"complete\") \\\n    .format(\"console\") \\\n    .start()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:15:23.526207Z","iopub.execute_input":"2026-03-06T09:15:23.527007Z","iopub.status.idle":"2026-03-06T09:15:23.589867Z","shell.execute_reply.started":"2026-03-06T09:15:23.526973Z","shell.execute_reply":"2026-03-06T09:15:23.588186Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"query.awaitTermination(10)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:15:38.884757Z","iopub.execute_input":"2026-03-06T09:15:38.885298Z","iopub.status.idle":"2026-03-06T09:15:48.890226Z","shell.execute_reply.started":"2026-03-06T09:15:38.885272Z","shell.execute_reply":"2026-03-06T09:15:48.889518Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"**unstrctured data**","metadata":{}},{"cell_type":"code","source":"# Extract images\n!unzip -q /kaggle/input/competitions/dogs-vs-cats/train.zip -d /kaggle/working/","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:52:33.611155Z","iopub.execute_input":"2026-03-06T09:52:33.611594Z","iopub.status.idle":"2026-03-06T09:52:33.752991Z","shell.execute_reply.started":"2026-03-06T09:52:33.611566Z","shell.execute_reply":"2026-03-06T09:52:33.752028Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\n\nfiles = os.listdir(\"/kaggle/working/train\")\n\ndata = []\n\nfor file in files:\n    \n    if \"cat\" in file:\n        label = \"cat\"\n    else:\n        label = \"dog\"\n        \n    data.append((file,label))\n\ndf_images = spark.createDataFrame(data, [\"filename\",\"label\"])\n\ndf_images.show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:52:42.930345Z","iopub.execute_input":"2026-03-06T09:52:42.93135Z","iopub.status.idle":"2026-03-06T09:52:43.448624Z","shell.execute_reply.started":"2026-03-06T09:52:42.931305Z","shell.execute_reply":"2026-03-06T09:52:43.447646Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"Descriptive Analytics","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df_images.groupBy(\"label\").count().show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:53:06.492883Z","iopub.execute_input":"2026-03-06T09:53:06.493194Z","iopub.status.idle":"2026-03-06T09:53:07.020239Z","shell.execute_reply.started":"2026-03-06T09:53:06.493167Z","shell.execute_reply":"2026-03-06T09:53:07.019436Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print(\"Total Images:\", df_images.count())","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:53:17.634958Z","iopub.execute_input":"2026-03-06T09:53:17.635243Z","iopub.status.idle":"2026-03-06T09:53:18.068653Z","shell.execute_reply.started":"2026-03-06T09:53:17.635222Z","shell.execute_reply":"2026-03-06T09:53:18.067658Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Predictive Analytics","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.functions import length\n\ndf_pred = df_images.withColumn(\"name_length\", length(\"filename\"))\n\ndf_pred.show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:53:36.783573Z","iopub.execute_input":"2026-03-06T09:53:36.784234Z","iopub.status.idle":"2026-03-06T09:53:36.983232Z","shell.execute_reply.started":"2026-03-06T09:53:36.784195Z","shell.execute_reply":"2026-03-06T09:53:36.981866Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.feature import VectorAssembler\n\nassembler = VectorAssembler(\n    inputCols=[\"name_length\"],\n    outputCol=\"features\"\n)\n\ndata = assembler.transform(df_pred)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:53:46.534762Z","iopub.execute_input":"2026-03-06T09:53:46.535463Z","iopub.status.idle":"2026-03-06T09:53:46.576125Z","shell.execute_reply.started":"2026-03-06T09:53:46.535436Z","shell.execute_reply":"2026-03-06T09:53:46.575458Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.feature import StringIndexer\n\nindexer = StringIndexer(inputCol=\"label\", outputCol=\"label_index\")\n\ndata = indexer.fit(data).transform(data)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:53:55.542674Z","iopub.execute_input":"2026-03-06T09:53:55.543432Z","iopub.status.idle":"2026-03-06T09:53:56.675478Z","shell.execute_reply.started":"2026-03-06T09:53:55.543404Z","shell.execute_reply":"2026-03-06T09:53:56.673975Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.classification import LogisticRegression\n\nlr = LogisticRegression(\n    featuresCol=\"features\",\n    labelCol=\"label_index\"\n)\n\nmodel = lr.fit(data)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:54:19.24242Z","iopub.execute_input":"2026-03-06T09:54:19.243012Z","iopub.status.idle":"2026-03-06T09:54:22.493576Z","shell.execute_reply.started":"2026-03-06T09:54:19.242986Z","shell.execute_reply":"2026-03-06T09:54:22.492805Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"predictions = model.transform(data)\n\npredictions.select(\"filename\",\"label\",\"prediction\").show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:54:35.45833Z","iopub.execute_input":"2026-03-06T09:54:35.458973Z","iopub.status.idle":"2026-03-06T09:54:35.754088Z","shell.execute_reply.started":"2026-03-06T09:54:35.458943Z","shell.execute_reply":"2026-03-06T09:54:35.753332Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n\nevaluator = MulticlassClassificationEvaluator(\n    labelCol=\"label_index\",\n    predictionCol=\"prediction\",\n    metricName=\"accuracy\"\n)\n\naccuracy = evaluator.evaluate(predictions)\n\nprint(\"Model Accuracy:\", accuracy)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:54:44.494824Z","iopub.execute_input":"2026-03-06T09:54:44.495116Z","iopub.status.idle":"2026-03-06T09:54:45.08438Z","shell.execute_reply.started":"2026-03-06T09:54:44.495091Z","shell.execute_reply":"2026-03-06T09:54:45.083174Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"MapReduce Analytics","metadata":{}},{"cell_type":"code","source":"rdd = df_images.rdd","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:55:04.028073Z","iopub.execute_input":"2026-03-06T09:55:04.028641Z","iopub.status.idle":"2026-03-06T09:55:04.046831Z","shell.execute_reply.started":"2026-03-06T09:55:04.028611Z","shell.execute_reply":"2026-03-06T09:55:04.046024Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"mapped = rdd.map(lambda x: (x.label,1))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:55:12.221035Z","iopub.execute_input":"2026-03-06T09:55:12.22172Z","iopub.status.idle":"2026-03-06T09:55:12.226231Z","shell.execute_reply.started":"2026-03-06T09:55:12.221689Z","shell.execute_reply":"2026-03-06T09:55:12.225468Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"reduced = mapped.reduceByKey(lambda a,b: a+b)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:55:20.627809Z","iopub.execute_input":"2026-03-06T09:55:20.62844Z","iopub.status.idle":"2026-03-06T09:55:20.648496Z","shell.execute_reply.started":"2026-03-06T09:55:20.628404Z","shell.execute_reply":"2026-03-06T09:55:20.647663Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print(reduced.collect())","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:55:31.869336Z","iopub.execute_input":"2026-03-06T09:55:31.86994Z","iopub.status.idle":"2026-03-06T09:55:32.137968Z","shell.execute_reply.started":"2026-03-06T09:55:31.869913Z","shell.execute_reply":"2026-03-06T09:55:32.137352Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Streaming Analytics","metadata":{}},{"cell_type":"code","source":"schema = reviews_df.schema","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T09:59:54.846268Z","iopub.execute_input":"2026-03-06T09:59:54.846873Z","iopub.status.idle":"2026-03-06T09:59:54.850354Z","shell.execute_reply.started":"2026-03-06T09:59:54.846845Z","shell.execute_reply":"2026-03-06T09:59:54.849699Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"stream_df = spark.readStream \\\n    .schema(schema) \\\n    .json(\"/kaggle/input/datasets/organizations/yelp-dataset/yelp-dataset\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T10:00:11.85516Z","iopub.execute_input":"2026-03-06T10:00:11.855495Z","iopub.status.idle":"2026-03-06T10:00:11.872186Z","shell.execute_reply.started":"2026-03-06T10:00:11.855452Z","shell.execute_reply":"2026-03-06T10:00:11.871382Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import count\n\nrating_stream = stream_df.groupBy(\"stars\") \\\n                         .agg(count(\"*\").alias(\"review_count\"))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T10:00:21.587939Z","iopub.execute_input":"2026-03-06T10:00:21.588466Z","iopub.status.idle":"2026-03-06T10:00:21.615257Z","shell.execute_reply.started":"2026-03-06T10:00:21.588436Z","shell.execute_reply":"2026-03-06T10:00:21.61433Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"query = rating_stream.writeStream \\\n    .outputMode(\"complete\") \\\n    .format(\"console\") \\\n    .start()\n\nquery.awaitTermination(10)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-06T10:00:31.709803Z","iopub.execute_input":"2026-03-06T10:00:31.710436Z","iopub.status.idle":"2026-03-06T10:00:41.786941Z","shell.execute_reply.started":"2026-03-06T10:00:31.710398Z","shell.execute_reply":"2026-03-06T10:00:41.784967Z"}},"outputs":[],"execution_count":null}]}