0
# Flink Scala API
1
2
Apache Flink Scala API provides type-safe operations and functional programming paradigms for distributed stream and batch processing applications. It offers elegant Scala APIs with case class support, pattern matching, and functional composition patterns for building scalable data processing pipelines.
3
4
**⚠️ Deprecation Notice**: All Flink Scala APIs are deprecated and will be removed in a future Flink version. You can still build your application in Scala, but you should move to the Java version of either the DataStream and/or Table API.
5
6
## Package Information
7
8
- **Package Name**: flink-scala_2.12
9
- **Package Type**: maven
10
- **Language**: Scala
11
- **Installation**: Add to Maven pom.xml:
12
13
```xml
14
<dependency>
15
<groupId>org.apache.flink</groupId>
16
<artifactId>flink-scala_2.12</artifactId>
17
<version>1.20.2</version>
18
</dependency>
19
```
20
21
## Core Imports
22
23
```scala
24
import org.apache.flink.api.scala._
25
import org.apache.flink.api.scala.ExecutionEnvironment
26
```
27
28
For specific functionality:
29
30
```scala
31
import org.apache.flink.api.scala.{DataSet, GroupedDataSet}
32
import org.apache.flink.api.scala.typeutils.Types
33
import org.apache.flink.api.scala.extensions._ // For partial function support
34
```
35
36
## Basic Usage
37
38
```scala
39
import org.apache.flink.api.scala._
40
41
// Create execution environment
42
val env = ExecutionEnvironment.getExecutionEnvironment
43
44
// Create dataset from elements
45
val data: DataSet[String] = env.fromElements("Hello", "World", "from", "Flink")
46
47
// Transform data
48
val wordCounts = data
49
.flatMap(_.toLowerCase.split("\\W+"))
50
.filter(_.nonEmpty)
51
.map((_, 1))
52
.groupBy(0)
53
.sum(1)
54
55
// Output results
56
wordCounts.print()
57
58
// Execute the program
59
env.execute("Word Count Example")
60
61
// Example with partial functions (requires extensions import)
62
import org.apache.flink.api.scala.extensions._
63
64
case class Sale(region: String, product: String, amount: Double)
65
val sales = env.fromElements(
66
Sale("US", "ProductA", 100.0),
67
Sale("EU", "ProductA", 150.0)
68
)
69
70
val result = sales
71
.filterWith { case Sale(region, _, _) => region == "US" }
72
.mapWith { case Sale(region, product, amount) => (product, amount) }
73
.groupingBy(_._1)
74
.sum(1)
75
```
76
77
## Architecture
78
79
The Flink Scala API is built around these core concepts:
80
81
- **ExecutionEnvironment**: Context for creating and executing Flink programs
82
- **DataSet**: Immutable collection representing distributed data
83
- **Transformations**: Operations like map, filter, join that create new DataSets
84
- **Actions**: Operations like collect, print that trigger execution
85
- **Type System**: Automatic TypeInformation generation for Scala types
86
87
## Capabilities
88
89
### Execution Environment
90
91
The entry point for all Flink Scala programs, providing methods to create DataSets and configure execution.
92
93
```scala { .api }
94
object ExecutionEnvironment {
95
def getExecutionEnvironment: ExecutionEnvironment
96
def createLocalEnvironment(parallelism: Int = Runtime.getRuntime.availableProcessors()): ExecutionEnvironment
97
def createRemoteEnvironment(host: String, port: Int, jarFiles: String*): ExecutionEnvironment
98
}
99
100
class ExecutionEnvironment {
101
def setParallelism(parallelism: Int): Unit
102
def getParallelism: Int
103
def getConfig: ExecutionConfig
104
105
// Data source creation
106
def fromElements[T: ClassTag: TypeInformation](data: T*): DataSet[T]
107
def fromCollection[T: ClassTag: TypeInformation](data: Iterable[T]): DataSet[T]
108
def readTextFile(filePath: String, charsetName: String = "UTF-8"): DataSet[String]
109
def readCsvFile[T: ClassTag: TypeInformation](filePath: String, /* ... */): DataSet[T]
110
def generateSequence(from: Long, to: Long): DataSet[Long]
111
112
// Execution
113
def execute(): JobExecutionResult
114
def execute(jobName: String): JobExecutionResult
115
def executeAsync(): JobClient
116
}
117
```
118
119
[Execution Environment](./execution-environment.md)
120
121
### DataSet Operations
122
123
Core data transformation and processing operations on distributed datasets.
124
125
```scala { .api }
126
class DataSet[T] {
127
// Basic transformations
128
def map[R: TypeInformation: ClassTag](fun: T => R): DataSet[R]
129
def flatMap[R: TypeInformation: ClassTag](fun: T => TraversableOnce[R]): DataSet[R]
130
def filter(fun: T => Boolean): DataSet[T]
131
def distinct(): DataSet[T]
132
133
// Grouping
134
def groupBy[K: TypeInformation](fun: T => K): GroupedDataSet[T]
135
def groupBy(fields: Int*): GroupedDataSet[T]
136
137
// Aggregations
138
def reduce(fun: (T, T) => T): DataSet[T]
139
def aggregate(agg: Aggregations, field: Int): AggregateDataSet[T]
140
def sum(field: Int): AggregateDataSet[T]
141
def max(field: Int): AggregateDataSet[T]
142
def min(field: Int): AggregateDataSet[T]
143
144
// Output operations
145
def collect(): Seq[T]
146
def print(): Unit
147
def count(): Long
148
def writeAsText(filePath: String, writeMode: FileSystem.WriteMode = FileSystem.WriteMode.NO_OVERWRITE): DataSink[T]
149
}
150
```
151
152
[DataSet Operations](./dataset-operations.md)
153
154
### Join Operations
155
156
Joining datasets on keys with various join types and optimization hints.
157
158
```scala { .api }
159
class DataSet[T] {
160
def join[O](other: DataSet[O]): UnfinishedJoinOperation[T, O]
161
def leftOuterJoin[O](other: DataSet[O]): UnfinishedOuterJoinOperation[T, O]
162
def rightOuterJoin[O](other: DataSet[O]): UnfinishedOuterJoinOperation[T, O]
163
def fullOuterJoin[O](other: DataSet[O]): UnfinishedOuterJoinOperation[T, O]
164
}
165
166
class UnfinishedJoinOperation[L, R] {
167
def where[K: TypeInformation](fun: L => K): HalfUnfinishedJoinOperation[L, R]
168
def where(fields: Int*): HalfUnfinishedJoinOperation[L, R]
169
}
170
171
class JoinDataSet[L, R] {
172
def apply[O: TypeInformation: ClassTag](fun: (L, R) => O): DataSet[O]
173
}
174
```
175
176
[Join Operations](./join-operations.md)
177
178
### Grouped DataSet Operations
179
180
Operations available on grouped datasets including sorting and specialized aggregations.
181
182
```scala { .api }
183
class GroupedDataSet[T] {
184
def sortGroup(field: Int, order: Order): GroupedDataSet[T]
185
def reduce(fun: (T, T) => T): DataSet[T]
186
def reduceGroup[R: TypeInformation: ClassTag](fun: Iterator[T] => R): DataSet[R]
187
def aggregate(agg: Aggregations, field: Int): AggregateDataSet[T]
188
def sum(field: Int): AggregateDataSet[T]
189
def max(field: Int): AggregateDataSet[T]
190
def min(field: Int): AggregateDataSet[T]
191
def maxBy(fields: Int*): DataSet[T]
192
def minBy(fields: Int*): DataSet[T]
193
def first(n: Int): DataSet[T]
194
}
195
```
196
197
[Grouped DataSet Operations](./grouped-dataset-operations.md)
198
199
### Type System
200
201
Scala-specific type information system for serialization and type safety.
202
203
```scala { .api }
204
object Types {
205
// Basic types
206
val STRING: TypeInformation[String]
207
val INT: TypeInformation[Int]
208
val LONG: TypeInformation[Long]
209
val DOUBLE: TypeInformation[Double]
210
val BOOLEAN: TypeInformation[Boolean]
211
212
// Factory methods
213
def of[T: TypeInformation]: TypeInformation[T]
214
def TUPLE[T: TypeInformation]: TypeInformation[T]
215
def CASE_CLASS[T: TypeInformation]: TypeInformation[T]
216
def OPTION[A, T <: Option[A]](valueType: TypeInformation[A]): TypeInformation[T]
217
def EITHER[A, B](leftType: TypeInformation[A], rightType: TypeInformation[B]): TypeInformation[Either[A, B]]
218
}
219
220
// Implicit type information generation (macro-based)
221
implicit def createTypeInformation[T]: TypeInformation[T] = macro TypeUtils.createTypeInfo[T]
222
```
223
224
[Type System](./type-system.md)
225
226
### Partitioned and Sorted DataSets
227
228
Specialized DataSet types for partitioned and sorted data operations.
229
230
```scala { .api }
231
class PartitionSortedDataSet[T] extends DataSet[T] {
232
def sortPartition(field: Int, order: Order): DataSet[T]
233
def sortPartition(field: String, order: Order): DataSet[T]
234
// Note: Cannot chain key selector functions
235
}
236
```
237
238
### Extension Methods for Partial Functions
239
240
Scala-friendly extension methods that accept partial functions for pattern matching.
241
242
```scala { .api }
243
// Import extensions
244
import org.apache.flink.api.scala.extensions._
245
246
class OnDataSet[T] {
247
def mapWith[R: TypeInformation: ClassTag](fun: T => R): DataSet[R]
248
def mapPartitionWith[R: TypeInformation: ClassTag](fun: Stream[T] => R): DataSet[R]
249
def flatMapWith[R: TypeInformation: ClassTag](fun: T => TraversableOnce[R]): DataSet[R]
250
def filterWith(fun: T => Boolean): DataSet[T]
251
def reduceWith(fun: (T, T) => T): DataSet[T]
252
def reduceGroupWith[R: TypeInformation: ClassTag](fun: Stream[T] => R): DataSet[R]
253
def groupingBy[K: TypeInformation](fun: T => K): GroupedDataSet[T]
254
}
255
```
256
257
### Utility Functions
258
259
Additional utilities for DataSet operations including sampling, partitioning, and indexing.
260
261
```scala { .api }
262
// Import utilities
263
import org.apache.flink.api.scala.utils._
264
265
class DataSet[T] {
266
// Available via implicit conversions from utils package
267
def countElementsPerPartition(): DataSet[(Int, Long)]
268
def zipWithIndex(): DataSet[(Long, T)]
269
def zipWithUniqueId(): DataSet[(Long, T)]
270
def sample(withReplacement: Boolean, fraction: Double, seed: Long = Utils.RNG.nextLong()): DataSet[T]
271
def sampleWithSize(withReplacement: Boolean, numSamples: Int, seed: Long = Utils.RNG.nextLong()): DataSet[T]
272
def checksumHashCode(): ChecksumHashCode
273
}
274
```
275
276
[Utility Functions](./utility-functions.md)
277
278
## Common Types
279
280
```scala { .api }
281
// Execution configuration
282
class ExecutionConfig {
283
def setParallelism(parallelism: Int): ExecutionConfig
284
def getParallelism: Int
285
def enableClosureCleaner(): ExecutionConfig
286
def disableClosureCleaner(): ExecutionConfig
287
}
288
289
// Job execution result
290
class JobExecutionResult {
291
def getJobExecutionTime: Long
292
def getAccumulatorResult[T](accumulatorName: String): T
293
}
294
295
// Aggregation types
296
object Aggregations extends Enumeration {
297
val SUM, MAX, MIN = Value
298
}
299
300
// File system write modes
301
object FileSystem {
302
object WriteMode extends Enumeration {
303
val NO_OVERWRITE, OVERWRITE = Value
304
}
305
}
306
307
// Ordering for sorting
308
object Order extends Enumeration {
309
val ASCENDING, DESCENDING = Value
310
}
311
```
312
313
## Error Handling
314
315
The Flink Scala API can throw these exceptions:
316
317
- `IllegalArgumentException` - Invalid parameters or field names
318
- `UnsupportedOperationException` - Unsupported operations on certain data types
319
- `RuntimeException` - Runtime execution errors
320
- `IOException` - File I/O related errors
321
- `JobExecutionException` - Job execution failures