big_data_file_formats.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ef88ea59-de9b-4bd0-a69a-c6b23638e043",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting s3fs\n",
      "  Downloading s3fs-2023.12.2-py3-none-any.whl.metadata (1.6 kB)\n",
      "Collecting aiobotocore<3.0.0,>=2.5.4 (from s3fs)\n",
      "  Downloading aiobotocore-2.9.0-py3-none-any.whl.metadata (20 kB)\n",
      "Collecting fsspec==2023.12.2 (from s3fs)\n",
      "  Downloading fsspec-2023.12.2-py3-none-any.whl.metadata (6.8 kB)\n",
      "Collecting aiohttp!=4.0.0a0,!=4.0.0a1 (from s3fs)\n",
      "  Downloading aiohttp-3.9.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (7.4 kB)\n",
      "Collecting botocore<1.33.14,>=1.33.2 (from aiobotocore<3.0.0,>=2.5.4->s3fs)\n",
      "  Downloading botocore-1.33.13-py3-none-any.whl.metadata (6.1 kB)\n",
      "Collecting wrapt<2.0.0,>=1.10.10 (from aiobotocore<3.0.0,>=2.5.4->s3fs)\n",
      "  Downloading wrapt-1.16.0-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (6.6 kB)\n",
      "Collecting aioitertools<1.0.0,>=0.5.1 (from aiobotocore<3.0.0,>=2.5.4->s3fs)\n",
      "  Downloading aioitertools-0.11.0-py3-none-any.whl (23 kB)\n",
      "Requirement already satisfied: attrs>=17.3.0 in /opt/conda/lib/python3.11/site-packages (from aiohttp!=4.0.0a0,!=4.0.0a1->s3fs) (23.1.0)\n",
      "Collecting multidict<7.0,>=4.5 (from aiohttp!=4.0.0a0,!=4.0.0a1->s3fs)\n",
      "  Downloading multidict-6.0.4-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (117 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m117.4/117.4 kB\u001b[0m \u001b[31m6.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hCollecting yarl<2.0,>=1.0 (from aiohttp!=4.0.0a0,!=4.0.0a1->s3fs)\n",
      "  Downloading yarl-1.9.4-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (31 kB)\n",
      "Collecting frozenlist>=1.1.1 (from aiohttp!=4.0.0a0,!=4.0.0a1->s3fs)\n",
      "  Downloading frozenlist-1.4.1-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (12 kB)\n",
      "Collecting aiosignal>=1.1.2 (from aiohttp!=4.0.0a0,!=4.0.0a1->s3fs)\n",
      "  Downloading aiosignal-1.3.1-py3-none-any.whl (7.6 kB)\n",
      "Collecting jmespath<2.0.0,>=0.7.1 (from botocore<1.33.14,>=1.33.2->aiobotocore<3.0.0,>=2.5.4->s3fs)\n",
      "  Downloading jmespath-1.0.1-py3-none-any.whl (20 kB)\n",
      "Requirement already satisfied: python-dateutil<3.0.0,>=2.1 in /opt/conda/lib/python3.11/site-packages (from botocore<1.33.14,>=1.33.2->aiobotocore<3.0.0,>=2.5.4->s3fs) (2.8.2)\n",
      "Requirement already satisfied: urllib3<2.1,>=1.25.4 in /opt/conda/lib/python3.11/site-packages (from botocore<1.33.14,>=1.33.2->aiobotocore<3.0.0,>=2.5.4->s3fs) (2.0.7)\n",
      "Requirement already satisfied: idna>=2.0 in /opt/conda/lib/python3.11/site-packages (from yarl<2.0,>=1.0->aiohttp!=4.0.0a0,!=4.0.0a1->s3fs) (3.4)\n",
      "Requirement already satisfied: six>=1.5 in /opt/conda/lib/python3.11/site-packages (from python-dateutil<3.0.0,>=2.1->botocore<1.33.14,>=1.33.2->aiobotocore<3.0.0,>=2.5.4->s3fs) (1.16.0)\n",
      "Downloading s3fs-2023.12.2-py3-none-any.whl (28 kB)\n",
      "Downloading fsspec-2023.12.2-py3-none-any.whl (168 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m169.0/169.0 kB\u001b[0m \u001b[31m10.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading aiobotocore-2.9.0-py3-none-any.whl (75 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m76.0/76.0 kB\u001b[0m \u001b[31m4.7 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading aiohttp-3.9.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (1.3 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.3/1.3 MB\u001b[0m \u001b[31m22.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0ma \u001b[36m0:00:01\u001b[0m\n",
      "\u001b[?25hDownloading botocore-1.33.13-py3-none-any.whl (11.8 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m11.8/11.8 MB\u001b[0m \u001b[31m25.2 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m00:01\u001b[0m\n",
      "\u001b[?25hDownloading frozenlist-1.4.1-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl (272 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m272.3/272.3 kB\u001b[0m \u001b[31m15.7 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading wrapt-1.16.0-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl (80 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m80.7/80.7 kB\u001b[0m \u001b[31m5.8 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading yarl-1.9.4-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (328 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m328.1/328.1 kB\u001b[0m \u001b[31m19.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hInstalling collected packages: wrapt, multidict, jmespath, fsspec, frozenlist, aioitertools, yarl, botocore, aiosignal, aiohttp, aiobotocore, s3fs\n",
      "  Attempting uninstall: fsspec\n",
      "    Found existing installation: fsspec 2023.9.2\n",
      "    Uninstalling fsspec-2023.9.2:\n",
      "      Successfully uninstalled fsspec-2023.9.2\n",
      "Successfully installed aiobotocore-2.9.0 aiohttp-3.9.1 aioitertools-0.11.0 aiosignal-1.3.1 botocore-1.33.13 frozenlist-1.4.1 fsspec-2023.12.2 jmespath-1.0.1 multidict-6.0.4 s3fs-2023.12.2 wrapt-1.16.0 yarl-1.9.4\n"
     ]
    }
   ],
   "source": [
    "!pip install s3fs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "9f1df9bf-f0fc-479d-9850-9e77370be239",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import s3fs\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql import Row\n",
    "import pyspark.sql.types as T\n",
    "import datetime\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "1f88645b-00be-4362-af9f-e89d7895ec4f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define environment variables\n",
    "os.environ[\"MINIO_KEY\"] = \"minio\"\n",
    "os.environ[\"MINIO_SECRET\"] = \"minio123\"\n",
    "os.environ[\"MINIO_ENDPOINT\"] = \"http://minio1:9000\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "2d9485a1-28b7-49b8-8369-18631d17f21e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create Spark session\n",
    "spark = SparkSession.builder \\\n",
    "    .appName(\"big_data_file_formats\") \\\n",
    "    .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.1026,org.apache.spark:spark-avro_2.12:3.5.0,io.delta:delta-spark_2.12:3.0.0\") \\\n",
    "    .config(\"spark.hadoop.fs.s3a.endpoint\", os.environ[\"MINIO_ENDPOINT\"]) \\\n",
    "    .config(\"spark.hadoop.fs.s3a.access.key\", os.environ[\"MINIO_KEY\"]) \\\n",
    "    .config(\"spark.hadoop.fs.s3a.secret.key\", os.environ[\"MINIO_SECRET\"]) \\\n",
    "    .config(\"spark.hadoop.fs.s3a.path.style.access\", \"true\") \\\n",
    "    .config(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\") \\\n",
    "    .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\") \\\n",
    "    .config(\"spark.sql.catalog.spark_catalog\", \"org.apache.spark.sql.delta.catalog.DeltaCatalog\") \\\n",
    "    .enableHiveSupport() \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "b1f2911a-d948-4b41-a137-4d7fca49cb61",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "version 2.12.18\n"
     ]
    }
   ],
   "source": [
    "# Print Scala version\n",
    "scala_version = spark.sparkContext._jvm.scala.util.Properties.versionString()\n",
    "print(scala_version)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "6ba686e9-96fa-4ee0-a6b5-7441dec3f261",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10000000"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Generate sample data\n",
    "num_rows = 10000000\n",
    "df = spark.range(0, num_rows)\n",
    "\n",
    "# Add columns\n",
    "for i in range(1, 10):  # Since we already have one column\n",
    "    if i % 2 == 0:\n",
    "        # Integer column\n",
    "        df = df.withColumn(f\"int_col_{i}\", (F.randn() * 100).cast(T.IntegerType()))\n",
    "    else:\n",
    "        # String column\n",
    "        df = df.withColumn(f\"str_col_{i}\", (F.rand() * num_rows).cast(T.IntegerType()).cast(\"string\"))\n",
    "\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "a3ac25f2-5f1d-4db9-8a7c-8503a4fbba50",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---------+---------+---------+---------+---------+---------+---------+---------+---------+\n",
      "|id |str_col_1|int_col_2|str_col_3|int_col_4|str_col_5|int_col_6|str_col_7|int_col_8|str_col_9|\n",
      "+---+---------+---------+---------+---------+---------+---------+---------+---------+---------+\n",
      "|0  |7764018  |128      |1632029  |-15      |5858297  |114      |1025493  |-88      |7376083  |\n",
      "|1  |2618524  |118      |912383   |235      |6684042  |-115     |9882176  |170      |3220749  |\n",
      "|2  |6351000  |75       |3515510  |26       |2605886  |89       |3217428  |87       |4045983  |\n",
      "|3  |4346827  |-70      |2627979  |-23      |9543505  |69       |2421674  |-141     |7049734  |\n",
      "|4  |9458796  |-106     |6374672  |-142     |5550170  |25       |4842269  |-97      |5265771  |\n",
      "|5  |9203992  |23       |4818602  |42       |530044   |28       |5560538  |-75      |2307858  |\n",
      "|6  |8900698  |-130     |2735238  |-135     |1308929  |22       |3279458  |-22      |3412851  |\n",
      "|7  |6876605  |-35      |6690534  |-41      |273737   |-178     |8789689  |88       |4200849  |\n",
      "|8  |3274838  |-42      |1270841  |-62      |4592242  |133      |4665549  |-125     |3993964  |\n",
      "|9  |4904488  |206      |2176042  |58       |1388630  |-63      |9364695  |78       |2657371  |\n",
      "+---+---------+---------+---------+---------+---------+---------+---------+---------+---------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Show rows from sample data\n",
    "df.show(10,truncate = False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "57369f38-6ad5-4e29-ab08-78c23a7b33ab",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: long (nullable = false)\n",
      " |-- str_col_1: string (nullable = true)\n",
      " |-- int_col_2: integer (nullable = true)\n",
      " |-- str_col_3: string (nullable = true)\n",
      " |-- int_col_4: integer (nullable = true)\n",
      " |-- str_col_5: string (nullable = true)\n",
      " |-- int_col_6: integer (nullable = true)\n",
      " |-- str_col_7: string (nullable = true)\n",
      " |-- int_col_8: integer (nullable = true)\n",
      " |-- str_col_9: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Print schema of sample data\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2cf149ea-7f5f-4818-9e2f-ff8bc0a3d3a7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write 4 CSVs for comparing performance for every file type\n",
    "df.write.csv(\"s3a://mybucket/ten_million_parquet.csv\")\n",
    "df.write.csv(\"s3a://mybucket/ten_million_avro.csv\")\n",
    "df.write.csv(\"s3a://mybucket/ten_million_orc.csv\")\n",
    "df.write.csv(\"s3a://mybucket/ten_million_delta.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "9a31fb32-3af5-4c9f-bb58-4c3455651835",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read all four CSVs to create dataframes\n",
    "schema = T.StructType([\n",
    "    T.StructField(\"id\", T.LongType(), nullable=False),\n",
    "    T.StructField(\"str_col_1\", T.StringType(), nullable=True),\n",
    "    T.StructField(\"int_col_2\", T.IntegerType(), nullable=True),\n",
    "    T.StructField(\"str_col_3\", T.StringType(), nullable=True),\n",
    "    T.StructField(\"int_col_4\", T.IntegerType(), nullable=True),\n",
    "    T.StructField(\"str_col_5\", T.StringType(), nullable=True),\n",
    "    T.StructField(\"int_col_6\", T.IntegerType(), nullable=True),\n",
    "    T.StructField(\"str_col_7\", T.StringType(), nullable=True),\n",
    "    T.StructField(\"int_col_8\", T.IntegerType(), nullable=True),\n",
    "    T.StructField(\"str_col_9\", T.StringType(), nullable=True)\n",
    "])\n",
    "\n",
    "df_csv_parquet = spark.read.format(\"csv\").option(\"header\",True).schema(schema).load(\"s3a://mybucket/ten_million_parquet.csv\")\n",
    "df_csv_avro = spark.read.format(\"csv\").option(\"header\",True).schema(schema).load(\"s3a://mybucket/ten_million_avro.csv\")\n",
    "df_csv_orc = spark.read.format(\"csv\").option(\"header\",True).schema(schema).load(\"s3a://mybucket/ten_million_orc.csv\")\n",
    "df_csv_delta = spark.read.format(\"csv\").option(\"header\",True).schema(schema).load(\"s3a://mybucket/ten_million_delta.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "e69d3fc8-bd8c-4d4c-a2c7-d5bc48aa5e44",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken to write as Parquet: 15.137483358383179 seconds\n"
     ]
    }
   ],
   "source": [
    "# Write data as Parquet\n",
    "start_time = time.time()\n",
    "df_csv_parquet.write.parquet(\"s3a://mybucket/ten_million_parquet2.parquet\")\n",
    "end_time = time.time()\n",
    "print(f\"Time taken to write as Parquet: {end_time - start_time} seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "90e169d2-c51c-40dc-a435-58b381ddd2a3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken to write as Avro: 12.808173656463623 seconds\n"
     ]
    }
   ],
   "source": [
    "# Write data as Avro\n",
    "start_time = time.time()\n",
    "df_csv_avro.write.format(\"avro\").save(\"s3a://mybucket/ten_million_avro2.avro\")\n",
    "end_time = time.time()\n",
    "print(f\"Time taken to write as Avro: {end_time - start_time} seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "be43020c-ea10-43b7-94a6-f838e5fb496f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken to write as ORC: 12.936703443527222 seconds\n"
     ]
    }
   ],
   "source": [
    "# Write data as ORC\n",
    "start_time = time.time()\n",
    "df_csv_orc.write.orc(\"s3a://mybucket/ten_million_orc2.orc\")\n",
    "end_time = time.time()\n",
    "print(f\"Time taken to write as ORC: {end_time - start_time} seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "b4bcc910-f7d2-4d7e-8c8d-a1609b642ba4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken to write as Delta Lake: 17.779020309448242 seconds\n"
     ]
    }
   ],
   "source": [
    "# Write data as Delta\n",
    "start_time = time.time()\n",
    "df_csv_delta.write.format(\"delta\").save(\"s3a://mybucket/ten_million_delta2.delta\")\n",
    "end_time = time.time()\n",
    "print(f\"Time taken to write as Delta Lake: {end_time - start_time} seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "1fb897d0-13cb-4c8b-86dd-d71633fa001d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----+\n",
      "|str_col_5|str_col_7|count|\n",
      "+---------+---------+-----+\n",
      "|1        |6429997  |1    |\n",
      "+---------+---------+-----+\n",
      "\n",
      "Time taken for query: 12.326371192932129 seconds\n"
     ]
    }
   ],
   "source": [
    "# Perfom aggregation query using Parquet data\n",
    "start_time = time.time()\n",
    "df_parquet = spark.read.parquet(\"s3a://mybucket/ten_million_parquet2.parquet\")\n",
    "df_parquet \\\n",
    ".select(\"str_col_5\",\"str_col_7\",\"int_col_2\") \\\n",
    ".groupBy(\"str_col_5\",\"str_col_7\") \\\n",
    ".count() \\\n",
    ".orderBy(\"count\") \\\n",
    ".limit(1) \\\n",
    ".show(truncate = False)\n",
    "end_time = time.time()\n",
    "print(f\"Time taken for query: {end_time - start_time} seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "b9f10339-ed55-4c5c-a5a1-2357319f6683",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----+\n",
      "|str_col_5|str_col_7|count|\n",
      "+---------+---------+-----+\n",
      "|1        |6429997  |1    |\n",
      "+---------+---------+-----+\n",
      "\n",
      "Time taken for query: 15.421005725860596 seconds\n"
     ]
    }
   ],
   "source": [
    "# Perform aggregation using Avro data\n",
    "df_avro = spark.read.format(\"avro\").load(\"s3a://mybucket/ten_million_avro2.avro\")\n",
    "start_time = time.time()\n",
    "df_avro \\\n",
    ".select(\"str_col_5\",\"str_col_7\",\"int_col_2\") \\\n",
    ".groupBy(\"str_col_5\",\"str_col_7\") \\\n",
    ".count() \\\n",
    ".orderBy(\"count\") \\\n",
    ".limit(1) \\\n",
    ".show(truncate = False)\n",
    "end_time = time.time()\n",
    "print(f\"Time taken for query: {end_time - start_time} seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "633d0a07-6cb3-445d-a30f-e20d7168d760",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----+\n",
      "|str_col_5|str_col_7|count|\n",
      "+---------+---------+-----+\n",
      "|1        |2906292  |1    |\n",
      "+---------+---------+-----+\n",
      "\n",
      "Time taken for query: 13.445027828216553 seconds\n"
     ]
    }
   ],
   "source": [
    "# Perform aggregation using ORC data\n",
    "df_orc = spark.read.orc(\"s3a://mybucket/ten_million_orc2.orc\")\n",
    "start_time = time.time()\n",
    "df_orc \\\n",
    ".select(\"str_col_5\",\"str_col_7\",\"int_col_2\") \\\n",
    ".groupBy(\"str_col_5\",\"str_col_7\") \\\n",
    ".count() \\\n",
    ".orderBy(\"count\") \\\n",
    ".limit(1) \\\n",
    ".show(truncate = False)\n",
    "end_time = time.time()\n",
    "print(f\"Time taken for query: {end_time - start_time} seconds\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "2431c25f-9be0-4deb-9432-c4fa296bef28",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----+\n",
      "|str_col_5|str_col_7|count|\n",
      "+---------+---------+-----+\n",
      "|1        |2906292  |1    |\n",
      "+---------+---------+-----+\n",
      "\n",
      "Time taken for query: 15.512012004852295 seconds\n"
     ]
    }
   ],
   "source": [
    "# Perform aggregation using Delta data\n",
    "df_delta = spark.read.format(\"delta\").load(\"s3a://mybucket/ten_million_delta2.delta\")\n",
    "start_time = time.time()\n",
    "df_delta \\\n",
    ".select(\"str_col_5\",\"str_col_7\",\"int_col_2\") \\\n",
    ".groupBy(\"str_col_5\",\"str_col_7\") \\\n",
    ".count() \\\n",
    ".orderBy(\"count\") \\\n",
    ".limit(1) \\\n",
    ".show(truncate = False)\n",
    "end_time = time.time()\n",
    "print(f\"Time taken for query: {end_time - start_time} seconds\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}