Skip to main content
Version: Next

Hbase

Hbase sink connector

Description​

Writes SeaTunnel rows to Apache HBase. The sink can create or reuse target tables, write one or multiple upstream tables, map fields to one column family or multiple column families, and control how null values, row keys, WAL, timestamps, and existing data are handled.

Key features​

Options​

NameTypeRequiredDefaultDescription
zookeeper_quorumstringyes-HBase ZooKeeper quorum, for example hadoop001:2181,hadoop002:2181.
tablestringyes-Target HBase table. Use placeholders such as ${table_name} for multi-table writes.
rowkey_columnlistyes-Upstream field names used to build the HBase row key.
family_nameconfigyes-Mapping from upstream fields to HBase column families, or all_columns for one family.
rowkey_delimiterstringno""Delimiter used when multiple fields form the row key.
version_columnstringno-Upstream field used as the HBase cell timestamp.
null_modestringnoskipHow null values are written. Supported values are skip and empty.
wal_writebooleannofalseWhether writes should be recorded in the HBase WAL.
write_buffer_sizeintno8 1024 1024HBase client write buffer size in bytes.
encodingstringnoutf8Encoding for STRING/DECIMAL/DATE/TIME/TIMESTAMP/ARRAY values. Supported values are utf8 and gbk.
schema_save_modeenumnoCREATE_SCHEMA_WHEN_NOT_EXISTHow to handle the target table before writing.
data_save_modeenumnoAPPEND_DATAHow to handle existing data before writing.
hbase_extra_configconfigno-Extra HBase or Hadoop client configuration.
multi_table_sink_replicaintno1Number of sink writer replicas for each table in a multi-table job.
common-optionsno-Sink plugin common parameters, such as plugin_input.

zookeeper_quorum [string]​

The zookeeper cluster host of hbase, example: "hadoop001:2181,hadoop002:2181,hadoop003:2181"

table [string]​

The table name you want to write, example: "seatunnel" If your table is under a custom namespace, use namespace:table (for example, ns1:seatunnel_test); if omitted, SeaTunnel will write to HBase's default namespace (default).

rowkey_column [list]​

The column name list of row keys, example: ["id", "uuid"]

family_name [config]​

The family name mapping of fields. For example the row from upstream like the following shown:

idnameage
1tyrantlucifer27

id as the row key and other fields written to the different families, you can assign

family_name { name = "info1" age = "info2" }

this means that name will be written to the family info1 and the age will be written to the family info2

if you want other fields written to the same family, you can assign

family_name { all_columns = "info" }

this means that all fields will be written to the family info

rowkey_delimiter [string]​

The delimiter of joining multi row keys, default ""

version_column [string]​

The version column name, you can use it to assign timestamp for hbase record

null_mode [string]​

The mode of writing null value, support [skip, empty], default skip

  • skip: When the field is null, connector will not write this field to hbase
  • empty: When the field is null, connector will write generate empty value for this field

wal_write [boolean]​

The wal log write flag, default false

write_buffer_size [int]​

The write buffer size of hbase client, default 8 * 1024 * 1024

encoding [string]​

The encoding used for STRING/DECIMAL/DATE/TIME/TIMESTAMP/ARRAY fields, support [utf8, gbk], default utf8

Data types​

Hbase stores bytes. The connector supports:

  • TINYINT/SMALLINT/INT/BIGINT/FLOAT/DOUBLE/BOOLEAN/BYTES
  • STRING/DECIMAL/DATE/TIME/TIMESTAMP/ARRAY (serialized as strings using encoding)

hbase_extra_config [config]​

The extra configuration of hbase

multi_table_sink_replica [int]​

The replica number of sink writers used for each table in a multi-table sink job. Keep the default value unless one table needs more sink writer parallelism.

schema_save_mode [enum]​

Controls how the sink handles the target HBase table before writing. Supported values include RECREATE_SCHEMA, CREATE_SCHEMA_WHEN_NOT_EXIST, and ERROR_WHEN_SCHEMA_NOT_EXIST.

data_save_mode [enum]​

Controls how the sink handles existing target data before writing. Supported values are DROP_DATA, APPEND_DATA, and ERROR_WHEN_DATA_EXISTS.

common options​

Sink plugin common parameters, please refer to Sink Common Options for details

Notes​

  • table = "${table_name}" routes each upstream table to an HBase table with the same name. You can also combine placeholders with fixed text, such as ods_${table_name}.
  • family_name { all_columns = "info" } writes every non-rowkey field to one column family. Use per-field mappings when fields should go to different column families.
  • schema_save_mode controls table creation or recreation. data_save_mode controls whether existing rows are kept, dropped, or treated as an error.

Example​

Write One Table​


Hbase {
  zookeeper_quorum = "hadoop001:2181,hadoop002:2181,hadoop003:2181"
  table = "seatunnel_test"
  rowkey_column = ["name"]
  family_name {
    all_columns = seatunnel
  }
}

Create Table When It Does Not Exist​

sink {
  Hbase {
    zookeeper_quorum = "hbase_e2e:2181"
    table = "seatunnel_test_with_create_when_not_exists"
    rowkey_column = ["name"]
    family_name {
      all_columns = info
    }
    schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
    data_save_mode = "APPEND_DATA"
  }
}

Kerberos Example​

Note:

  • connector-hbase does not parse krb5_path, kerberos_principal, or kerberos_keytab_path.
  • Prepare Kerberos credentials and krb5.conf in the runtime environment (for example, kinit -kt ... or JVM -Djava.security.krb5.conf=...), and put HBase/Hadoop security settings into hbase_extra_config.
sink {
  Hbase {
    zookeeper_quorum = "zk1:2181,zk2:2181,zk3:2181"
    table = "target_table"
    rowkey_column = ["rowkey"]
    family_name {
      all_columns = "info"
    }

    # HBase security config
    hbase_extra_config = {
      "hbase.security.authentication" = "kerberos"
      "hadoop.security.authentication" = "kerberos"
      "hbase.master.kerberos.principal" = "hbase/_HOST@REALM"
      "hbase.regionserver.kerberos.principal" = "hbase/_HOST@REALM"
      "hbase.rpc.protection" = "authentication"
      "hbase.zookeeper.useSasl" = "false"
    }
  }
}

Multiple Table​

env {
  # You can set engine configuration here
  execution.parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    tables_configs = [
       {
        schema = {
          table = "hbase_sink_1"
         fields {
                    name = STRING
                    c_string = STRING
                    c_double = DOUBLE
                    c_bigint = BIGINT
                    c_float = FLOAT
                    c_int = INT
                    c_smallint = SMALLINT
                    c_boolean = BOOLEAN
                    time = BIGINT
           }
        }
            rows = [
              {
                kind = INSERT
                fields = ["label_1", "sink_1", 4.3, 200, 2.5, 2, 5, true, 1627529632356]
              }
              ]
       },
       {
       schema = {
         table = "hbase_sink_2"
              fields {
                    name = STRING
                    c_string = STRING
                    c_double = DOUBLE
                    c_bigint = BIGINT
                    c_float = FLOAT
                    c_int = INT
                    c_smallint = SMALLINT
                    c_boolean = BOOLEAN
                    time = BIGINT
              }
       }
           rows = [
             {
               kind = INSERT
               fields = ["label_2", "sink_2", 4.3, 200, 2.5, 2, 5, true, 1627529632357]
             }
             ]
      }
    ]
  }
}

sink {
  Hbase {
    zookeeper_quorum = "hadoop001:2181,hadoop002:2181,hadoop003:2181"
    table = "${table_name}"
    rowkey_column = ["name"]
    family_name {
      all_columns = info
    }
  }
}

Writes To The Specified Column Family​

Hbase {
  zookeeper_quorum = "hbase_e2e:2181"
  table = "assign_cf_table"
  rowkey_column = ["id"]
  family_name {
    c_double = "cf1"
    c_bigint = "cf2"
  }
}

Changelog​

Change Log
ChangeCommitVersion
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve] hbase options (#8923)https://github.com/apache/seatunnel/commit/b6a702b58f2.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Fix][Connector-V2] Fix known directory create and delete ignore issues (#7700)https://github.com/apache/seatunnel/commit/e2fb6795772.3.8
[Feature][Connector-V2][Hbase] implement hbase catalog (#7516)https://github.com/apache/seatunnel/commit/b978792cb12.3.8
[Feature][Connector-V2] Support multi-table sink feature for HBase (#7169)https://github.com/apache/seatunnel/commit/025fa3bb882.3.8
[hotfix][connector-v2-hbase]fix and optimize hbase source problem (#7148)https://github.com/apache/seatunnel/commit/34a6b8e9f62.3.7
[Improve][hbase] The specified column is written to the specified column family (#5234)https://github.com/apache/seatunnel/commit/49d397c61d2.3.6
[feature][connector-v2-hbase-sink] Support Connector v2 HBase sink TTL data writing (#7116)https://github.com/apache/seatunnel/commit/adafd802552.3.6
[E2E][HBase]Refactor hbase e2e (#6859)https://github.com/apache/seatunnel/commit/1da9bd6ce42.3.6
[Connector]Add hbase source connector (#6348)https://github.com/apache/seatunnel/commit/f108a5e6582.3.6
[Feature][HbaseSink]support array data. (#6100)https://github.com/apache/seatunnel/commit/b5920147662.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
[Hotfix][Connector-v2][HbaseSink]Fix default timestamp (#4958)https://github.com/apache/seatunnel/commit/3d8f3bf9022.3.3
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector-V2][Hbase] Introduce hbase sink connector (#4049)https://github.com/apache/seatunnel/commit/68bda94a4c2.3.1