org.apache.paimon.flink.action.cdc.mysql

Class MySqlSyncDatabaseAction

java.lang.Object

org.apache.paimon.flink.action.ActionBase

org.apache.paimon.flink.action.cdc.SynchronizationActionBase

org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase

org.apache.paimon.flink.action.cdc.mysql.MySqlSyncDatabaseAction

All Implemented Interfaces:

Action

public class MySqlSyncDatabaseAction
extends SyncDatabaseActionBase

An Action which synchronize the whole MySQL database into one Paimon database.

You should specify MySQL source database in mySqlConfig. See document of flink-cdc-connectors for detailed keys and values.

For each MySQL table to be synchronized, if the corresponding Paimon table does not exist, this action will automatically create the table. Its schema will be derived from all specified MySQL tables. If the Paimon table already exists, its schema will be compared against the schema of all specified MySQL tables.

This action supports a limited number of schema changes. Currently, the framework can not drop columns, so the behaviors of `DROP` will be ignored, `RENAME` will add a new column. Currently supported schema changes includes:

• Adding columns.
• Altering column types. More specifically,
  • altering from a string type (char, varchar, text) to another string type with longer length,
  • altering from a binary type (binary, varbinary, blob) to another binary type with longer length,
  • altering from an integer type (tinyint, smallint, int, bigint) to another integer type with wider range,
  • altering from a floating-point type (float, double) to another floating-point type with wider range,
  are supported.

This action creates a Paimon table sink for each Paimon table to be written, so this action is not very efficient in resource saving. We may optimize this action by merging all sinks into one instance in the future.

Field Summary

Fields inherited from class org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase

excludingTables, includingTables, mergeShards, mode, partitionKeyMultiple, partitionKeys, primaryKeys, tablePrefix, tables, tableSuffix

Fields inherited from class org.apache.paimon.flink.action.cdc.SynchronizationActionBase

allowUpperCase, cdcSourceConfig, database, metadataConverters, syncJobHandler, tableConfig, typeMapping

Fields inherited from class org.apache.paimon.flink.action.ActionBase

batchTEnv, catalog, catalogName, catalogOptions, env, flinkCatalog

Constructor Summary

Constructors

Constructor and Description
MySqlSyncDatabaseAction(String warehouse, String database, Map<String,String> catalogConfig, Map<String,String> mySqlConfig)

Method Summary

Modifier and Type	Method and Description
protected void	beforeBuildingSourceSink()
protected org.apache.flink.cdc.connectors.mysql.source.MySqlSource<CdcSourceRecord>	buildSource()
protected CdcTimestampExtractor	createCdcTimestampExtractor()
List<Identifier>	excludedTables()
MySqlSyncDatabaseAction	ignoreIncompatible(boolean ignoreIncompatible)
List<Identifier>	monitoredTables()
protected boolean	requirePrimaryKeys()

Methods inherited from class org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase

buildEventParserFactory, buildSink, excludingTables, includingTables, mergeShards, recordParse, validateCaseSensitivity, withMode, withPartitionKeyMultiple, withPartitionKeys, withPrimaryKeys, withTablePrefix, withTableSuffix

Methods inherited from class org.apache.paimon.flink.action.cdc.SynchronizationActionBase

alterTableOptions, build, run, tableConfig, validateRuntimeExecutionMode, withMetadataColumns, withTableConfig, withTypeMapping

Methods inherited from class org.apache.paimon.flink.action.ActionBase

catalogConfig, catalogLoader, compatibleCheck, execute, initFlinkCatalog, initFlinkEnv, initPaimonCatalog, toPaimonTypes, withStreamExecutionEnvironment

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

MySqlSyncDatabaseAction

public MySqlSyncDatabaseAction(String warehouse,
                               String database,
                               Map<String,String> catalogConfig,
                               Map<String,String> mySqlConfig)

Method Detail

ignoreIncompatible

public MySqlSyncDatabaseAction ignoreIncompatible(boolean ignoreIncompatible)

beforeBuildingSourceSink

protected void beforeBuildingSourceSink()
                                 throws Exception

Overrides:

beforeBuildingSourceSink in class SynchronizationActionBase

Throws:

Exception

createCdcTimestampExtractor

protected CdcTimestampExtractor createCdcTimestampExtractor()

Overrides:

createCdcTimestampExtractor in class SynchronizationActionBase

buildSource

protected org.apache.flink.cdc.connectors.mysql.source.MySqlSource<CdcSourceRecord> buildSource()

Overrides:

buildSource in class SynchronizationActionBase

monitoredTables

@VisibleForTesting
public List<Identifier> monitoredTables()

excludedTables

@VisibleForTesting
public List<Identifier> excludedTables()

requirePrimaryKeys

protected boolean requirePrimaryKeys()

Specified by:

requirePrimaryKeys in class SyncDatabaseActionBase