Hudi Pipeline Connector

The Hudi Pipeline Connector functions as a Data Sink for data pipelines, enabling data writes to Apache Hudi tables. This document explains how to configure the connector.

What can the connector do?

• Create table automatically if not exist
• Schema change synchronization
• Data synchronization

How to create Pipeline

The pipeline for reading data from MySQL and sink to Hudi can be defined as follows:

source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\.*, bdb.user_table_[0-9]+
  server-id: 5401-5404

sink:
  type: hudi 
  name: Hudi Sink
  path: /path/warehouse
  hoodie.table.type: MERGE_ON_READ

transform:
  - source-table: adb.\.*
    table-options: ordering.fields=ts1
  - source-table: bdb.user_table_[0-9]+
    table-options: ordering.fields=ts2

pipeline:
  name: MySQL to Hudi Pipeline
  parallelism: 4

Pipeline Connector Options

Option	Required	Default	Type	Description
type	required	(none)	String	Specify what connector to use, here should be hudi
path	required	(none)	String	Specify base path for all hudi tables
name	optional	(none)	String	The name of the sink
hoodie.table.type	optional	(none)	String	Type of table to write, currently only MERGE_ON_READ is supported
index.type	optional	(none)	String	Index type of Flink write job, currently only BUCKET is supported
table.properties.*	optional	(none)	String	Pass Hudi table options to the pipeline. See Hudi table options

Usage Notes

• The source table must have a primary key. Users can use transform rule to override the primary key configuration or set up ordering fields configuration.
• Only MERGE_ON_READ table with simple BUCKET index is supported now.
• Exactly-once semantics are not supported. The connector uses at-least-once + the table's primary key for idempotent writing.

Data Type Mapping

CDC type	Hudi type	NOTE
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)
BOOLEAN	BOOLEAN
DATE	DATE
TIME	TIME
TIMESTAMP	TIMESTAMP
BINARY	BINARY
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)