Kudu sink connector

Support Kudu Version

  • 1.11.1/1.12.0/1.13.0/1.14.0/1.15.0

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Key Features

Data Type Mapping

SeaTunnel Data Type Kudu Data Type
BOOLEAN BOOL
INT INT8
INT16
INT32
BIGINT INT64
DECIMAL DECIMAL
FLOAT FLOAT
DOUBLE DOUBLE
STRING STRING
TIMESTAMP UNIXTIME_MICROS
BYTES BINARY

Sink Options

Name Type Required Default Description
kudu_masters String Yes - Kudu master address. Separated by ‘,’,such as ‘192.168.88.110:7051’.
table_name String Yes - The name of kudu table.
client_worker_count Int No 2 * Runtime.getRuntime().availableProcessors() Kudu worker count. Default value is twice the current number of cpu cores.
client_default_operation_timeout_ms Long No 30000 Kudu normal operation time out.
client_default_admin_operation_timeout_ms Long No 30000 Kudu admin operation time out.
enable_kerberos Bool No false Kerberos principal enable.
kerberos_principal String No - Kerberos principal. Note that all zeta nodes require have this file.
kerberos_keytab String No - Kerberos keytab. Note that all zeta nodes require have this file.
kerberos_krb5conf String No - Kerberos krb5 conf. Note that all zeta nodes require have this file.
save_mode String No - Storage mode, support overwrite and append.
session_flush_mode String No AUTO_FLUSH_SYNC Kudu flush mode. Default AUTO_FLUSH_SYNC.
batch_size Int No 1024 The flush max size (includes all append, upsert and delete records), over this number of records, will flush data. The default value is 100
buffer_flush_interval Int No 10000 The flush interval mills, over this time, asynchronous threads will flush data.
ignore_not_found Bool No false If true, ignore all not found rows.
ignore_not_duplicate Bool No false If true, ignore all dulicate rows.
common-options No - Source plugin common parameters, please refer to Source Common Options for details.

Task Example

Simple:

The following example refers to a FakeSource named “kudu” cdc write kudu table “kudu_sink_table”

  2. env {
  3.   parallelism = 1
  4.   job.mode = "BATCH"
  5. }
  6.     source {
  7.       FakeSource {
  8.        result_table_name = "kudu"
  9.         schema = {
  10.           fields {
  11.                     id = int
  12.                     val_bool = boolean
  13.                     val_int8 = tinyint
  14.                     val_int16 = smallint
  15.                     val_int32 = int
  16.                     val_int64 = bigint
  17.                     val_float = float
  18.                     val_double = double
  19.                     val_decimal = "decimal(16, 1)"
  20.                     val_string = string
  21.                     val_unixtime_micros = timestamp
  22.           }
  23.         }
  24.         rows = [
  25.           {
  26.             kind = INSERT
  27.             fields = [1, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  28.           },
  29.           {
  30.             kind = INSERT
  31.             fields = [2, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  32.           },
  33.           {
  34.             kind = INSERT
  35.             fields = [3, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  36.           },
  37.           {
  38.             kind = UPDATE_BEFORE
  39.             fields = [1, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  40.           },
  41.           {
  42.             kind = UPDATE_AFTER
  43.            fields = [1, true, 2, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  44.           },
  45.           {
  46.             kind = DELETE
  47.             fields = [2, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  48.           }
  49.         ]
  50.       }
  51.     }
  53. sink {
  54.    kudu{
  55.     source_table_name = "kudu"
  56.     kudu_masters = "kudu-master-cdc:7051"
  57.     table_name = "kudu_sink_table"
  58.     enable_kerberos = true
  59.     kerberos_principal = "xx@xx.COM"
  60.     kerberos_keytab = "xx.keytab"
  61.  }
  62. }

Multiple Table

  1. env {
  2.   # You can set engine configuration here
  3.   parallelism = 1
  4.   job.mode = "BATCH"
  5. }
  7. source {
  8.   FakeSource {
  9.     tables_configs = [
  10.        {
  11.         schema = {
  12.           table = "kudu_sink_1"
  13.          fields {
  14.                 id = int
  15.                 val_bool = boolean
  16.                 val_int8 = tinyint
  17.                 val_int16 = smallint
  18.                 val_int32 = int
  19.                 val_int64 = bigint
  20.                 val_float = float
  21.                 val_double = double
  22.                 val_decimal = "decimal(16, 1)"
  23.                 val_string = string
  24.                 val_unixtime_micros = timestamp
  25.       }
  26.         }
  27.             rows = [
  28.               {
  29.                 kind = INSERT
  30.                 fields = [1, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  31.               }
  32.               ]
  33.        },
  34.        {
  35.        schema = {
  36.          table = "kudu_sink_2"
  37.               fields {
  38.                         id = int
  39.                         val_bool = boolean
  40.                         val_int8 = tinyint
  41.                         val_int16 = smallint
  42.                         val_int32 = int
  43.                         val_int64 = bigint
  44.                         val_float = float
  45.                         val_double = double
  46.                         val_decimal = "decimal(16, 1)"
  47.                         val_string = string
  48.                         val_unixtime_micros = timestamp
  49.               }
  50.        }
  51.            rows = [
  52.              {
  53.                kind = INSERT
  54.                fields = [1, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
  55.              }
  56.              ]
  57.       }
  58.     ]
  59.   }
  60. }
  63. sink {
  64.    kudu{
  65.     kudu_masters = "kudu-master-multiple:7051"
  66.  }
  67. }

Changelog

2.2.0-beta 2022-09-26

  • Add Kudu Sink Connector

2.3.0-beta 2022-10-20

  • [Improve] Kudu Sink Connector Support to upsert row (2881)

Next Version

  • Change plugin name from KuduSink to Kudu 3432