0
# Flink HCatalog
1
2
Flink HCatalog is a connector library for Apache Flink that enables reading data from Apache Hive HCatalog tables. It provides both Java and Scala APIs with support for schema projection, partition filtering, and automatic type mapping between HCatalog schemas and Flink's type system.
3
4
## Package Information
5
6
- **Package Name**: flink-hcatalog_2.11
7
- **Package Type**: maven
8
- **Language**: Java/Scala
9
- **Installation**: Add to your Maven pom.xml:
10
11
```xml
12
<dependency>
13
<groupId>org.apache.flink</groupId>
14
<artifactId>flink-hcatalog_2.11</artifactId>
15
<version>1.14.6</version>
16
</dependency>
17
```
18
19
## Core Imports
20
21
**Java:**
22
```java
23
import org.apache.flink.hcatalog.java.HCatInputFormat;
24
import org.apache.flink.hcatalog.HCatInputFormatBase;
25
import org.apache.hadoop.conf.Configuration;
26
import org.apache.hive.hcatalog.data.HCatRecord;
27
import org.apache.hive.hcatalog.data.schema.HCatSchema;
28
import org.apache.hive.hcatalog.data.schema.HCatFieldSchema;
29
import org.apache.hive.hcatalog.common.HCatException;
30
```
31
32
**Scala:**
33
```scala
34
import org.apache.flink.hcatalog.scala.HCatInputFormat
35
import org.apache.hadoop.conf.Configuration
36
import org.apache.hive.hcatalog.data.HCatRecord
37
```
38
39
## Basic Usage
40
41
**Java Example:**
42
```java
43
import org.apache.flink.api.java.ExecutionEnvironment;
44
import org.apache.flink.api.java.DataSet;
45
import org.apache.flink.hcatalog.java.HCatInputFormat;
46
import org.apache.hive.hcatalog.data.HCatRecord;
47
48
// Create execution environment
49
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
50
51
// Create HCatalog input format
52
HCatInputFormat<HCatRecord> hCatFormat = new HCatInputFormat<>("mydb", "mytable");
53
54
// Optional: Project specific fields
55
hCatFormat.getFields("name", "age", "city");
56
57
// Optional: Apply partition filter
58
hCatFormat.withFilter("year=2023 AND month='01'");
59
60
// Create DataSet and process
61
DataSet<HCatRecord> input = env.createInput(hCatFormat);
62
```
63
64
**Scala Example:**
65
```scala
66
import org.apache.flink.api.scala._
67
import org.apache.flink.hcatalog.scala.HCatInputFormat
68
import org.apache.hive.hcatalog.data.HCatRecord
69
70
// Create execution environment
71
val env = ExecutionEnvironment.getExecutionEnvironment
72
73
// Create HCatalog input format with field projection
74
val hCatFormat = new HCatInputFormat[HCatRecord]("mydb", "mytable")
75
.getFields("name", "age", "city")
76
.withFilter("year=2023")
77
78
// Create DataSet
79
val input = env.createInput(hCatFormat)
80
```
81
82
## Architecture
83
84
The Flink HCatalog connector is built around several key components:
85
86
- **Abstract Base Class**: `HCatInputFormatBase` provides common functionality for both Java and Scala implementations
87
- **Language-Specific Implementations**: Separate classes for Java (`HCatInputFormat`) and Scala (`HCatInputFormat`) that handle tuple conversion
88
- **Type System Integration**: Automatic mapping between HCatalog data types and Flink's type system
89
- **Hadoop Integration**: Built on top of Hadoop's MapReduce input format framework for distributed data reading
90
- **Schema Management**: Support for dynamic schema projection and filtering at the partition level
91
92
## Capabilities
93
94
### Java HCatalog Input Format
95
96
Java implementation supporting Flink Tuples up to 25 fields and HCatRecord output.
97
98
```java { .api }
99
/**
100
* Java HCatalog input format for reading from Hive tables
101
* Supports conversion to Flink Tuples (up to 25 fields) or HCatRecord objects
102
*/
103
public class HCatInputFormat<T> extends HCatInputFormatBase<T> {
104
public HCatInputFormat();
105
public HCatInputFormat(String database, String table) throws Exception;
106
public HCatInputFormat(String database, String table, Configuration config) throws Exception;
107
}
108
```
109
110
**Usage Example:**
111
```java
112
// Reading as HCatRecord (default)
113
HCatInputFormat<HCatRecord> recordFormat = new HCatInputFormat<>("sales", "transactions");
114
115
// Reading as Flink Tuple
116
HCatInputFormat<Tuple3<String, Integer, Double>> tupleFormat =
117
new HCatInputFormat<Tuple3<String, Integer, Double>>("sales", "transactions")
118
.getFields("product_name", "quantity", "price")
119
.asFlinkTuples();
120
```
121
122
### Scala HCatalog Input Format
123
124
Scala implementation supporting Scala tuples up to 22 fields and HCatRecord output.
125
126
```scala { .api }
127
/**
128
* Scala HCatalog input format for reading from Hive tables
129
* Supports conversion to Scala tuples (up to 22 fields) or HCatRecord objects
130
*/
131
class HCatInputFormat[T](database: String, table: String, config: Configuration)
132
extends HCatInputFormatBase[T](database, table, config) {
133
def this(database: String, table: String) {
134
this(database, table, new Configuration)
135
}
136
}
137
```
138
139
**Usage Example:**
140
```scala
141
// Reading as HCatRecord (default)
142
val recordFormat = new HCatInputFormat[HCatRecord]("sales", "transactions")
143
144
// Reading as Scala Tuple
145
val tupleFormat = new HCatInputFormat[(String, Int, Double)]("sales", "transactions")
146
.getFields("product_name", "quantity", "price")
147
.asFlinkTuples()
148
```
149
150
### Field Projection
151
152
Select and reorder specific fields from HCatalog tables to reduce data transfer and processing overhead.
153
154
```java { .api }
155
/**
156
* Specifies the fields which are returned by the InputFormat and their order
157
* @param fields The fields and their order which are returned by the InputFormat
158
* @return This InputFormat with specified return fields
159
* @throws IOException if field projection fails
160
*/
161
public HCatInputFormatBase<T> getFields(String... fields) throws IOException;
162
```
163
164
**Usage Examples:**
165
```java
166
// Java: Select specific fields
167
hCatFormat.getFields("customer_id", "order_date", "total_amount");
168
169
// Java: Reorder fields
170
hCatFormat.getFields("total_amount", "customer_id", "order_date");
171
```
172
173
```scala
174
// Scala: Select specific fields
175
hCatFormat.getFields("customer_id", "order_date", "total_amount")
176
177
// Scala: Reorder fields
178
hCatFormat.getFields("total_amount", "customer_id", "order_date")
179
```
180
181
### Partition Filtering
182
183
Apply SQL-like filter conditions on partition columns to significantly reduce the amount of data to be read.
184
185
```java { .api }
186
/**
187
* Specifies a SQL-like filter condition on the table's partition columns
188
* Filter conditions on non-partition columns are invalid
189
* @param filter A SQL-like filter condition on the table's partition columns
190
* @return This InputFormat with specified partition filter
191
* @throws IOException if filter application fails
192
*/
193
public HCatInputFormatBase<T> withFilter(String filter) throws IOException;
194
```
195
196
**Usage Examples:**
197
```java
198
// Java: Single partition filter
199
hCatFormat.withFilter("year=2023");
200
201
// Java: Multiple partition conditions
202
hCatFormat.withFilter("year=2023 AND month='12' AND day>=15");
203
204
// Java: Range and comparison operators
205
hCatFormat.withFilter("year>=2020 AND region IN ('US', 'EU')");
206
```
207
208
```scala
209
// Scala: Single partition filter
210
hCatFormat.withFilter("year=2023")
211
212
// Scala: Multiple partition conditions
213
hCatFormat.withFilter("year=2023 AND month='12' AND day>=15")
214
```
215
216
### Tuple Conversion
217
218
Convert HCatRecord output to native Flink or Scala tuples for improved type safety and performance.
219
220
```java { .api }
221
/**
222
* Specifies that the InputFormat returns Flink tuples instead of HCatRecord
223
* Note: Flink tuples might only support a limited number of fields (depending on the API)
224
* @return This InputFormat configured to return tuples
225
* @throws HCatException if tuple conversion setup fails
226
*/
227
public HCatInputFormatBase<T> asFlinkTuples() throws HCatException;
228
```
229
230
**Usage Examples:**
231
```java
232
// Java: Convert to Flink Tuple (up to 25 fields)
233
HCatInputFormat<Tuple2<String, Integer>> tupleFormat =
234
new HCatInputFormat<Tuple2<String, Integer>>("mydb", "mytable")
235
.getFields("name", "age")
236
.asFlinkTuples();
237
```
238
239
```scala
240
// Scala: Convert to Scala tuple (up to 22 fields)
241
val tupleFormat = new HCatInputFormat[(String, Int)]("mydb", "mytable")
242
.getFields("name", "age")
243
.asFlinkTuples()
244
```
245
246
### Configuration Access
247
248
Access and modify the underlying Hadoop configuration for advanced customization.
249
250
```java { .api }
251
/**
252
* Returns the Hadoop Configuration of the HCatInputFormat
253
* @return The Configuration of the HCatInputFormat
254
*/
255
public Configuration getConfiguration();
256
```
257
258
**Usage Example:**
259
```java
260
// Access configuration for customization
261
Configuration config = hCatFormat.getConfiguration();
262
config.set("hive.metastore.uris", "thrift://metastore:9083");
263
config.setInt("mapreduce.input.fileinputformat.split.minsize", 1024000);
264
```
265
266
### Schema Information
267
268
Retrieve schema information for the HCatalog table being read.
269
270
```java { .api }
271
/**
272
* Returns the HCatSchema of the HCatRecord returned by this InputFormat
273
* @return The HCatSchema of the HCatRecords returned by this InputFormat
274
*/
275
public HCatSchema getOutputSchema();
276
```
277
278
**Usage Example:**
279
```java
280
// Inspect table schema
281
HCatSchema schema = hCatFormat.getOutputSchema();
282
List<HCatFieldSchema> fields = schema.getFields();
283
for (HCatFieldSchema field : fields) {
284
System.out.println("Field: " + field.getName() + ", Type: " + field.getType());
285
}
286
```
287
288
## Types
289
290
```java { .api }
291
/**
292
* Abstract base class for HCatalog input formats
293
* Provides common functionality for reading from HCatalog tables
294
*/
295
public abstract class HCatInputFormatBase<T> extends RichInputFormat<T, HadoopInputSplit>
296
implements ResultTypeQueryable<T> {
297
298
protected HCatInputFormatBase();
299
protected HCatInputFormatBase(String database, String table) throws IOException;
300
protected HCatInputFormatBase(String database, String table, Configuration config) throws IOException;
301
302
// Abstract methods to be implemented by language-specific subclasses
303
protected abstract int getMaxFlinkTupleSize();
304
protected abstract T buildFlinkTuple(T t, HCatRecord record) throws HCatException;
305
}
306
307
/**
308
* Hadoop input split wrapper for Flink integration
309
*/
310
class HadoopInputSplit implements InputSplit {
311
// Implementation details handled internally
312
}
313
314
/**
315
* HCatalog record representing a row of data from Hive table
316
* Contains field values accessible by name or position
317
*/
318
interface HCatRecord {
319
Object get(String fieldName, HCatSchema schema);
320
Object get(int fieldPos);
321
List<Object> getAll();
322
// Additional methods for data access
323
}
324
325
/**
326
* Schema definition for HCatalog table structure
327
* Contains field definitions and metadata
328
*/
329
class HCatSchema {
330
List<HCatFieldSchema> getFields();
331
HCatFieldSchema get(String fieldName);
332
HCatFieldSchema get(int position);
333
int getPosition(String fieldName);
334
List<String> getFieldNames();
335
// Additional schema methods
336
}
337
338
/**
339
* Individual field schema definition
340
*/
341
class HCatFieldSchema {
342
String getName();
343
Type getType();
344
String getComment();
345
// Additional field metadata methods
346
347
enum Type {
348
INT, TINYINT, SMALLINT, BIGINT, BOOLEAN, FLOAT, DOUBLE,
349
STRING, BINARY, ARRAY, MAP, STRUCT
350
}
351
}
352
```
353
354
## Supported Data Types
355
356
The connector provides automatic type mapping between HCatalog and Flink types:
357
358
| HCatalog Type | Flink Type | Java Type | Scala Type |
359
|---------------|------------|-----------|------------|
360
| INT | BasicTypeInfo.INT_TYPE_INFO | Integer | Int |
361
| TINYINT | BasicTypeInfo.BYTE_TYPE_INFO | Byte | Byte |
362
| SMALLINT | BasicTypeInfo.SHORT_TYPE_INFO | Short | Short |
363
| BIGINT | BasicTypeInfo.LONG_TYPE_INFO | Long | Long |
364
| BOOLEAN | BasicTypeInfo.BOOLEAN_TYPE_INFO | Boolean | Boolean |
365
| FLOAT | BasicTypeInfo.FLOAT_TYPE_INFO | Float | Float |
366
| DOUBLE | BasicTypeInfo.DOUBLE_TYPE_INFO | Double | Double |
367
| STRING | BasicTypeInfo.STRING_TYPE_INFO | String | String |
368
| BINARY | PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO | byte[] | Array[Byte] |
369
| ARRAY | GenericTypeInfo(List.class) | List<Object> | List[Object] |
370
| MAP | GenericTypeInfo(Map.class) | Map<Object, Object> | Map[Object, Object] |
371
| STRUCT | GenericTypeInfo(List.class) | List<Object> | List[Object] |
372
373
## Error Handling
374
375
The connector handles various error conditions:
376
377
**Configuration Errors:**
378
- `IOException`: Thrown for invalid database/table names, connection issues, or configuration problems
379
- `HCatException`: Thrown for HCatalog-specific errors during schema access or filtering
380
381
**Type Conversion Errors:**
382
- `IllegalArgumentException`: Thrown when requesting more fields than supported by tuple type (25 for Java, 22 for Scala)
383
- `RuntimeException`: Thrown for unsupported partition key types (BINARY, ARRAY, MAP, STRUCT as partition keys)
384
385
**Usage Errors:**
386
- Invalid filter conditions on non-partition columns will cause runtime failures
387
- Requesting non-existent fields will result in schema validation errors
388
389
**Example Error Handling:**
390
```java
391
try {
392
HCatInputFormat<HCatRecord> format = new HCatInputFormat<>("mydb", "mytable");
393
format.withFilter("year=2023");
394
format.getFields("name", "age", "salary");
395
} catch (IOException e) {
396
// Handle configuration or connection errors
397
logger.error("Failed to configure HCatalog input: " + e.getMessage());
398
} catch (HCatException e) {
399
// Handle HCatalog-specific errors
400
logger.error("HCatalog error: " + e.getMessage());
401
}
402
```