Paimon sink connector

Description

Sink connector for Apache Paimon. It can support cdc mode 、auto create table.

Key features

Options

name type required default value Description
warehouse String Yes - Paimon warehouse path
database String Yes - The database you want to access
table String Yes - The table you want to access
hdfs_site_path String No - The path of hdfs-site.xml
schema_save_mode Enum No CREATE_SCHEMA_WHEN_NOT_EXIST The schema save mode
data_save_mode Enum No APPEND_DATA The data save mode
paimon.table.primary-keys String No - Default comma-separated list of columns (primary key) that identify a row in tables.(Notice: The partition field needs to be included in the primary key fields)
paimon.table.partition-keys String No - Default comma-separated list of partition fields to use when creating tables.
paimon.table.write-props Map No - Properties passed through to paimon table initialization, reference.
paimon.hadoop.conf Map No - Properties in hadoop conf
paimon.hadoop.conf-path String No - The specified loading path for the ‘core-site.xml’, ‘hdfs-site.xml’, ‘hive-site.xml’ files

Examples

Single table

  1. env {
  2.   parallelism = 1
  3.   job.mode = "STREAMING"
  4.   checkpoint.interval = 5000
  5. }
  7. source {
  8.   Mysql-CDC {
  9.     base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
  10.     username = "root"
  11.     password = "******"
  12.     table-names = ["seatunnel.role"]
  13.   }
  14. }
  16. transform {
  17. }
  19. sink {
  20.   Paimon {
  21.     catalog_name="seatunnel_test"
  22.     warehouse="file:///tmp/seatunnel/paimon/hadoop-sink/"
  23.     database="seatunnel"
  24.     table="role"
  25.   }
  26. }

Single table(Specify hadoop HA config and kerberos config)

  1. env {
  2.   parallelism = 1
  3.   job.mode = "STREAMING"
  4.   checkpoint.interval = 5000
  5. }
  7. source {
  8.   Mysql-CDC {
  9.     base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
  10.     username = "root"
  11.     password = "******"
  12.     table-names = ["seatunnel.role"]
  13.   }
  14. }
  16. transform {
  17. }
  19. sink {
  20.   Paimon {
  21.     catalog_name="seatunnel_test"
  22.     warehouse="hdfs:///tmp/seatunnel/paimon/hadoop-sink/"
  23.     database="seatunnel"
  24.     table="role"
  25.     paimon.hadoop.conf = {
  26.       fs.defaultFS = "hdfs://nameservice1"
  27.       dfs.nameservices = "nameservice1"
  28.       dfs.ha.namenodes.nameservice1 = "nn1,nn2"
  29.       dfs.namenode.rpc-address.nameservice1.nn1 = "hadoop03:8020"
  30.       dfs.namenode.rpc-address.nameservice1.nn2 = "hadoop04:8020"
  31.       dfs.client.failover.proxy.provider.nameservice1 = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
  32.       dfs.client.use.datanode.hostname = "true"
  33.       security.kerberos.login.principal = "your-kerberos-principal"
  34.       security.kerberos.login.keytab = "your-kerberos-keytab-path"
  35.     }
  36.   }
  37. }

Single table with write props of paimon

  1. env {
  2.   parallelism = 1
  3.   job.mode = "STREAMING"
  4.   checkpoint.interval = 5000
  5. }
  7. source {
  8.   Mysql-CDC {
  9.     base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
  10.     username = "root"
  11.     password = "******"
  12.     table-names = ["seatunnel.role"]
  13.   }
  14. }
  16. sink {
  17.   Paimon {
  18.     catalog_name="seatunnel_test"
  19.     warehouse="file:///tmp/seatunnel/paimon/hadoop-sink/"
  20.     database="seatunnel"
  21.     table="role"
  22.     paimon.table.write-props = {
  23.         bucket = 2
  24.         file.format = "parquet"
  25.     }
  26.     paimon.table.partition-keys = "dt"
  27.     paimon.table.primary-keys = "pk_id,dt"
  28.   }
  29. }

Multiple table

  1. env {
  2.   parallelism = 1
  3.   job.mode = "STREAMING"
  4.   checkpoint.interval = 5000
  5. }
  7. source {
  8.   Mysql-CDC {
  9.     base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
  10.     username = "root"
  11.     password = "******"
  12.     table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]
  13.   }
  14. }
  16. transform {
  17. }
  19. sink {
  20.   Paimon {
  21.     catalog_name="seatunnel_test"
  22.     warehouse="file:///tmp/seatunnel/paimon/hadoop-sink/"
  23.     database="${database_name}"
  24.     table="${table_name}"
  25.   }
  26. }