0
# Apache Spark SQL
1

2
Apache Spark SQL is a distributed SQL engine and structured data processing framework that provides a programming interface for manipulating structured data using SQL queries and DataFrame/Dataset APIs. It serves as the foundation for relational query processing in the Apache Spark ecosystem, offering a unified engine that can execute SQL queries, work with various data sources, and integrate seamlessly with Spark's distributed computing capabilities.
3

4
## Package Information
5

6
- **Package Name**: org.apache.spark:spark-sql_2.12
7
- **Package Type**: maven
8
- **Language**: Scala (with Java interoperability)
9
- **Installation**: `libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.4.8"`
10

11
## Core Imports
12

13
```scala
14
import org.apache.spark.sql.{SparkSession, DataFrame, Dataset, Column, Row}
15
import org.apache.spark.sql.functions._
16
import org.apache.spark.sql.types._
17
```
18

19
For Java:
20

21
```java
22
import org.apache.spark.sql.SparkSession;
23
import org.apache.spark.sql.Dataset;
24
import org.apache.spark.sql.Row;
25
import static org.apache.spark.sql.functions.*;
26
```
27

28
## Basic Usage
29

30
```scala
31
import org.apache.spark.sql.{SparkSession, DataFrame}
32
import org.apache.spark.sql.functions._
33

34
// Create SparkSession (entry point)
35
val spark = SparkSession.builder()
36
  .appName("Spark SQL Example")
37
  .master("local[*]")
38
  .getOrCreate()
39

40
// Create DataFrame from data
41
val df = spark.read
42
  .option("header", "true")
43
  .option("inferSchema", "true")
44
  .csv("path/to/data.csv")
45

46
// SQL operations
47
val result = df
48
  .select("name", "age")
49
  .filter(col("age") > 18)
50
  .groupBy("department")
51
  .agg(avg("age").alias("avg_age"))
52

53
// Show results
54
result.show()
55

56
// Execute SQL queries
57
val sqlResult = spark.sql("""
58
  SELECT department, AVG(age) as avg_age 
59
  FROM employees 
60
  WHERE age > 18 
61
  GROUP BY department
62
""")
63

64
spark.stop()
65
```
66

67
## Architecture
68

69
Apache Spark SQL is built around several key components:
70

71
- **SparkSession**: Main entry point providing access to all SQL functionality and configuration
72
- **Dataset/DataFrame API**: Strongly-typed (Dataset) and untyped (DataFrame) abstractions for structured data
73
- **Catalyst Optimizer**: Query planning and optimization engine for efficient execution
74
- **SQL Parser**: Supports ANSI SQL with extensions for distributed computing
75
- **Data Sources**: Unified interface for reading/writing various formats (Parquet, JSON, CSV, JDBC, etc.)
76
- **Streaming Engine**: Real-time data processing with micro-batch execution model
77
- **Type System**: Rich type definitions with encoders for JVM object serialization
78

79
## Capabilities
80

81
### Session Management
82

83
Core session management and configuration for Spark SQL applications. SparkSession serves as the main entry point providing access to all functionality.
84

85
```scala { .api }
86
object SparkSession {
87
  def builder(): Builder
88
  def active: SparkSession
89
}
90

91
class SparkSession {
92
  def conf: RuntimeConfig
93
  def catalog: Catalog
94
  def udf: UDFRegistration
95
  def sql(sqlText: String): DataFrame
96
  def read: DataFrameReader
97
  def readStream: DataStreamReader
98
}
99
```
100

101
[Session Management](./session-management.md)
102

103
### Dataset and DataFrame Operations
104

105
Core data structures and operations for structured data manipulation. Dataset provides type-safe operations while DataFrame offers untyped flexibility.
106

107
```scala { .api }
108
class Dataset[T] {
109
  def select(cols: Column*): DataFrame
110
  def filter(condition: Column): Dataset[T]
111
  def groupBy(cols: Column*): RelationalGroupedDataset
112
  def join(right: Dataset[_]): DataFrame
113
  def agg(expr: Column, exprs: Column*): DataFrame
114
}
115

116
type DataFrame = Dataset[Row]
117
```
118

119
[Dataset and DataFrame Operations](./dataset-dataframe.md)
120

121
### Column Operations and SQL Functions
122

123
Column expressions and built-in SQL functions for data transformation. Provides both operator overloading and function-based APIs.
124

125
```scala { .api }
126
class Column {
127
  def ===(other: Any): Column
128
  def >(other: Any): Column  
129
  def &&(other: Column): Column
130
  def as(alias: String): Column
131
  def cast(to: DataType): Column
132
}
133

134
object functions {
135
  def col(colName: String): Column
136
  def lit(literal: Any): Column
137
  def when(condition: Column, value: Any): Column
138
  def avg(e: Column): Column
139
  def sum(e: Column): Column
140
}
141
```
142

143
[Column Operations and Functions](./columns-functions.md)
144

145
### Data I/O Operations
146

147
Reading and writing data from various sources and formats. Supports batch and streaming data with extensive configuration options.
148

149
```scala { .api }
150
class DataFrameReader {
151
  def format(source: String): DataFrameReader
152
  def option(key: String, value: String): DataFrameReader
153
  def schema(schema: StructType): DataFrameReader
154
  def load(): DataFrame
155
  def json(path: String): DataFrame
156
  def parquet(path: String): DataFrame
157
  def csv(path: String): DataFrame
158
}
159

160
class DataFrameWriter[T] {
161
  def mode(saveMode: SaveMode): DataFrameWriter[T]
162
  def format(source: String): DataFrameWriter[T]
163
  def save(path: String): Unit
164
  def saveAsTable(tableName: String): Unit
165
}
166
```
167

168
[Data I/O Operations](./data-io.md)
169

170
### Streaming Operations
171

172
Real-time data processing with structured streaming. Provides unified batch and streaming APIs with exactly-once processing guarantees.
173

174
```scala { .api }
175
class DataStreamReader {
176
  def format(source: String): DataStreamReader
177
  def option(key: String, value: String): DataStreamReader
178
  def load(): DataFrame
179
  def kafka(): DataFrame
180
  def socket(host: String, port: Int): DataFrame
181
}
182

183
class DataStreamWriter[T] {
184
  def outputMode(outputMode: String): DataStreamWriter[T]
185
  def trigger(trigger: Trigger): DataStreamWriter[T]
186
  def start(): StreamingQuery
187
  def foreach(writer: ForeachWriter[T]): DataStreamWriter[T]
188
}
189
```
190

191
[Streaming Operations](./streaming.md)
192

193
### Type System and Encoders
194

195
Data type definitions and encoders for converting between JVM objects and Spark SQL internal representations.
196

197
```scala { .api }
198
object DataTypes {
199
  val StringType: DataType
200
  val IntegerType: DataType
201
  val DoubleType: DataType
202
  val BooleanType: DataType
203
  val DateType: DataType
204
  val TimestampType: DataType
205
}
206

207
object Encoders {
208
  def STRING: Encoder[String]
209
  def INT: Encoder[Int]
210
  def product[T <: Product]: Encoder[T]
211
  def bean[T](beanClass: Class[T]): Encoder[T]
212
}
213

214
trait Row {
215
  def getString(i: Int): String
216
  def getInt(i: Int): Int
217
  def getDouble(i: Int): Double
218
  def isNullAt(i: Int): Boolean
219
}
220
```
221

222
[Type System and Encoders](./types-encoders.md)
223

224
### User-Defined Functions
225

226
Registration and usage of custom user-defined functions (UDFs) and user-defined aggregate functions (UDAFs).
227

228
```scala { .api }
229
class UDFRegistration {
230
  def register[RT](name: String, func: () => RT): UserDefinedFunction
231
  def register[RT, A1](name: String, func: A1 => RT): UserDefinedFunction
232
  def register[RT, A1, A2](name: String, func: (A1, A2) => RT): UserDefinedFunction
233
}
234

235
class UserDefinedFunction {
236
  def apply(exprs: Column*): Column
237
}
238
```
239

240
[User-Defined Functions](./udfs.md)
241

242
### Metadata and Catalog Operations
243

244
Database, table, and metadata management through the catalog interface. Provides programmatic access to metastore operations.
245

246
```scala { .api }
247
trait Catalog {
248
  def currentDatabase(): String
249
  def setCurrentDatabase(dbName: String): Unit
250
  def listDatabases(): Dataset[Database]
251
  def listTables(): Dataset[Table]
252
  def listColumns(tableName: String): Dataset[Column]
253
  def tableExists(tableName: String): Boolean
254
  def cacheTable(tableName: String): Unit
255
}
256
```
257

258
[Metadata and Catalog](./catalog.md)