diff --git a/business_model.md b/business_model.md new file mode 100644 index 0000000..3ab2cb8 --- /dev/null +++ b/business_model.md @@ -0,0 +1,43 @@ +Yes. Most business questions map to a model family once you work out the unit of analysis (who or what is being counted) and the type of outcome. Here's the mapping using your Trakt and Spotify examples. + +**Who** +- *Questions:* Who are my heavy skippers? Who is most like whom? +- *Frameworks:* clustering (K-means) for segments, logistic models for propensity (e.g. who completes shows), similarity scores (Jaccard, cosine) for how alike two users are. +- *Example:* Spotify user A vs. B overlap, or the viewing-day personas from K-means. + +**What** +- *Questions:* What should we recommend? What drives completion? +- *Frameworks:* recommenders (item-item, Word2Vec), association rules (FP-growth), and GLM coefficients that rank drivers. +- *Example:* tracks B plays that A hasn't heard, or runtime's effect on finishing a show. + +**When** +- *Questions:* When will demand peak? How long until someone abandons a show? When did habits change? +- *Frameworks:* time series (ETS, Prophet) for forecasts, survival analysis for time-to-event, Poisson with seasonality terms, change-point detection. +- *Example:* predicted days until a 50% chance of abandoning a series. + +**Where** +- *Questions:* Which region, device or channel? +- *Frameworks:* group-level GLMs with country or platform terms, hierarchical models to shrink small-sample regions toward the average. +- *Example:* Spotify's country and platform fields. Trakt has almost no "where", so there it's mostly a time-zone question. + +**How many / how much** +- *Frameworks:* Poisson or negative binomial for counts, gamma for skewed amounts, with an offset for exposure. +- *Example:* plays per day on weekends vs. weekdays. + +**Why** +- *Frameworks:* GLM coefficients show association. Causal designs (A/B tests, difference-in-differences) show effect. +- *Example:* "Did B hearing it first cause A's plays?" needs more than a coefficient. + +**The shortcut:** ask what the outcome looks like. +- A count → Poisson +- Yes/no → logistic +- Time until something → survival +- A skewed amount → gamma +- Groups of similar things → clustering +- Next best item → recommender + +**Two cautions** +- Plain SQL answers many who/what/when/where questions descriptively (top artists, busiest days). Reach for a model when you need prediction, comparison with uncertainty, or an effect size. +- Models give probabilities and rates, not certainties, and sparse data (one or two users) limits which ones are valid. + +Want to take one real business question from your data and walk it through question → outcome type → model → features? \ No newline at end of file diff --git a/etl_spark.md b/etl_spark.md new file mode 100644 index 0000000..34da1bd --- /dev/null +++ b/etl_spark.md @@ -0,0 +1,490 @@ +# Building an ETL Pipeline with PySpark and Spark SQL + +**Setup:** `pip install pyspark`. Java 8, 11, or 17 must be installed. + +--- + +## Extract Phase + +The extract phase is the first step in an ETL pipeline, where raw data is collected from files, databases, and APIs. In PySpark, every extract starts with a `SparkSession`, the entry point for DataFrames and Spark SQL. + +In this task you will learn how to: + +- Create a SparkSession +- Read flat files into Spark DataFrames +- Read data from a SQL database over JDBC +- Read data from REST APIs into Spark DataFrames + +To get started open the `extract.py` file. + +### Task 1.0.1: Create a SparkSession + +```python +from pyspark.sql import SparkSession + +spark = ( + SparkSession.builder + .appName("etl_pipeline") + .config("spark.jars.packages", "com.mysql:mysql-connector-j:8.3.0") # MySQL JDBC driver, used in Task 1.0.3 + .getOrCreate() +) +``` + +`.master("local[*]")` runs Spark on your machine. On a real cluster you would remove `.master()` and let the cluster manager supply it. + +0 / 1 checks: Create a SparkSession named `spark`. + +### Task 1.0.2: Extracting Data from Flat Files + +Spark reads these formats natively: + +- `.csv` +- `.json` +- `.parquet` + +Excel is not supported natively, so you would read it with Pandas and convert it (see Task 1.0.4). + +For this task: + +1. Read `data/example.csv` into `df1` with `spark.read.csv`. Set `header=True` and `inferSchema=True`, otherwise every column is read as a string. +2. Read `data/example.json` into `df2` with `spark.read.json`. +3. Read `data/example.parquet` into `df3` with `spark.read.parquet`. + +```python +df1 = spark.read.csv("data/example.csv", header=True, inferSchema=True) +df2 = spark.read.json("data/example.json") +df3 = spark.read.parquet("data/example.parquet") + +df1.printSchema() +df1.show(5) +``` + +0 / 1 checks: Read different file types into Spark DataFrames using `spark.read`. + +### Task 1.0.3: Loading Multiple Files with Wildcards + +Spark is built for parallel processing, so reading many files at once is a core use case. Spark will split the files into partitions and process them across your cores or cluster. + +1. Observe the `dask_data_*.csv` files in the `data` folder. +2. Assign `df4` the result of: + +```python +df4 = spark.read.csv("data/dask_data_*.csv", header=True, inferSchema=True) +``` + +3. Optionally inspect metadata about the DataFrame: + +```python +print(df4.count()) # row count (triggers a job) +print(len(df4.columns)) # column count +print(df4.rdd.getNumPartitions()) # how many partitions Spark created +``` + +0 / 1 checks: Read the `dask_data` files in `data` into the `df4` DataFrame. + +### Task 1.0.4: Extract Data from Database Tables (JDBC) + +Spark reads databases through JDBC. Instead of a SQLAlchemy engine, you pass connection options to `spark.read.format("jdbc")`. + +1. Assign the query to the `query` variable: + +```python +query = "SELECT TABLE_NAME FROM information_schema.TABLES" +``` + +2. Read the results into `df5`: + +```python +df5 = ( + spark.read.format("jdbc") + .option("url", "jdbc:mysql://localhost:3306/mydb") + .option("query", query) + .option("user", "root") + .option("password", "your_password") + .option("driver", "com.mysql.cj.jdbc.Driver") + .load() +) +df5.show() +``` + +For large tables, add `partitionColumn`, `lowerBound`, `upperBound`, and `numPartitions` so Spark reads in parallel rather than through a single connection. + +0 / 1 checks: Read the provided query results into the `df5` Spark DataFrame. + +### Task 1.0.5: Extracting REST API Data with Requests + +Spark has no built-in REST reader. The common pattern is to fetch the JSON with `requests` on the driver, then convert it into a DataFrame. + +1. Open a new terminal tab and run `python3 scripts/app.py` to start the Flask app. +2. Fetch the data and create a DataFrame: + +```python +import requests + +url = "http://localhost:5000/random-user" +r = requests.get(url) +print(r.json()) + +df6 = spark.createDataFrame([r.json()]) # a dict becomes a one-row DataFrame +df6.show() +``` + +For a list of records, pass the list directly: `spark.createDataFrame(r.json())`. + +### Task 1.0.6: Authenticating with Requests + +Add a Bearer token to the request headers to authenticate to the `/protected` endpoint. + +1. In the `headers` dictionary, add the key `Authorization`. +2. Set the value to `Bearer mysecrettoken123`. +3. Assign the result to `response`: + +```python +headers = {"Authorization": "Bearer mysecrettoken123"} +response = requests.get("http://localhost:5000/protected", headers=headers) +df7 = spark.createDataFrame([response.json()]) +``` + +0 / 1 checks: Assign the header values and authenticate to the `/protected` endpoint, assigning the results to `response`. + +### (Optional) Reading JSON strings with a schema + +`createDataFrame` infers types, which can be fragile with nested or inconsistent API data. Define a schema explicitly for predictable results: + +```python +from pyspark.sql.types import StructType, StructField, StringType, IntegerType + +schema = StructType([ + StructField("name", StringType(), True), + StructField("age", IntegerType(), True), +]) +df8 = spark.createDataFrame(response.json(), schema=schema) +``` + +--- + +## Transform Phase + +The transform phase is where raw data is cleansed, standardized, aggregated, and joined. PySpark gives you two interchangeable APIs for this: the **DataFrame API** (`pyspark.sql.functions`) and **Spark SQL**. Both compile to the same optimized execution plan, so use whichever reads better. + +In this task you will learn how to: + +- Convert data types and date formats +- Replace values and clean dirty data with regex +- Filter, aggregate, and create calculated columns +- Join DataFrames +- Run the same logic with Spark SQL using temp views + +**Example Data* + +```python +data = { + "Name": ["Adam", "Bob", "Charlie"], + "Children": [1, 2, 0], + "ZipCode":[10001,90001,60601], + "City": ["New York", "Los Angeles", "Chicago"], + "Birthdate": [pd.Timestamp('20000101'),pd.Timestamp('19900418'),pd.Timestamp('20180310')], + "CharacterSeq": ['****','____','****'], + "UserId":['A111','B222','C333'], + "TShirtSize":['M','XL','XL'], + "Salary":[50000,60000,70000] +} +``` + +```python +from pyspark.sql import functions as F +from pyspark.sql.types import StringType, DoubleType +``` + +### Task 2.0.1: Convert Column Data Types + +Observe the data dictionary. Notice that `ZipCode` is an integer. Convert it to a string with `cast`: + +```python +df = df.withColumn("ZipCode", F.col("ZipCode").cast(StringType())) +df.printSchema() +``` + +DataFrames are immutable in Spark. `withColumn` returns a new DataFrame, so you must assign the result back. + +0 / 1 checks: Convert the `ZipCode` column to a string data type. + +### Task 2.0.2: Convert Date Formats + +`Birthdate` is in `yyyy-MM-dd` format. We want `MM-dd-yyyy`. Parse the string into a date, then format it back out: + +```python +df = df.withColumn( + "Birthdate", + F.date_format(F.to_date("Birthdate", "yyyy-MM-dd"), "MM-dd-yyyy") +) +df.select("Birthdate").show(5) +``` + +Spark date patterns are case-sensitive. `MM` is month and `mm` is minutes. + +0 / 1 checks: Convert the `Birthdate` column to the `MM-dd-yyyy` format. + +### Task 2.0.3: Replace Values with a Mapping + +1. Observe the `replace_map` dictionary. The value being replaced comes first, followed by the replacement. +2. Apply it to the `City` column: + +```python +replace_map = {"NYC": "New York", "LA": "Los Angeles"} +df = df.replace(replace_map, subset=["City"]) +``` + +0 / 1 checks: Convert the existing `City` values using `df.replace()`. + +### Task 2.0.4: Using Regex to Replace Dirty Data + +Use `regexp_replace` to find and replace patterns. `*` is a special regex character, so it must be escaped. + +1. Observe the `CharacterSeq` column. +2. Write a regex pattern that matches four asterisks: `\*{4}` (or `\*\*\*\*`). +3. Replace them with underscores: + +```python +df = df.withColumn("CharacterSeq", F.regexp_replace("CharacterSeq", r"\*{4}", "____")) +``` + +0 / 1 checks: Use a regex on `CharacterSeq` to replace `****` with `____`. + +### (Optional) DataFrame Filtering + +```python +# Column expression syntax +df.filter(F.col("Name") == "Adam") + +# SQL string syntax +df.filter("Name = 'Adam'") + +# Assign to a new DataFrame for a sub-dataset +adams = df.filter(F.col("Name") == "Adam") +adams.show() +``` + +`where()` is an alias for `filter()`. + +### Task 2.0.5: Creating Custom Calculated Columns + +1. Observe the `cost_of_living_adjust` function, which takes an original value and an `inc_percent`. +2. Wrap it in a UDF (user-defined function) and create `Salary_COL`: + +```python +def cost_of_living_adjust(value, inc_percent): + return value * (1 + inc_percent / 100) if value is not None else None + +col_udf = F.udf(lambda s: cost_of_living_adjust(s, 5), DoubleType()) +df = df.withColumn("Salary_COL", col_udf(F.col("Salary"))) +``` + +Python UDFs are slow because every row is serialized between the JVM and Python. When you can, use native column expressions, which Spark can optimize: + +```python +df = df.withColumn("Salary_COL", F.col("Salary") * (1 + 5 / 100)) +``` + +0 / 1 checks: Create a new `Salary_COL` column by applying `cost_of_living_adjust` to `Salary` with `inc_percent=5`. + +### (Optional) DataFrame Aggregating + +```python +# Whole-DataFrame aggregate +df.agg( + F.count("Salary").alias("count"), + F.min("Salary").alias("min"), + F.max("Salary").alias("max"), + F.mean("Salary").alias("mean"), + F.sum("Salary").alias("sum"), +).show() + +# Grouped aggregate +df.groupBy("TShirtSize").agg(F.mean("Salary").alias("avg_salary")).show() +``` + +### Task 2.0.6: Joining Data + +Spark's `.join()` mirrors SQL joins. The join types are `inner`, `left`, `right`, `outer`, `left_semi`, and `left_anti`. + +1. Observe both DataFrames (`df` and `df2`). +2. Use `UserId` as the `on` parameter. +3. Use `left` as the `how` parameter. +4. Assign the result to `df_merged`: + +```python +df_merged = df.join(df2, on="UserId", how="left") +df_merged.show() +``` + +Passing `on="UserId"` (a string) keeps a single `UserId` column. Joining on `df.UserId == df2.UserId` keeps both, which causes ambiguous-column errors later. + +0 / 1 checks: Use `.join` to combine data in `df` and `df2` into `df_merged`. + +### Task 2.0.7: Transforming with Spark SQL + +Spark SQL lets you query DataFrames with plain SQL. You first register a DataFrame as a temporary view: + +1. Register the view: + +```python +df_merged.createOrReplaceTempView("users") +``` + +2. Query it with `spark.sql()`, which returns a DataFrame: + +```python +result = spark.sql(""" + SELECT + City, + COUNT(*) AS user_count, + ROUND(AVG(Salary_COL), 2) AS avg_salary + FROM users + WHERE Salary_COL IS NOT NULL + GROUP BY City + HAVING COUNT(*) > 1 + ORDER BY avg_salary DESC +""") +result.show() +``` + +3. Because the result is a DataFrame, you can switch back to the DataFrame API at any point and mix the two styles freely. + +0 / 1 checks: Register `df_merged` as a temp view named `users` and aggregate it with `spark.sql()`. + +--- + +## Loading Data + +Data writing in PySpark refers to saving a Spark DataFrame to files or databases through the `df.write` interface. Spark supports `csv()`, `json()`, `parquet()`, `orc()`, and `jdbc()`, allowing you to export data to files, data lakes, or relational databases. + +There are three key differences from Pandas. Spark writes a **directory** of part files (one per partition) rather than a single file. You control overwrite behavior with **save modes** (`overwrite`, `append`, `ignore`, `errorifexists`). And **Parquet** is the preferred format for analytics because it is columnar, compressed, and keeps the schema. + +To get started open the `load.py` file. + +### Task 3.0.1: Loading Data to a CSV File + +1. Create a dictionary with the following python code: + +```python +data = { + 'Transaction_ID': ['T1001', 'T1002', 'T1003', 'T1004'], + 'Customer_Name': ['Alice Johnson', 'Bob Lee', 'Charlie Kim', 'David Clark'], + 'Product': ['Laptop', 'Smartphone', 'Tablet', 'Smartwatch'], + 'Quantity': [1, 2, 3, 1], + 'Price_Per_Unit': [1200, 800, 400, 200], + 'Total_Sale': [1200, 1600, 1200, 200], + 'Sale_Date': ['2025-03-20', '2025-03-21', '2025-03-21', '2025-03-22'] +} +``` + +2. Convert the dictionary into a Spark DataFrame variable `df`. Spark expects a list of rows, so reshape the dictionary first: + +```python +import pandas as pd + +df = spark.createDataFrame(pd.DataFrame(data)) +``` + +3. Use `df.write.csv` to write the contents to `data/Transactions`. Include the following options: + - `header=True` + - `mode="overwrite"` + +```python +( + df.coalesce(1) # collapse to one partition, so one part file + .write + .mode("overwrite") + .option("header", True) + .csv("data/Transactions") +) +``` + +Spark creates a folder named `Transactions` containing `part-00000-....csv` and a `_SUCCESS` marker. `coalesce(1)` is fine for small data, but avoid it on large datasets because it funnels everything through one task. + +0 / 1 checks: Convert the dictionary values to a DataFrame and load it to `Transactions` in the `data` directory. + +### Task 3.0.2: Loading Data to Parquet + +```python +df.write.mode("overwrite").parquet("data/Transactions_parquet") +``` + +Partitioning by a column splits the output into subfolders, which speeds up later filtered reads: + +```python +df.write.mode("overwrite").partitionBy("Sale_Date").parquet("data/Transactions_by_date") +``` + +0 / 1 checks: Write the DataFrame to Parquet. + +### Task 3.0.3: Loading Data to a Database with JDBC + +```python +( + df.write.format("jdbc") + .option("url", "jdbc:mysql://localhost:3306/mydb") + .option("dbtable", "transactions") + .option("user", "root") + .option("password", "your_password") + .option("driver", "com.mysql.cj.jdbc.Driver") + .mode("append") + .save() +) +``` + +Use `append` to add rows. `overwrite` drops and recreates the table unless you also set `.option("truncate", True)`. + +0 / 1 checks: Load the DataFrame into a `transactions` table. + +### (Optional) Validating Data Operations + +Use `assert` statements to validate that the pipeline finished successfully. Check that row counts, column types, and totals match the expected results. + +```python +csv_df = spark.read.csv("data/Transactions", header=True, inferSchema=True) + +assert csv_df.count() == 4, "Row counts do not match expected values!" +print(f"Row counts match: {csv_df.count()} rows") + +assert csv_df.agg(F.sum("Total_Sale")).first()[0] == 4200, "Total sales mismatch!" + +expected_cols = {"Transaction_ID", "Customer_Name", "Product", "Quantity", + "Price_Per_Unit", "Total_Sale", "Sale_Date"} +assert set(csv_df.columns) == expected_cols, "Schema mismatch!" +``` + +To compare against the database table, read it back with `spark.read.format("jdbc")` and compare `.count()` to the source DataFrame. + +--- + +## Putting It All Together + +Wrap the three phases in functions so the pipeline is testable and reusable: + +```python +def extract(spark): + return spark.read.csv("data/example.csv", header=True, inferSchema=True) + +def transform(df): + return ( + df.withColumn("ZipCode", F.col("ZipCode").cast("string")) + .withColumn("Birthdate", F.date_format(F.to_date("Birthdate"), "MM-dd-yyyy")) + .withColumn("Salary_COL", F.col("Salary") * 1.05) + .dropDuplicates() + ) + +def load(df, path): + df.write.mode("overwrite").parquet(path) + +if __name__ == "__main__": + spark = SparkSession.builder.appName("etl").master("local[*]").getOrCreate() + raw = extract(spark) + clean = transform(raw) + load(clean, "data/output") + assert spark.read.parquet("data/output").count() == clean.count() + spark.stop() +``` + +I can also put this into a downloadable file or turn it into a Jupyter-style notebook. \ No newline at end of file diff --git a/spotify_bronze_silver.ipynb b/spotify_bronze_silver.ipynb index 684dec5..6654fc2 100644 --- a/spotify_bronze_silver.ipynb +++ b/spotify_bronze_silver.ipynb @@ -59,13 +59,14 @@ }, { "cell_type": "code", - "execution_count": 16, + "execution_count": null, "id": "a898326e", "metadata": {}, "outputs": [], "source": [ " ## Transform ##\n", "def null_counts_by_column(df):\n", + " \"\"\"Count nulls in each column of the DataFrame.\"\"\"\n", " null_counts = df.select([\n", " F.count(F.when(F.col(c).isNull(), c)).alias(c)\n", " for c in df.columns\n", @@ -73,9 +74,11 @@ " return null_counts\n", "\n", "def metadata_df(df):\n", + " \"\"\"Rename verbose album/artist/track metadata columns to short names.\"\"\"\n", " return df.withColumnRenamed(\"master_metadata_album_album_name\", \"album_name\").withColumnRenamed(\"master_metadata_album_artist_name\", \"artist_name\").withColumnRenamed(\"master_metadata_track_name\", \"track_name\")\n", "\n", "def rename_columns(df, rename_map, drop_columns):\n", + " \"\"\"Drop specified columns and rename remaining ones per the rename map.\"\"\"\n", " missing_rename = set(rename_map) - set(df.columns)\n", " missing_dropped = set(drop_columns) - set(df.columns)\n", "\n", @@ -89,21 +92,26 @@ " return df.toDF(*[rename_map.get(c, c) for c in df.columns])\n", "\n", "def track_df(df):\n", + " \"\"\"Filter to rows where track_name is not null.\"\"\"\n", " return df.where(df[\"track_name\"].isNotNull())\n", "\n", "def parse_timestamps(df: DataFrame, columns) -> DataFrame:\n", + " \"\"\"Cast each column in the list to timestamp type.\"\"\"\n", " for column in columns:\n", " df = df.withColumn(column, F.to_timestamp(column))\n", " return df\n", "\n", "def parse_booleans(df: DataFrame, columns) -> DataFrame:\n", + " \"\"\"Cast each column in the list to string type.\"\"\"\n", " df = df.select([F.col(c).cast(\"string\") for c in columns])\n", " return df\n", "\n", "def add_minutes_played(df: DataFrame, src, dest) -> DataFrame:\n", + " \"\"\"Convert milliseconds played to minutes and round to 2 decimals.\"\"\"\n", " return (df.withColumn(dest, F.round(F.col(src) / 60000, 2)))\n", "\n", "def add_time_parts(df: DataFrame, ts_col) -> DataFrame:\n", + " \"\"\"Add time/date/day-of-week/hour parts derived from the timestamp columns.\"\"\"\n", " return (df\n", " .withColumn(\"ts\", F.to_timestamp(\"ts\"))\n", " .withColumn(\"offline_timestamped\", F.to_timestamp(\"offline_timestamp\"))\n", @@ -116,6 +124,7 @@ " )\n", "\n", "def clean_track_df(df: DataFrame) -> DataFrame:\n", + " \"\"\"Apply the full transform pipeline: rename, filter tracks, add minutes and time parts.\"\"\"\n", " return(df\n", " .transform(rename_columns, SPOTIFY_RENAMES, MUSIC_DROP_COLUMNS)\n", " .transform(track_df)\n", diff --git a/spotify_silver_gold.ipynb b/spotify_silver_gold.ipynb index 2c7e1c3..96275e9 100644 --- a/spotify_silver_gold.ipynb +++ b/spotify_silver_gold.ipynb @@ -24,9 +24,9 @@ "source": [ " ## Extract (Load in) ##\n", "def start_spark(app_name):\n", + " \"\"\"Initialize and return a SparkSession with the PostgreSQL JDBC driver.\"\"\"\n", " jar_path = str(Path.cwd() / \"postgresql.jar\")\n", " spark = SparkSession.builder.appName(app_name).config(\"spark.jars\", jar_path).config(\"spark.driver.extraClassPath\", jar_path).getOrCreate()\n", - " spark.log.level(\"ERROR\")\n", " return spark\n", "\n", "def get_jdbc_config():\n", @@ -56,6 +56,7 @@ " return url, properties\n", "\n", "def read_jdbc(spark, table_name):\n", + " \"\"\"Read a table from Postgres via JDBC into a Spark DataFrame.\"\"\"\n", " url, properties = get_jdbc_config()\n", " return spark.read.jdbc(url=url, table=table_name, properties=properties)\n", "\n",