0
# Apache Flink SQL Connector for Hive 2.3.9
1

2
Apache Flink SQL connector for Apache Hive 2.3.9 enables seamless integration between Flink's streaming and batch processing capabilities and Apache Hive data warehouses. This connector provides comprehensive access to Hive tables, metastore operations, and built-in functions, allowing developers to leverage existing Hive infrastructure within Flink applications.
3

4
## Package Information
5

6
- **Package Name**: flink-sql-connector-hive-2.3.9_2.12
7
- **Package Type**: maven
8
- **Group ID**: org.apache.flink
9
- **Language**: Java
10
- **Installation**: Add dependency to pom.xml
11
- **Hive Version**: 2.3.9
12
- **Scala Version**: 2.12
13

14
```xml
15
<dependency>
16
    <groupId>org.apache.flink</groupId>
17
    <artifactId>flink-sql-connector-hive-2.3.9_2.12</artifactId>
18
    <version>1.19.3</version>
19
</dependency>
20
```
21

22
## Core Imports
23

24
```java
25
// Main factories
26
import org.apache.flink.connectors.hive.HiveDynamicTableFactory;
27
import org.apache.flink.table.catalog.hive.HiveCatalog;
28
import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactory;
29
import org.apache.flink.table.module.hive.HiveModule;
30

31
// Source and sink classes
32
import org.apache.flink.connectors.hive.HiveSource;
33
import org.apache.flink.connectors.hive.HiveSourceBuilder;
34
import org.apache.flink.connectors.hive.HiveTableSource;
35
import org.apache.flink.connectors.hive.HiveTableSink;
36

37
// Configuration
38
import org.apache.flink.connectors.hive.HiveOptions;
39
import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
40

41
// Data structures
42
import org.apache.flink.connectors.hive.HiveTablePartition;
43
import org.apache.flink.connectors.hive.FlinkHiveException;
44
```
45

46
## Basic Usage
47

48
### Table API Integration
49

50
```java
51
import org.apache.flink.table.api.EnvironmentSettings;
52
import org.apache.flink.table.api.TableEnvironment;
53
import org.apache.flink.table.catalog.hive.HiveCatalog;
54

55
// Create table environment
56
EnvironmentSettings settings = EnvironmentSettings
57
    .newInstance()
58
    .inBatchMode()
59
    .build();
60
TableEnvironment tableEnv = TableEnvironment.create(settings);
61

62
// Create and register Hive catalog
63
String catalogName = "myhive";
64
String defaultDatabase = "mydatabase";
65
String hiveConfDir = "/opt/hive-conf";
66
String version = "2.3.9";
67

68
HiveCatalog hive = new HiveCatalog(catalogName, defaultDatabase, hiveConfDir, null, version);
69
tableEnv.registerCatalog("myhive", hive);
70
tableEnv.useCatalog("myhive");
71

72
// Query Hive tables
73
Table result = tableEnv.sqlQuery("SELECT * FROM mytable WHERE active = true");
74
result.execute().print();
75
```
76

77
### DataStream API Integration
78

79
```java
80
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
81
import org.apache.flink.connectors.hive.HiveSource;
82
import org.apache.flink.connectors.hive.HiveSourceBuilder;
83
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
84

85
// Create execution environment
86
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
87

88
// Build Hive source
89
JobConf jobConf = new JobConf();
90
jobConf.set("hive.metastore.uris", "thrift://localhost:9083");
91

92
HiveSource<RowData> source = new HiveSourceBuilder(
93
    jobConf,
94
    new Configuration(),
95
    "2.3.9", // hiveVersion
96
    "mydb", 
97
    "mytable",
98
    Collections.emptyMap()
99
).buildWithDefaultBulkFormat();
100

101
// Create data stream
102
DataStream<RowData> stream = env
103
    .fromSource(source, WatermarkStrategy.noWatermarks(), "hive-source");
104

105
stream.print();
106
env.execute("Hive Stream Processing");
107
```
108

109
## Architecture
110

111
The Flink Hive connector is built around several key architectural components:
112

113
- **Factory System**: Plugin-based factories (`HiveDynamicTableFactory`, `HiveCatalogFactory`) for dynamic registration and configuration
114
- **Source/Sink Framework**: Unified data access layer with streaming and batch support through `HiveSource` and `HiveTableSink`
115
- **Catalog Integration**: Full Hive metastore integration via `HiveCatalog` for schema discovery and table management
116
- **Function Module**: Native Hive function support through `HiveModule` for UDF compatibility
117
- **Partition Management**: Intelligent partition handling with pruning and dynamic discovery capabilities
118
- **Type System**: Seamless type mapping between Hive and Flink data types with full serialization support
119

120
## Capabilities
121

122
### Table Source and Sink Operations
123

124
Core functionality for reading from and writing to Hive tables using Flink's Table API and DataStream API. Supports both batch and streaming modes with comprehensive partition handling.
125

126
```java { .api }
127
// Main source builder
128
public class HiveSourceBuilder {
129
    public HiveSourceBuilder(JobConf jobConf, ReadableConfig flinkConf, 
130
                           String hiveVersion, String dbName, 
131
                           String tableName, Map<String, String> tableOptions);
132
    public HiveSource<RowData> buildWithDefaultBulkFormat();
133
    public <T> HiveSource<T> buildWithBulkFormat(BulkFormat<T, HiveSourceSplit> bulkFormat);
134
    public HiveSourceBuilder setPartitions(List<HiveTablePartition> partitions);
135
    public HiveSourceBuilder setLimit(Long limit);
136
    public HiveSourceBuilder setProjectedFields(int[] projectedFields);
137
}
138

139
// Dynamic table source
140
public class HiveTableSource implements ScanTableSource, SupportsPartitionPushDown, 
141
                                       SupportsProjectionPushDown, SupportsLimitPushDown {
142
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext);
143
    public Result applyPartitions(List<Map<String, String>> remainingPartitions);
144
    public Result applyProjection(int[][] projectedFields, DataType producedDataType);
145
    public Result applyLimit(long limit);
146
}
147

148
// Dynamic table sink  
149
public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, SupportsOverwrite {
150
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context);
151
    public Result applyStaticPartition(Map<String, String> partition);
152
    public Result applyOverwrite(boolean overwrite);
153
}
154
```
155

156
[Table Source and Sink Operations](./table-source-sink.md)
157

158
### Catalog Integration
159

160
Complete Hive metastore integration providing schema discovery, table management, and metadata operations. Enables transparent access to existing Hive data warehouses.
161

162
```java { .api }
163
public class HiveCatalog extends AbstractCatalog {
164
    public HiveCatalog(String catalogName, String defaultDatabase, 
165
                      String hiveConfDir, String hadoopConfDir, String hiveVersion);
166
    public List<String> listDatabases() throws DatabaseNotExistException, CatalogException;
167
    public List<String> listTables(String databaseName) throws DatabaseNotExistException, CatalogException;
168
    public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException;
169
    public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists);
170
    public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists);
171
    public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath);
172
}
173

174
public class HiveCatalogFactory implements CatalogFactory {
175
    public Catalog createCatalog(Context context);
176
    public String factoryIdentifier();
177
}
178
```
179

180
[Catalog Integration](./catalog-integration.md)
181

182
### Function Module Integration
183

184
Hive built-in function support enabling use of Hive UDFs within Flink SQL queries. Provides seamless function compatibility and registration.
185

186
```java { .api }
187
public class HiveModule implements Module {
188
    public HiveModule();
189
    public HiveModule(String hiveVersion);
190
    public Set<String> listFunctions();
191
    public Optional<FunctionDefinition> getFunctionDefinition(String name);
192
}
193
194
```
195

196
[Function Module Integration](./function-module.md)
197

198
### Configuration and Options
199

200
Comprehensive configuration system for customizing connector behavior, performance tuning, and environment-specific settings.
201

202
```java { .api }
203
public class HiveOptions {
204
    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER;
205
    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_FALLBACK_MAPRED_WRITER;
206
    public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE;
207
    public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL;
208
    public static final ConfigOption<Duration> LOOKUP_JOIN_CACHE_TTL;
209
    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM;
210
    public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_KIND;
211
}
212

213
public class HiveCatalogFactoryOptions {
214
    public static final ConfigOption<String> DEFAULT_DATABASE;
215
    public static final ConfigOption<String> HIVE_CONF_DIR;
216
    public static final ConfigOption<String> HIVE_VERSION;
217
    public static final ConfigOption<String> HADOOP_CONF_DIR;
218
}
219
```
220

221
[Configuration and Options](./configuration.md)
222

223
### Partition Management and Utilities
224

225
Advanced partition handling utilities for efficient data access, partition pruning, and metadata management in partitioned Hive tables.
226

227
```java { .api }
228
public class HiveTablePartition implements Serializable {
229
    public static HiveTablePartition ofTable(StorageDescriptor storageDescriptor, 
230
                                            Map<String, String> tableParameters);
231
    public static HiveTablePartition ofPartition(StorageDescriptor storageDescriptor,
232
                                                Map<String, String> partitionSpec,
233
                                                Map<String, String> tableParameters);
234
    public StorageDescriptor getStorageDescriptor();
235
    public Map<String, String> getPartitionSpec();
236
}
237

238
public class HivePartitionUtils {
239
    public static List<HiveTablePartition> getAllPartitions(JobConf jobConf, String catalogName,
240
                                                           ObjectPath tablePath, List<String> partitionColNames);
241
    public static byte[] serializeHiveTablePartition(List<HiveTablePartition> partitions);
242
    public static List<HiveTablePartition> deserializeHiveTablePartition(byte[] bytes, ClassLoader classLoader);
243
}
244
```
245

246
[Partition Management and Utilities](./partition-management.md)
247

248
## Types
249

250
```java { .api }
251
// Core exception type
252
public class FlinkHiveException extends RuntimeException {
253
    public FlinkHiveException(String message);
254
    public FlinkHiveException(Throwable cause);
255
    public FlinkHiveException(String message, Throwable cause);
256
}
257

258
// Configuration wrapper
259
public class JobConfWrapper implements Serializable {
260
    public JobConfWrapper(JobConf jobConf);
261
    public JobConf conf();
262
}
263
264
```