Getting Started with Spark Structured Streaming - Current 22

Introduction to Spark Structured Streaming. Presented at Current 22 conference.

1. Getting Started with Spark Structured Streaming Dustin Vannoy dustinvannoy.com

2. Dustin Consultant – Data Engineer /in/ dustinvannoy youtube.com/DustinVannoy dustinvannoy.com Data Engineering SD meetup Technologies Azure & AWS Apache Spark Apache Kafka Azure Synapse Analytics Python & Scala Vannoy @dustinvannoy

3. Apache Spark 3

4. Why Spark? Batch and Streaming framework that has large community and wide adoption. Spark is a leader in data processing that scales across many machines. It is relatively easy to get started.

5. Yeah, but what is Spark? Spark consists of a programming API and execution engine Worker Worker Worker Worker Master from pyspark.sql import SparkSession from pyspark.sql.functions import col spark = SparkSession.builder.getOrCreate ()    song_df = spark.read \ .option(' sep ','\t') \ .option(" inferSchema ","true") \ .csv("/ databricks -datasets/songs/data-001/part-0000*")   tempo_df = song_df.select ( col('_c4').alias(' artist_name '), col('_c14').alias('tempo'), )   avg_tempo_df = tempo_df \ . groupBy (' artist_name ') \ .avg('tempo') \ . orderBy ('avg(tempo)',ascending=False)   avg_tempo_df.show (truncate=False)

6. Simple code, parallel compute Worker Controller Worker Worker

7. What is Spark? Fast, general purpose engine for large-scale data processing Replaces MapReduce as Hadoop parallel programming API Many options: Yarn / Spark Cluster / Local Scala / Python / Java / R Spark Core / SQL / Streaming / MLLib / Graph

8. What is Spark Structured Streaming? Alternative to traditional Spark Streaming which used DStreams If you are familiar with Spark, it is best to think of Structured Streaming as Spark SQL API but for streaming Use import spark.sql.streaming

9. What is Spark Structured Streaming? Tathagata Das “TD” - Lead Developer on Spark Streaming “Fast, fault-tolerant, exactly-once stateful stream processing without having to reason about streaming" "The simplest way to perform streaming analytics is not having to reason about streaming at all" A table that is constantly appended with each micro-batch Reference : https://youtu.be/rl8dIzTpxrI

10. Lazy Evaluation with Query Optimization Transformations not run immediately Builds up a DAG Optimizes steps Runs when an Action is called

11. Why use Spark Structured Streaming? Run Spark as batch or streaming Open-source with many connectors Stateful streaming and joins Kafka is not required for every stream

12. Apache Spark Options 12

13. Unified Analytics Platform Managed Apache Spark Performance optimizations Auto-scale and auto-terminate AWS, Azure, GCP Databricks

14. Amazon EMR Azure Synapse Spark Google Cloud Proc Cloud Specific Options

15. Install on Mac or Linux Use WSL on Windows Docker Container PySpark package Local

16. Spark and storage together Many containers per worker Use what you have Hadoop / YARN

17. On-premises or multi-cloud Scale size with familiar tools Use what you know Kubernetes

18. Apache Spark Syntax 18

19. Spark Example - Read df = spark. read .format(" kafka ") .options(**config) .load() df = spark. readStream .format(" kafka ") .options(**config) .load() Batch Streaming

20. Spark Example - Write df. write .format(" kafka ") … df. writeStream .format(" kafka ") … Batch Streaming

21. Structured Streaming - Read df = spark.readStream .format("delta") .options(** delta_config ) .load(path)

22. Structured Streaming - Transformations new_df = df.withColumn (" new_column ", concat ("col1", "col2" ))

23. Structured Streaming - Transformations new_df = df.selectExpr ( " * ", " concat (col1, '-',col2) as new_col ", " current_timestamp () as load_tm ")

24. Structured Streaming - Actions writeStream.format (“console”).start() writeStream.format (“ kafka ”).start() writeStream.foreach (…).start() writeStream.foreachBatch (…).start()

25. Structured Streaming - Write df.writeStream .format(" kafka ") .options(** producer_config ) .option(" checkpointLocation ","/ tmp /cp001") .start()

26. Structured Streaming – Output Mode How to output data when triggered Append - Just keep adding newest records Update – Only those updated since last trigger Complete mode - Output latest state of table (useful for aggregation results)

27. Example from docs (Structured Streaming Programming Guide)

28. Structured Streaming – Write with Output Mode df.writeStream .format(" kafka ") . outputMode ("complete") .options(** producer_config ) .option(" checkpointLocation ","/ tmp /cp001") .start()

29. Structured Streaming - Triggers Specify how often to check for a new data Default: micro-batch mode, as soon as possible Fixed: micro-batch mode, wait specified time Only once: just one micro-batch then job stops Continuous: continuously read from source, experimental and only works for certain types of queries

30. Structured Streaming – Write with Trigger df.writeStream .format(" kafka ") .trigger( processingTime ="5 seconds") .options(** producer_config ) .option(" checkpointLocation ","/ tmp /cp001") .start()

31. Structured Streaming - Windows Calculate over time using windows -> Avg trips every 5 minutes Tumbling window: specific period of time, such as 1:00 to 1:30 Sliding windows: fixed window, but can track multiple groups... 1:00 – 1:30, 1:05 – 1:35, 1:10 – 1:40, etc Session windows: similar to an inactive session timeout… group together until a long enough gap in time between events

32. Comparing windows (Structured Streaming Programming Guide)

33. Window Example df2.groupBy( col(" VendorID "), window(col(" pickup_dt "), "10 min")) .avg(" trip_distance ")

34. Streaming Joins Stream-static Simple join Does not tracking state Options: Inner, Left Outer, Left Semi Stream-stream Complex join Data may arrive early or late Options: Inner, Left/Right Outer, Full Outer, Left Semi

35. Stream to Static DataFrame Join zdf = spark.read.format ("delta").load(path) trip_df = ( tdf.join ( zdf , tdf [" PULocId "] == zdf [" LocID "] , how="left")

36. Stream to Stream Join trip_df = ( tdf.withWatermark ("eventTime","1 minute") .join( pmt.withWatermark (" pTime ", "1 hour"), expr(""" LocId = PULocId AND zTime >= eventTime AND zTime <= eventTime + interval 1 hour """), how=" leftOuter ")

37. Stateful Streaming Aggregations, dropDuplicates , and Joins will require storing state Custom stateful operations possible HDFS or RocksDB State Store (Spark 3.2) If retaining many keys, HDFSBackedStateStore is likely to have JVM memory issues

38. Example Structured Streaming Application 38

39. Databricks Notebook - Setup

40. Databricks Notebook – Confluent Cloud Config

41. Databricks Notebook – Spark Schema

42. Databricks Notebook – Read Kafka Record

43. Databricks Notebook – Convert JSON

44. Databricks Notebook – Read Lookup Table

45. Databricks Notebook – Join and write

46. Databricks Notebook – Check destination

47. Monitoring 47

48. SQL UI

49. SQL UI - Detail

50. Structured Streaming UI

51. Structured Streaming UI

52. Notebook Streaming UI - Timeseries

53. Notebook Streaming UI – Query Progress

54. Cloud Logging Services

55. Why Kafka? Streaming data directly from one system to another often problematic Kafka serves as the scalable broker, keeping up with producer and persisting data for all consumers

56. The Log “It is an append-only, totally-ordered sequence of records ordered by time.” - Jay Kreps Reference: The Log - Jay Kreps