GCP_Youtube_Trends_GCS_Iceberg_BigQuery_Pipeline/etl_engagement_ratio.py at main · ViinayKumaarMamidi/GCP_Youtube_Trends_GCS_Iceberg_BigQuery_Pipeline · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# etl_engagement_ratio.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, sum as spark_sum
from pyspark.sql.functions import col
from pyspark.sql.types import *

GCS_WAREHOUSE = "gs://yt-iceberg-warehouse1/"


spark = (SparkSession.builder
    .appName("YTEngagementRatioETL")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.sql.catalog.hadoop_catalog",
            "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.hadoop_catalog.type", "hadoop")
    .config("spark.sql.catalog.hadoop_catalog.warehouse", GCS_WAREHOUSE)
    .getOrCreate()
)

raw = spark.read.format("iceberg").load("hadoop_catalog.default.raw_events")
eng  = raw.filter(col("type") == "engagement")

# roll up like/view ratio per region per minute
ratio = (eng
    .filter(col("event_type").isin("view", "like"))
    .withWatermark("event_timestamp", "5 minutes")
    .groupBy(
      col("region"),
      window("event_timestamp", "1 minute").alias("minute")
    )
    .agg(
      spark_sum((col("event_type")=="view").cast("integer")).alias("views"),
      spark_sum((col("event_type")=="like").cast("integer")).alias("likes")
    )
    .select(
      col("region"),
      col("minute.start").alias("window_start"),
      (col("likes")/col("views")).alias("like_view_ratio")
    )
)

# Write the engagement_ratios data into ICeberg Table in GCS location using overwrite mode

(ratio.write
    .format("iceberg")
    .mode("overwrite")
    .saveAsTable("hadoop_catalog.default.engagement_ratios")
)