Spark Streaming

%spark

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
import spark.implicits._

// Step 1: Read data from Kafka
val kafkaStream = spark.readStream
.format(“kafka”)
.option(“kafka.bootstrap.servers”, “localhost:9092”)
.option(“subscribe”, “clickstreamdata”)
.option(“startingOffsets”, “earliest”)
.load()

val columnNames = Array(
“user_id”, “session_id”, “timestamp”, “session_start_time”, “session_end_time”,
“url_visited”, “referrer_url”, “click_path”, “page_load_time_ms”, “scroll_depth_pct”,
“button_clicks”, “form_submissions”, “exit_page”, “time_spent_secs”,
“transactions”, “signups”, “product_ids”, “product_categories”, “error_codes”
)

val parsedStream = kafkaStream
.selectExpr(“CAST(value AS STRING) as csv_value”)
.withColumn(“csv_array”, split(col(“csv_value”), “,(?=(?:[^\”]*\”[^\”]*\”)*[^\”]*$)”))
.filter(size(col(“csv_array”)) === columnNames.length)

// Apply schema to split columns
val structuredStream = columnNames.zipWithIndex.foldLeft(parsedStream) {
case (df, (colName, idx)) => df.withColumn(colName, trim(col(“csv_array”).getItem(idx)))
}.drop(“csv_array”, “csv_value”)

// Optional: Cast specific columns to correct data types

val filteredStream = structuredStream.filter($”user_id” =!= “user_id”)

val finalStream = filteredStream
.withColumn(“page_load_time_ms”, col(“page_load_time_ms”).cast(“int”))
.withColumn(“scroll_depth_pct”, col(“scroll_depth_pct”).cast(“int”))
.withColumn(“time_spent_secs”, col(“time_spent_secs”).cast(“int”))
.withColumn(“transactions”, col(“transactions”).cast(“int”))
.withColumn(“signups”, col(“signups”).cast(“int”))
.withColumn(“event_timestamp”, to_timestamp(col(“timestamp”), “yyyy-MM-dd HH:mm:ss”))

// Optional: You can add filters or transformations here if needed

// Step 4: Write to Parquet sink
val parquetOutputPath = “/home/dataengineer/clickstream/parquetoutput”
val checkpointPath = “/home/dataengineer/clickstream/checkpnt”

finalStream.writeStream
.format(“parquet”)
.option(“path”, parquetOutputPath)
.option(“checkpointLocation”, checkpointPath)
.outputMode(“append”)
.start()

Spark Core

%spark

val parquetDF = spark.read.parquet(“/home/dataengineer/clickstream/parquetoutput/”)

val jdbcUrl = “jdbc:mysql://localhost:3306/logsdb” // Replace with your DB details
val dbTable = “clickstream_data” // Replace with your target table name
val dbUser = “root” // Replace with your MySQL username
val dbPassword = “password” // Replace with your MySQL password

// Save DataFrame to MySQL
parquetDF.write
.format(“jdbc”)
.option(“url”, jdbcUrl)
.option(“dbtable”, dbTable)
.option(“user”, dbUser)
.option(“password”, dbPassword)
.option(“driver”, “com.mysql.cj.jdbc.Driver”) // Use com.mysql.jdbc.Driver for older versions
.mode(“append”) // use “overwrite” to replace existing data
.save()