{ "cells": [ { "cell_type": "markdown", "id": "69c5b011-a418-4e74-8246-9272115873e3", "metadata": {}, "source": [ "## Content\n", "- Creating a SparkSession\n", "- Working with RDDs: reading files, repartitioning, and lazy evaluation\n", "- RDD transformations and their combinations: `flatMap`, `map`, `filter`, `reduceByKey`, and `sortBy`\n", "- Inspecting RDD lineage\n", "- RDD caching and persistence\n", "- Broadcasting reference data" ] }, { "cell_type": "code", "execution_count": null, "id": "b09045a2-5312-4c19-9823-384ca8a35181", "metadata": {}, "outputs": [], "source": [ "from pyspark.sql import SparkSession\n", "import os\n", "\n", "username = os.environ.get(\"JUPYTERHUB_USER\", \"unknown\")\n", "\n", "spark = (\n", " SparkSession.builder\n", " .master(\"spark://192.168.70.21:7077\")\n", " .appName(f\"Spark_Lab1_{username}\")\n", " .config(\"spark.driver.host\", \"192.168.70.21\")\n", " .config(\"spark.driver.bindAddress\", \"192.168.70.21\") \n", " .config(\"spark.cores.max\", \"2\")\n", " .config(\"spark.pyspark.python\", \"/opt/jupyterhub/venv/bin/python\") \n", " .config(\n", " \"spark.executorEnv.PYSPARK_PYTHON\",\n", " \"/opt/jupyterhub/venv/bin/python\"\n", " )\n", " .getOrCreate()\n", ")\n", "\n", "spark.sparkContext.setLogLevel(\"ERROR\")" ] }, { "cell_type": "code", "execution_count": null, "id": "720313ca-15ed-4fed-94b2-5bde63a2313c", "metadata": {}, "outputs": [], "source": [ "!ls -al /datasets" ] }, { "cell_type": "code", "execution_count": null, "id": "d87be211-a6cf-499d-84ba-5a068a5b74a8", "metadata": {}, "outputs": [], "source": [ "sc = spark.sparkContext\n", "\n", "# main operation types\n", "# TRANSFORMATION - creates a new RDD, does not trigger execution (lazy evaluation)\n", "# ACTION - triggers execution of the computation needed to produce the result\n", "\n", "# read the text file into an RDD\n", "latinlinesRDD = sc.textFile(\"/datasets/latincorpus.txt\")\n", "# print the initial number of partitions\n", "print(latinlinesRDD.getNumPartitions())\n", "# ACTION: retrieve and print the first three lines (in original file order)\n", "print(\"Original:\", latinlinesRDD.take(2))\n", "\n", "# TRANSFORMATION: repartition the RDD into 4 partitions\n", "rplatinlinesRDD = latinlinesRDD.repartition(4)\n", "# print the new number of partitions\n", "print(rplatinlinesRDD.getNumPartitions())\n", "\n", "# ACTION: retrieve and print the first three lines (in partition-index order)\n", "print(\"Repartitioned:\", rplatinlinesRDD.take(2))" ] }, { "cell_type": "code", "execution_count": null, "id": "ed30a5c8-904c-498b-b2a2-274c675ad4a8", "metadata": {}, "outputs": [], "source": [ "# Transformations: flatMap, map\n", "\n", "# split each line into words and concatenate the result into single RDD\n", "rawlatinwordsRDD = rplatinlinesRDD.flatMap(lambda line: line.split(\" \"))\n", "\n", "print(rawlatinwordsRDD.getNumPartitions())\n", "print(rawlatinwordsRDD.take(20))\n", "\n", "# remove punctuation characters from each word\n", "latinwordsRDD = rawlatinwordsRDD.map(\n", " lambda word: ''.join(ch for ch in word if ch not in \".,:; '\")\n", ")\n", "print(latinwordsRDD.take(20))" ] }, { "cell_type": "code", "execution_count": null, "id": "37a25445-3296-4d08-8330-921fc914ddc1", "metadata": {}, "outputs": [], "source": [ "# print the RDD lineage (dependency information)\n", "print(latinwordsRDD.toDebugString().decode(\"utf-8\"))" ] }, { "cell_type": "code", "execution_count": null, "id": "460ead6e-5186-442c-bb62-d7720c446135", "metadata": {}, "outputs": [], "source": [ "# Transformation: reduceByKey\n", "\n", "# count number of all words\n", "print(\n", " latinwordsRDD.map(lambda word: (\"word\", 1)).reduceByKey(lambda a, b: a + b).collect()\n", ")\n", "# count number of occurrences for each word\n", "wordcountRDD = latinwordsRDD.map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)\n", "\n", "wordcountRDD.take(20)" ] }, { "cell_type": "code", "execution_count": null, "id": "7666e0ee-5f0e-4b89-a901-07b1a24e076f", "metadata": {}, "outputs": [], "source": [ "# Using accumulator\n", "\n", "# create an accumulator\n", "accum = sc.accumulator(0)\n", "\n", "# increment the accumulator for every word - foreach is an ACTION\n", "latinwordsRDD.foreach(lambda x: accum.add(1))\n", "\n", "# print the accumulated value\n", "print(accum.value)" ] }, { "cell_type": "code", "execution_count": null, "id": "07995ee8-6215-4dbb-b189-0fd8860cd516", "metadata": {}, "outputs": [], "source": [ "# Transformation: filter\n", "\n", "# keep only words that have at least 20 occurrences\n", "filteredcountRDD = wordcountRDD.filter(lambda x: x[1] >= 20)\n", "\n", "filteredcountRDD.take(20)" ] }, { "cell_type": "code", "execution_count": null, "id": "c401a9fa-4c00-424e-942c-f252ef84c23e", "metadata": {}, "outputs": [], "source": [ "# Caching RDD\n", "\n", "print(filteredcountRDD.toDebugString().decode(\"utf-8\"))\n", "# mark RDD for caching\n", "filteredcountRDD.cache()\n", "print(filteredcountRDD.toDebugString().decode(\"utf-8\"))\n", "# unmark the RDD for caching and remove cached data if any\n", "filteredcountRDD.unpersist()\n", "print(filteredcountRDD.toDebugString().decode(\"utf-8\"))" ] }, { "cell_type": "code", "execution_count": null, "id": "c3ac59f7-5b3a-4074-b517-4c8d082c64a3", "metadata": { "scrolled": true }, "outputs": [], "source": [ "# Transformation: sortBy\n", "\n", "sortedwordsRDD = filteredcountRDD.sortBy(lambda x: x[1], ascending=False)\n", "# return the first the 50 most frequent words\n", "sortedwordsRDD.take(50)" ] }, { "cell_type": "code", "execution_count": null, "id": "a564d82c-25ee-4c6c-9915-5d721d19f6ea", "metadata": {}, "outputs": [], "source": [ "import json\n", "\n", "with open(\"/datasets/latin_dict.json\", \"r\", encoding=\"utf-8\") as f:\n", " dictlines = f.read()\n", "\n", "dictMap = json.loads(dictlines)\n", "\n", "print(dictMap)" ] }, { "cell_type": "code", "execution_count": null, "id": "081b41d6-ed24-4b7e-b0cc-fe9e2c442e43", "metadata": {}, "outputs": [], "source": [ "# Broadcasting (small) reference data\n", "\n", "broadcastDict = sc.broadcast(dictMap)\n", "broadcastDict.value" ] }, { "cell_type": "code", "execution_count": null, "id": "80475f73-f999-4174-bc45-db00cde24e1b", "metadata": {}, "outputs": [], "source": [ "# Lookup achieved by filter / map combination\n", "\n", "# translate sorted words using the broadcast dictionary\n", "# keep only words for which translation exists\n", "\n", "translatedRDD = sortedwordsRDD.filter(\n", " lambda pair: pair[0] in broadcastDict.value\n", ").map(\n", " lambda pair: (pair[0], broadcastDict.value[pair[0]])\n", ")\n", "\n", "translatedRDD.take(50)" ] }, { "cell_type": "code", "execution_count": null, "id": "e10ad6a1-d891-494d-bafc-73b34e51ee27", "metadata": {}, "outputs": [], "source": [ "print(translatedRDD.toDebugString().decode(\"utf-8\"))" ] }, { "cell_type": "code", "execution_count": null, "id": "cdc4e1b3-3f08-40e3-875d-adc86cbe2db1", "metadata": {}, "outputs": [], "source": [ "# map & filter\n", "\n", "#keep words with at least 7 characters\n", "longwordsRDD = latinwordsRDD.filter(lambda word: len(word) >= 7)\n", "\n", "# create (word root, original word) pairs\n", "# remove the last four characters to obtain the word root\n", "wordrootsRDD = longwordsRDD.map(lambda word: (word[:-4], word))\n", "\n", "wordrootsRDD.take(20)" ] }, { "cell_type": "code", "execution_count": null, "id": "2ecd2450-1a27-4b14-98cb-58970bb6c624", "metadata": {}, "outputs": [], "source": [ "# reducebyKey\n", "\n", "conjwordsRDD = wordrootsRDD.reduceByKey(lambda a, b: a + \":\" + b)\n", "conjwordsRDD.take(3)" ] }, { "cell_type": "code", "execution_count": null, "id": "397ba8b2-a1d2-480f-9963-86ef3a7ffe67", "metadata": {}, "outputs": [], "source": [ "spark.stop()" ] } ], "metadata": { "kernelspec": { "display_name": "PySpark (Standalone Cluster)", "language": "python", "name": "pyspark" }, "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.13" } }, "nbformat": 4, "nbformat_minor": 5 }