0
# Apache Flink HBase Connector
1
2
The Apache Flink HBase connector provides comprehensive integration between Apache Flink's stream processing capabilities and Apache HBase's NoSQL database. It supports both DataStream API and Table API operations for reading from and writing to HBase tables, with features like lookup joins, buffered writes, and flexible serialization.
3
4
## Package Information
5
6
- **Package Name**: org.apache.flink:flink-hbase_2.11
7
- **Package Type**: maven
8
- **Language**: Java
9
- **Version**: 1.10.3
10
- **Installation**: Add to your Maven pom.xml:
11
12
```xml
13
<dependency>
14
<groupId>org.apache.flink</groupId>
15
<artifactId>flink-hbase_${scala.binary.version}</artifactId>
16
<version>1.10.3</version>
17
</dependency>
18
```
19
20
## Core Imports
21
22
```java
23
// DataStream API - Input Format
24
import org.apache.flink.addons.hbase.HBaseRowInputFormat;
25
import org.apache.flink.addons.hbase.HBaseTableSchema;
26
27
// DataStream API - Sink Function
28
import org.apache.flink.addons.hbase.HBaseUpsertSinkFunction;
29
30
// Table API - Source and Sink
31
import org.apache.flink.addons.hbase.HBaseTableSource;
32
import org.apache.flink.addons.hbase.HBaseUpsertTableSink;
33
import org.apache.flink.addons.hbase.HBaseTableFactory;
34
35
// Configuration
36
import org.apache.flink.addons.hbase.HBaseOptions;
37
import org.apache.flink.addons.hbase.HBaseWriteOptions;
38
39
// Table API Descriptors
40
import org.apache.flink.table.descriptors.HBase;
41
```
42
43
## Basic Usage
44
45
### Reading from HBase (DataStream API)
46
47
```java
48
import org.apache.flink.addons.hbase.HBaseRowInputFormat;
49
import org.apache.flink.addons.hbase.HBaseTableSchema;
50
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
51
import org.apache.hadoop.conf.Configuration;
52
53
// Configure HBase connection
54
Configuration conf = new Configuration();
55
conf.set("hbase.zookeeper.quorum", "localhost:2181");
56
57
// Define table schema
58
HBaseTableSchema schema = new HBaseTableSchema();
59
schema.setRowKey("rowkey", String.class);
60
schema.addColumn("cf1", "col1", String.class);
61
schema.addColumn("cf1", "col2", Integer.class);
62
63
// Create input format
64
HBaseRowInputFormat inputFormat = new HBaseRowInputFormat(conf, "my_table", schema);
65
66
// Read as DataStream
67
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
68
DataStream<Row> hbaseData = env.createInput(inputFormat);
69
```
70
71
### Writing to HBase (DataStream API)
72
73
```java
74
import org.apache.flink.addons.hbase.HBaseUpsertSinkFunction;
75
import org.apache.flink.types.Row;
76
77
// Create sink function with buffering
78
HBaseUpsertSinkFunction sinkFunction = new HBaseUpsertSinkFunction(
79
"my_table", // table name
80
schema, // table schema
81
conf, // HBase configuration
82
2 * 1024 * 1024, // buffer flush max size (2MB)
83
1000, // buffer flush max mutations
84
5000 // buffer flush interval (5 seconds)
85
);
86
87
// Apply to DataStream
88
DataStream<Tuple2<Boolean, Row>> upsertStream = // your stream of upserts
89
upsertStream.addSink(sinkFunction);
90
```
91
92
### Table API Usage
93
94
```java
95
import org.apache.flink.table.api.EnvironmentSettings;
96
import org.apache.flink.table.api.TableEnvironment;
97
import org.apache.flink.table.descriptors.HBase;
98
import org.apache.flink.table.descriptors.Schema;
99
100
TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());
101
102
// Register HBase table
103
tableEnv.connect(
104
new HBase()
105
.version("1.4.3")
106
.tableName("my_table")
107
.zookeeperQuorum("localhost:2181")
108
)
109
.withSchema(
110
new Schema()
111
.field("rowkey", DataTypes.STRING())
112
.field("cf1_col1", DataTypes.STRING())
113
.field("cf1_col2", DataTypes.INT())
114
)
115
.createTemporaryTable("hbase_table");
116
117
// Query the table
118
Table result = tableEnv.sqlQuery("SELECT * FROM hbase_table WHERE cf1_col2 > 100");
119
```
120
121
## Architecture
122
123
The HBase connector is organized into several key components:
124
125
- **Input Formats**: For batch and streaming reads (`HBaseRowInputFormat`, `AbstractTableInputFormat`)
126
- **Sink Functions**: For streaming writes with buffering (`HBaseUpsertSinkFunction`)
127
- **Table API Integration**: Source/sink factories and descriptors (`HBaseTableFactory`, `HBaseTableSource`)
128
- **Schema Definition**: Type-safe column family and qualifier mapping (`HBaseTableSchema`)
129
- **Configuration**: Connection and write performance options (`HBaseOptions`, `HBaseWriteOptions`)
130
- **Utilities**: Type conversion and HBase operation helpers
131
132
## Capabilities
133
134
### DataStream API Input Formats
135
136
Read data from HBase tables using InputFormat classes with full control over scanning and result mapping.
137
138
```java { .api }
139
class HBaseRowInputFormat extends AbstractTableInputFormat<Row> {
140
public HBaseRowInputFormat(Configuration conf, String tableName, HBaseTableSchema schema);
141
public void configure(Configuration parameters);
142
public String getTableName();
143
public TypeInformation<Row> getProducedType();
144
}
145
```
146
147
[DataStream Input Formats](./input-formats.md)
148
149
### DataStream API Sink Functions
150
151
Write data to HBase tables with configurable buffering and automatic batching for optimal performance.
152
153
```java { .api }
154
class HBaseUpsertSinkFunction extends RichSinkFunction<Tuple2<Boolean, Row>>
155
implements CheckpointedFunction, BufferedMutator.ExceptionListener {
156
public HBaseUpsertSinkFunction(String hTableName, HBaseTableSchema schema,
157
Configuration conf, long bufferFlushMaxSizeInBytes,
158
long bufferFlushMaxMutations, long bufferFlushIntervalMillis);
159
public void open(Configuration parameters);
160
public void invoke(Tuple2<Boolean, Row> value, Context context);
161
public void close();
162
}
163
```
164
165
[DataStream Sink Functions](./sink-functions.md)
166
167
### Table API Sources and Sinks
168
169
Integrate HBase with Flink's Table API for SQL-based data processing and lookup joins.
170
171
```java { .api }
172
class HBaseTableSource implements StreamTableSource<Row>, LookupableTableSource<Row> {
173
public HBaseTableSource(Configuration conf, String tableName);
174
public void addColumn(String family, String qualifier, Class<?> clazz);
175
public void setRowKey(String rowKeyName, Class<?> clazz);
176
public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv);
177
public TableFunction<Row> getLookupFunction(String[] lookupKeys);
178
}
179
```
180
181
```java { .api }
182
class HBaseUpsertTableSink implements UpsertStreamTableSink<Row> {
183
public HBaseUpsertTableSink(HBaseTableSchema hbaseTableSchema,
184
HBaseOptions hbaseOptions, HBaseWriteOptions writeOptions);
185
public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream);
186
}
187
```
188
189
[Table API Integration](./table-api.md)
190
191
### Schema Configuration
192
193
Define type-safe mappings between Flink data types and HBase column families and qualifiers.
194
195
```java { .api }
196
class HBaseTableSchema {
197
public void addColumn(String family, String qualifier, Class<?> clazz);
198
public void setRowKey(String rowKeyName, Class<?> clazz);
199
public void setCharset(String charset);
200
public String[] getFamilyNames();
201
public TypeInformation<?>[] getQualifierTypes(String family);
202
}
203
```
204
205
[Schema and Configuration](./schema-config.md)
206
207
### Lookup Functions
208
209
Enable temporal table joins by looking up dimension data from HBase in real-time.
210
211
```java { .api }
212
class HBaseLookupFunction extends TableFunction<Row> {
213
public HBaseLookupFunction(Configuration configuration, String hTableName,
214
HBaseTableSchema hbaseTableSchema);
215
public void eval(Object rowKey);
216
public TypeInformation<Row> getResultType();
217
}
218
```
219
220
[Lookup Functions](./lookup-functions.md)
221
222
### Utility Classes
223
224
Helper classes for type conversion, configuration serialization, and HBase operation creation.
225
226
```java { .api }
227
class HBaseTypeUtils {
228
public static Object deserializeToObject(byte[] value, int typeIdx, Charset stringCharset);
229
public static byte[] serializeFromObject(Object value, int typeIdx, Charset stringCharset);
230
public static int getTypeIndex(TypeInformation typeInfo);
231
public static boolean isSupportedType(Class<?> clazz);
232
}
233
```
234
235
[Utilities](./utilities.md)
236
237
## Configuration
238
239
### Connection Configuration
240
241
```java { .api }
242
class HBaseOptions {
243
public static Builder builder();
244
245
static class Builder {
246
public Builder setTableName(String tableName); // Required
247
public Builder setZkQuorum(String zkQuorum); // Required
248
public Builder setZkNodeParent(String zkNodeParent); // Optional
249
public HBaseOptions build();
250
}
251
}
252
```
253
254
### Write Performance Configuration
255
256
```java { .api }
257
class HBaseWriteOptions {
258
public static Builder builder();
259
260
static class Builder {
261
public Builder setBufferFlushMaxSizeInBytes(long bufferFlushMaxSizeInBytes);
262
public Builder setBufferFlushMaxRows(long bufferFlushMaxRows);
263
public Builder setBufferFlushIntervalMillis(long bufferFlushIntervalMillis);
264
public HBaseWriteOptions build();
265
}
266
}
267
```
268
269
## Supported Data Types
270
271
The connector supports automatic serialization/deserialization for:
272
273
- **Primitive types**: `byte[]`, `String`, `Byte`, `Short`, `Integer`, `Long`, `Float`, `Double`, `Boolean`
274
- **Temporal types**: `java.sql.Timestamp`, `java.sql.Date`, `java.sql.Time`
275
- **Numeric types**: `java.math.BigDecimal`, `java.math.BigInteger`
276
277
## Error Handling
278
279
Common exceptions and their meanings:
280
281
- `RuntimeException`: HBase connection or configuration errors
282
- `IllegalArgumentException`: Invalid parameters or unsupported data types
283
- `IOException`: HBase I/O operation failures
284
- `ValidationException`: Table configuration validation errors
285
- `TableNotFoundException`: Specified HBase table doesn't exist
286
- `RetriesExhaustedWithDetailsException`: BufferedMutator operation failures
287
288
## Requirements
289
290
- **HBase Version**: 1.4.3
291
- **Flink Version**: 1.10.3
292
- **Java Version**: 8+
293
- **Hadoop Configuration**: Properly configured `hbase-site.xml` or programmatic configuration