You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2023/12/06 04:09:50 UTC
(incubator-paimon) branch master updated: [cdc] Refactor database synchronization action (#2426)
This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 182d7c424 [cdc] Refactor database synchronization action (#2426)
182d7c424 is described below
commit 182d7c42452a254ac34124d8f3c9fb947158f32c
Author: Kerwin <37...@users.noreply.github.com>
AuthorDate: Wed Dec 6 12:09:46 2023 +0800
[cdc] Refactor database synchronization action (#2426)
---
.../cdc/MessageQueueSyncDatabaseActionBase.java | 128 +--------------
...ActionBase.java => SyncDatabaseActionBase.java} | 132 +++++++--------
.../cdc/mongodb/MongoDBSyncDatabaseAction.java | 124 +++-----------
.../action/cdc/mysql/MySqlSyncDatabaseAction.java | 178 +++++----------------
.../cdc/mysql/MySqlSyncDatabaseActionFactory.java | 6 +-
5 files changed, 133 insertions(+), 435 deletions(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
index 8ed169a73..cc869a85b 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
@@ -18,28 +18,15 @@
package org.apache.paimon.flink.action.cdc;
-import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.catalog.AbstractCatalog;
import org.apache.paimon.flink.action.Action;
-import org.apache.paimon.flink.action.ActionBase;
import org.apache.paimon.flink.action.MultiTablesSinkMode;
import org.apache.paimon.flink.action.cdc.format.DataFormat;
-import org.apache.paimon.flink.action.cdc.format.RecordParser;
-import org.apache.paimon.flink.sink.cdc.EventParser;
-import org.apache.paimon.flink.sink.cdc.FlinkCdcSyncDatabaseSinkBuilder;
-import org.apache.paimon.flink.sink.cdc.NewTableSchemaBuilder;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordEventParser;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
-
-import javax.annotation.Nullable;
+import org.apache.flink.api.common.functions.FlatMapFunction;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
-import java.util.regex.Pattern;
/**
* An {@link Action} which synchronize the Multiple message queue topics into one Paimon database.
@@ -73,125 +60,22 @@ import java.util.regex.Pattern;
* <p>To automatically synchronize new table, This action creates a single sink for all Paimon
* tables to be written. See {@link MultiTablesSinkMode#COMBINED}.
*/
-public abstract class MessageQueueSyncDatabaseActionBase extends ActionBase {
-
- protected final String database;
- protected final Configuration cdcSourceConfig;
-
- private Map<String, String> tableConfig = new HashMap<>();
- private String tablePrefix = "";
- private String tableSuffix = "";
- private String includingTables = ".*";
- @Nullable String excludingTables;
- private TypeMapping typeMapping = TypeMapping.defaultMapping();
+public abstract class MessageQueueSyncDatabaseActionBase extends SyncDatabaseActionBase {
public MessageQueueSyncDatabaseActionBase(
String warehouse,
String database,
Map<String, String> catalogConfig,
Map<String, String> mqConfig) {
- super(warehouse, catalogConfig);
- this.database = database;
- this.cdcSourceConfig = Configuration.fromMap(mqConfig);
- }
-
- public MessageQueueSyncDatabaseActionBase withTableConfig(Map<String, String> tableConfig) {
- this.tableConfig = tableConfig;
- return this;
- }
-
- public MessageQueueSyncDatabaseActionBase withTablePrefix(@Nullable String tablePrefix) {
- if (tablePrefix != null) {
- this.tablePrefix = tablePrefix;
- }
- return this;
- }
-
- public MessageQueueSyncDatabaseActionBase withTableSuffix(@Nullable String tableSuffix) {
- if (tableSuffix != null) {
- this.tableSuffix = tableSuffix;
- }
- return this;
- }
-
- public MessageQueueSyncDatabaseActionBase includingTables(@Nullable String includingTables) {
- if (includingTables != null) {
- this.includingTables = includingTables;
- }
- return this;
- }
-
- public MessageQueueSyncDatabaseActionBase excludingTables(@Nullable String excludingTables) {
- this.excludingTables = excludingTables;
- return this;
- }
-
- public MessageQueueSyncDatabaseActionBase withTypeMapping(TypeMapping typeMapping) {
- this.typeMapping = typeMapping;
- return this;
+ super(warehouse, database, catalogConfig, mqConfig);
}
- protected abstract DataStreamSource<String> buildSource() throws Exception;
-
- protected abstract String sourceName();
-
protected abstract DataFormat getDataFormat();
- protected abstract String jobName();
-
@Override
- public void build() throws Exception {
- boolean caseSensitive = catalog.caseSensitive();
-
- validateCaseInsensitive(caseSensitive);
-
- catalog.createDatabase(database, true);
-
+ protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
DataFormat format = getDataFormat();
- RecordParser recordParser =
- format.createParser(caseSensitive, typeMapping, Collections.emptyList());
- NewTableSchemaBuilder schemaBuilder = new NewTableSchemaBuilder(tableConfig, caseSensitive);
- Pattern includingPattern = Pattern.compile(includingTables);
- Pattern excludingPattern =
- excludingTables == null ? null : Pattern.compile(excludingTables);
- TableNameConverter tableNameConverter =
- new TableNameConverter(caseSensitive, true, tablePrefix, tableSuffix);
- EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
- () ->
- new RichCdcMultiplexRecordEventParser(
- schemaBuilder,
- includingPattern,
- excludingPattern,
- tableNameConverter);
-
- new FlinkCdcSyncDatabaseSinkBuilder<RichCdcMultiplexRecord>()
- .withInput(buildSource().flatMap(recordParser).name("Parse"))
- .withParserFactory(parserFactory)
- .withCatalogLoader(catalogLoader())
- .withDatabase(database)
- .withMode(MultiTablesSinkMode.COMBINED)
- .withTableOptions(tableConfig)
- .build();
- }
-
- private void validateCaseInsensitive(boolean caseSensitive) {
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database", database);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table prefix", tablePrefix);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table suffix", tableSuffix);
- }
-
- @VisibleForTesting
- public Map<String, String> tableConfig() {
- return tableConfig;
- }
-
- // ------------------------------------------------------------------------
- // Flink run methods
- // ------------------------------------------------------------------------
-
- @Override
- public void run() throws Exception {
- build();
- execute(jobName());
+ boolean caseSensitive = catalog.caseSensitive();
+ return format.createParser(caseSensitive, typeMapping, Collections.emptyList());
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
similarity index 57%
copy from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
copy to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
index 8ed169a73..d31c6cc12 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
@@ -23,139 +23,142 @@ import org.apache.paimon.catalog.AbstractCatalog;
import org.apache.paimon.flink.action.Action;
import org.apache.paimon.flink.action.ActionBase;
import org.apache.paimon.flink.action.MultiTablesSinkMode;
-import org.apache.paimon.flink.action.cdc.format.DataFormat;
-import org.apache.paimon.flink.action.cdc.format.RecordParser;
import org.apache.paimon.flink.sink.cdc.EventParser;
import org.apache.paimon.flink.sink.cdc.FlinkCdcSyncDatabaseSinkBuilder;
import org.apache.paimon.flink.sink.cdc.NewTableSchemaBuilder;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordEventParser;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import javax.annotation.Nullable;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.regex.Pattern;
-/**
- * An {@link Action} which synchronize the Multiple message queue topics into one Paimon database.
- *
- * <p>For each message queue topic's table to be synchronized, if the corresponding Paimon table
- * does not exist, this action will automatically create the table, and its schema will be derived
- * from all specified message queue topic's tables. If the Paimon table already exists and its
- * schema is different from that parsed from message queue record, this action will try to preform
- * schema evolution.
- *
- * <p>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:
- *
- * <ul>
- * <li>Adding columns.
- * <li>Altering column types. More specifically,
- * <ul>
- * <li>altering from a string type (char, varchar, text) to another string type with longer
- * length,
- * <li>altering from a binary type (binary, varbinary, blob) to another binary type with
- * longer length,
- * <li>altering from an integer type (tinyint, smallint, int, bigint) to another integer
- * type with wider range,
- * <li>altering from a floating-point type (float, double) to another floating-point type
- * with wider range,
- * </ul>
- * are supported.
- * </ul>
- *
- * <p>To automatically synchronize new table, This action creates a single sink for all Paimon
- * tables to be written. See {@link MultiTablesSinkMode#COMBINED}.
- */
-public abstract class MessageQueueSyncDatabaseActionBase extends ActionBase {
+import static org.apache.paimon.flink.action.MultiTablesSinkMode.COMBINED;
+
+/** Base {@link Action} for synchronizing into one Paimon database. */
+public abstract class SyncDatabaseActionBase extends ActionBase {
protected final String database;
protected final Configuration cdcSourceConfig;
-
- private Map<String, String> tableConfig = new HashMap<>();
- private String tablePrefix = "";
- private String tableSuffix = "";
- private String includingTables = ".*";
- @Nullable String excludingTables;
- private TypeMapping typeMapping = TypeMapping.defaultMapping();
-
- public MessageQueueSyncDatabaseActionBase(
+ protected Map<String, String> tableConfig = new HashMap<>();
+ protected boolean mergeShards = true;
+ protected MultiTablesSinkMode mode = COMBINED;
+ protected String tablePrefix = "";
+ protected String tableSuffix = "";
+ protected String includingTables = ".*";
+ @Nullable protected String excludingTables;
+ protected TypeMapping typeMapping = TypeMapping.defaultMapping();
+
+ protected CdcMetadataConverter[] metadataConverters = new CdcMetadataConverter[] {};
+ protected List<FileStoreTable> tables = new ArrayList<>();
+
+ public SyncDatabaseActionBase(
String warehouse,
String database,
Map<String, String> catalogConfig,
- Map<String, String> mqConfig) {
+ Map<String, String> cdcSourceConfig) {
super(warehouse, catalogConfig);
this.database = database;
- this.cdcSourceConfig = Configuration.fromMap(mqConfig);
+ this.cdcSourceConfig = Configuration.fromMap(cdcSourceConfig);
}
- public MessageQueueSyncDatabaseActionBase withTableConfig(Map<String, String> tableConfig) {
+ public SyncDatabaseActionBase withTableConfig(Map<String, String> tableConfig) {
this.tableConfig = tableConfig;
return this;
}
- public MessageQueueSyncDatabaseActionBase withTablePrefix(@Nullable String tablePrefix) {
+ public SyncDatabaseActionBase mergeShards(boolean mergeShards) {
+ this.mergeShards = mergeShards;
+ return this;
+ }
+
+ public SyncDatabaseActionBase withMode(MultiTablesSinkMode mode) {
+ this.mode = mode;
+ return this;
+ }
+
+ public SyncDatabaseActionBase withTablePrefix(@Nullable String tablePrefix) {
if (tablePrefix != null) {
this.tablePrefix = tablePrefix;
}
return this;
}
- public MessageQueueSyncDatabaseActionBase withTableSuffix(@Nullable String tableSuffix) {
+ public SyncDatabaseActionBase withTableSuffix(@Nullable String tableSuffix) {
if (tableSuffix != null) {
this.tableSuffix = tableSuffix;
}
return this;
}
- public MessageQueueSyncDatabaseActionBase includingTables(@Nullable String includingTables) {
+ public SyncDatabaseActionBase includingTables(@Nullable String includingTables) {
if (includingTables != null) {
this.includingTables = includingTables;
}
return this;
}
- public MessageQueueSyncDatabaseActionBase excludingTables(@Nullable String excludingTables) {
+ public SyncDatabaseActionBase excludingTables(@Nullable String excludingTables) {
this.excludingTables = excludingTables;
return this;
}
- public MessageQueueSyncDatabaseActionBase withTypeMapping(TypeMapping typeMapping) {
+ public SyncDatabaseActionBase withTypeMapping(TypeMapping typeMapping) {
this.typeMapping = typeMapping;
return this;
}
+ public SyncDatabaseActionBase withMetadataColumns(List<String> metadataColumns) {
+ this.metadataConverters =
+ metadataColumns.stream()
+ .map(this::metadataConverter)
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .toArray(CdcMetadataConverter[]::new);
+ return this;
+ }
+
+ protected Optional<CdcMetadataConverter<?>> metadataConverter(String column) {
+ return Optional.empty();
+ }
+
+ protected void checkCdcSourceArgument() {}
+
protected abstract DataStreamSource<String> buildSource() throws Exception;
protected abstract String sourceName();
- protected abstract DataFormat getDataFormat();
-
- protected abstract String jobName();
+ protected abstract FlatMapFunction<String, RichCdcMultiplexRecord> recordParse();
@Override
public void build() throws Exception {
+ checkCdcSourceArgument();
boolean caseSensitive = catalog.caseSensitive();
validateCaseInsensitive(caseSensitive);
catalog.createDatabase(database, true);
- DataFormat format = getDataFormat();
- RecordParser recordParser =
- format.createParser(caseSensitive, typeMapping, Collections.emptyList());
NewTableSchemaBuilder schemaBuilder = new NewTableSchemaBuilder(tableConfig, caseSensitive);
Pattern includingPattern = Pattern.compile(includingTables);
Pattern excludingPattern =
excludingTables == null ? null : Pattern.compile(excludingTables);
TableNameConverter tableNameConverter =
- new TableNameConverter(caseSensitive, true, tablePrefix, tableSuffix);
+ new TableNameConverter(caseSensitive, mergeShards, tablePrefix, tableSuffix);
+
+ DataStream<RichCdcMultiplexRecord> input =
+ buildSource().flatMap(recordParse()).name("Parse");
EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
() ->
new RichCdcMultiplexRecordEventParser(
@@ -165,16 +168,17 @@ public abstract class MessageQueueSyncDatabaseActionBase extends ActionBase {
tableNameConverter);
new FlinkCdcSyncDatabaseSinkBuilder<RichCdcMultiplexRecord>()
- .withInput(buildSource().flatMap(recordParser).name("Parse"))
+ .withInput(input)
.withParserFactory(parserFactory)
.withCatalogLoader(catalogLoader())
.withDatabase(database)
- .withMode(MultiTablesSinkMode.COMBINED)
+ .withTables(tables)
+ .withMode(mode)
.withTableOptions(tableConfig)
.build();
}
- private void validateCaseInsensitive(boolean caseSensitive) {
+ protected void validateCaseInsensitive(boolean caseSensitive) {
AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database", database);
AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table prefix", tablePrefix);
AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table suffix", tableSuffix);
@@ -189,6 +193,8 @@ public abstract class MessageQueueSyncDatabaseActionBase extends ActionBase {
// Flink run methods
// ------------------------------------------------------------------------
+ protected abstract String jobName();
+
@Override
public void run() throws Exception {
build();
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
index 03cb93c6d..2ced0edb9 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
@@ -18,31 +18,20 @@
package org.apache.paimon.flink.action.cdc.mongodb;
-import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.catalog.AbstractCatalog;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.flink.action.ActionBase;
-import org.apache.paimon.flink.action.MultiTablesSinkMode;
import org.apache.paimon.flink.action.cdc.CdcActionCommonUtils;
-import org.apache.paimon.flink.action.cdc.TableNameConverter;
-import org.apache.paimon.flink.sink.cdc.EventParser;
-import org.apache.paimon.flink.sink.cdc.FlinkCdcSyncDatabaseSinkBuilder;
-import org.apache.paimon.flink.sink.cdc.NewTableSchemaBuilder;
+import org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordEventParser;
import com.ververica.cdc.connectors.mongodb.source.MongoDBSource;
import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.configuration.Configuration;
-
-import javax.annotation.Nullable;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.regex.Pattern;
/**
* An action class responsible for synchronizing MongoDB databases with a target system.
@@ -63,119 +52,42 @@ import java.util.regex.Pattern;
* <p>Note: This action is primarily intended for use in Flink streaming applications that
* synchronize MongoDB data with other systems.
*/
-public class MongoDBSyncDatabaseAction extends ActionBase {
-
- private final String database;
- private final Configuration mongodbConfig;
- private Map<String, String> tableConfig = new HashMap<>();
- private String tablePrefix = "";
- private String tableSuffix = "";
- private String includingTables = ".*";
- @Nullable String excludingTables;
+public class MongoDBSyncDatabaseAction extends SyncDatabaseActionBase {
public MongoDBSyncDatabaseAction(
String warehouse,
String database,
Map<String, String> catalogConfig,
Map<String, String> mongodbConfig) {
- super(warehouse, catalogConfig);
- this.database = database;
- this.mongodbConfig = Configuration.fromMap(mongodbConfig);
- }
-
- public MongoDBSyncDatabaseAction withTableConfig(Map<String, String> tableConfig) {
- this.tableConfig = tableConfig;
- return this;
- }
-
- public MongoDBSyncDatabaseAction withTablePrefix(@Nullable String tablePrefix) {
- if (tablePrefix != null) {
- this.tablePrefix = tablePrefix;
- }
- return this;
- }
-
- public MongoDBSyncDatabaseAction withTableSuffix(@Nullable String tableSuffix) {
- if (tableSuffix != null) {
- this.tableSuffix = tableSuffix;
- }
- return this;
- }
-
- public MongoDBSyncDatabaseAction includingTables(@Nullable String includingTables) {
- if (includingTables != null) {
- this.includingTables = includingTables;
- }
- return this;
- }
-
- public MongoDBSyncDatabaseAction excludingTables(@Nullable String excludingTables) {
- this.excludingTables = excludingTables;
- return this;
+ super(warehouse, database, catalogConfig, mongodbConfig);
}
@Override
- public void build() throws Exception {
- boolean caseSensitive = catalog.caseSensitive();
-
- validateCaseInsensitive(caseSensitive);
-
- catalog.createDatabase(database, true);
-
+ protected DataStreamSource<String> buildSource() throws Exception {
List<Identifier> excludedTables = new ArrayList<>();
MongoDBSource<String> source =
MongoDBActionUtils.buildMongodbSource(
- mongodbConfig,
+ cdcSourceConfig,
CdcActionCommonUtils.combinedModeTableList(
- mongodbConfig.get(MongoDBSourceOptions.DATABASE),
+ cdcSourceConfig.get(MongoDBSourceOptions.DATABASE),
includingTables,
excludedTables));
-
- EventParser.Factory<RichCdcMultiplexRecord> parserFactory;
- NewTableSchemaBuilder schemaBuilder = new NewTableSchemaBuilder(tableConfig, caseSensitive);
- Pattern includingPattern = Pattern.compile(this.includingTables);
- Pattern excludingPattern =
- excludingTables == null ? null : Pattern.compile(excludingTables);
- TableNameConverter tableNameConverter =
- new TableNameConverter(caseSensitive, true, tablePrefix, tableSuffix);
- parserFactory =
- () ->
- new RichCdcMultiplexRecordEventParser(
- schemaBuilder,
- includingPattern,
- excludingPattern,
- tableNameConverter);
- new FlinkCdcSyncDatabaseSinkBuilder<RichCdcMultiplexRecord>()
- .withInput(
- env.fromSource(source, WatermarkStrategy.noWatermarks(), "MongoDB Source")
- .flatMap(new MongoDBRecordParser(caseSensitive, mongodbConfig))
- .name("Parse"))
- .withParserFactory(parserFactory)
- .withCatalogLoader(catalogLoader())
- .withDatabase(database)
- .withMode(MultiTablesSinkMode.COMBINED)
- .withTableOptions(tableConfig)
- .build();
+ return env.fromSource(source, WatermarkStrategy.noWatermarks(), sourceName());
}
- private void validateCaseInsensitive(boolean caseSensitive) {
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database", database);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table prefix", tablePrefix);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table suffix", tableSuffix);
+ @Override
+ protected String sourceName() {
+ return "MongoDB Source";
}
- @VisibleForTesting
- public Map<String, String> tableConfig() {
- return tableConfig;
+ @Override
+ protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
+ boolean caseSensitive = catalog.caseSensitive();
+ return new MongoDBRecordParser(caseSensitive, cdcSourceConfig);
}
- // ------------------------------------------------------------------------
- // Flink run methods
- // ------------------------------------------------------------------------
-
@Override
- public void run() throws Exception {
- build();
- execute(String.format("MongoDB-Paimon Database Sync: %s", database));
+ protected String jobName() {
+ return String.format("MongoDB-Paimon Database Sync: %s", database);
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index 11b532461..e44aeb452 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -19,23 +19,16 @@
package org.apache.paimon.flink.action.cdc.mysql;
import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.catalog.AbstractCatalog;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.action.Action;
-import org.apache.paimon.flink.action.ActionBase;
-import org.apache.paimon.flink.action.MultiTablesSinkMode;
import org.apache.paimon.flink.action.cdc.CdcActionCommonUtils;
import org.apache.paimon.flink.action.cdc.CdcMetadataConverter;
+import org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase;
import org.apache.paimon.flink.action.cdc.TableNameConverter;
-import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.flink.action.cdc.mysql.schema.MySqlSchemasInfo;
import org.apache.paimon.flink.action.cdc.mysql.schema.MySqlTableInfo;
-import org.apache.paimon.flink.sink.cdc.EventParser;
-import org.apache.paimon.flink.sink.cdc.FlinkCdcSyncDatabaseSinkBuilder;
-import org.apache.paimon.flink.sink.cdc.NewTableSchemaBuilder;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordEventParser;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
@@ -44,7 +37,8 @@ import org.apache.paimon.utils.Preconditions;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.config.MySqlSourceOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,9 +46,9 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -100,114 +94,55 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
* not very efficient in resource saving. We may optimize this action by merging all sinks into one
* instance in the future.
*/
-public class MySqlSyncDatabaseAction extends ActionBase {
+public class MySqlSyncDatabaseAction extends SyncDatabaseActionBase {
private static final Logger LOG = LoggerFactory.getLogger(MySqlSyncDatabaseAction.class);
- private final String database;
- private final Configuration mySqlConfig;
-
- private Map<String, String> tableConfig = new HashMap<>();
private boolean ignoreIncompatible = false;
- private boolean mergeShards = true;
- private String tablePrefix = "";
- private String tableSuffix = "";
- private String includingTables = ".*";
- @Nullable String excludingTables;
- private MultiTablesSinkMode mode = DIVIDED;
- private TypeMapping typeMapping = TypeMapping.defaultMapping();
// for test purpose
private final List<Identifier> monitoredTables = new ArrayList<>();
private final List<Identifier> excludedTables = new ArrayList<>();
- private List<String> metadataColumn = new ArrayList<>();
public MySqlSyncDatabaseAction(
String warehouse,
String database,
Map<String, String> catalogConfig,
Map<String, String> mySqlConfig) {
- super(warehouse, catalogConfig);
- this.database = database;
- this.mySqlConfig = Configuration.fromMap(mySqlConfig);
-
+ super(warehouse, database, catalogConfig, mySqlConfig);
+ this.mode = DIVIDED;
MySqlActionUtils.registerJdbcDriver();
}
- public MySqlSyncDatabaseAction withTableConfig(Map<String, String> tableConfig) {
- this.tableConfig = tableConfig;
- return this;
- }
-
public MySqlSyncDatabaseAction ignoreIncompatible(boolean ignoreIncompatible) {
this.ignoreIncompatible = ignoreIncompatible;
return this;
}
- public MySqlSyncDatabaseAction mergeShards(boolean mergeShards) {
- this.mergeShards = mergeShards;
- return this;
- }
-
- public MySqlSyncDatabaseAction withTablePrefix(@Nullable String tablePrefix) {
- if (tablePrefix != null) {
- this.tablePrefix = tablePrefix;
- }
- return this;
- }
-
- public MySqlSyncDatabaseAction withTableSuffix(@Nullable String tableSuffix) {
- if (tableSuffix != null) {
- this.tableSuffix = tableSuffix;
- }
- return this;
- }
-
- public MySqlSyncDatabaseAction includingTables(@Nullable String includingTables) {
- if (includingTables != null) {
- this.includingTables = includingTables;
- }
- return this;
- }
-
- public MySqlSyncDatabaseAction excludingTables(@Nullable String excludingTables) {
- this.excludingTables = excludingTables;
- return this;
- }
-
- public MySqlSyncDatabaseAction withMode(MultiTablesSinkMode mode) {
- this.mode = mode;
- return this;
- }
-
- public MySqlSyncDatabaseAction withTypeMapping(TypeMapping typeMapping) {
- this.typeMapping = typeMapping;
- return this;
- }
-
- public MySqlSyncDatabaseAction withMetadataKeys(List<String> metadataKeys) {
- this.metadataColumn = metadataKeys;
- return this;
+ @Override
+ protected Optional<CdcMetadataConverter<?>> metadataConverter(String column) {
+ return Optional.of(MySqlMetadataProcessor.converter(column));
}
@Override
- public void build() throws Exception {
+ protected void checkCdcSourceArgument() {
checkArgument(
- !mySqlConfig.contains(MySqlSourceOptions.TABLE_NAME),
+ !cdcSourceConfig.contains(MySqlSourceOptions.TABLE_NAME),
MySqlSourceOptions.TABLE_NAME.key()
+ " cannot be set for mysql-sync-database. "
+ "If you want to sync several MySQL tables into one Paimon table, "
+ "use mysql-sync-table instead.");
- boolean caseSensitive = catalog.caseSensitive();
-
- validateCaseInsensitive(caseSensitive);
+ }
+ @Override
+ protected DataStreamSource<String> buildSource() throws Exception {
+ boolean caseSensitive = catalog.caseSensitive();
Pattern includingPattern = Pattern.compile(includingTables);
Pattern excludingPattern =
excludingTables == null ? null : Pattern.compile(excludingTables);
MySqlSchemasInfo mySqlSchemasInfo =
MySqlActionUtils.getMySqlTableInfos(
- mySqlConfig,
+ cdcSourceConfig,
tableName ->
shouldMonitorTable(tableName, includingPattern, excludingPattern),
excludedTables,
@@ -218,21 +153,13 @@ public class MySqlSyncDatabaseAction extends ActionBase {
List<MySqlTableInfo> mySqlTableInfos = mySqlSchemasInfo.toMySqlTableInfos(mergeShards);
checkArgument(
- mySqlTableInfos.size() > 0,
+ !mySqlTableInfos.isEmpty(),
"No tables found in MySQL database "
- + mySqlConfig.get(MySqlSourceOptions.DATABASE_NAME)
+ + cdcSourceConfig.get(MySqlSourceOptions.DATABASE_NAME)
+ ", or MySQL database does not exist.");
- catalog.createDatabase(database, true);
TableNameConverter tableNameConverter =
new TableNameConverter(caseSensitive, mergeShards, tablePrefix, tableSuffix);
-
- CdcMetadataConverter<?>[] metadataConverters =
- metadataColumn.stream()
- .map(MySqlMetadataProcessor::converter)
- .toArray(CdcMetadataConverter[]::new);
-
- List<FileStoreTable> fileStoreTables = new ArrayList<>();
for (MySqlTableInfo tableInfo : mySqlTableInfos) {
Identifier identifier =
Identifier.create(
@@ -253,7 +180,7 @@ public class MySqlSyncDatabaseAction extends ActionBase {
Supplier<String> errMsg =
incompatibleMessage(table.schema(), tableInfo, identifier);
if (shouldMonitorTable(table.schema(), fromMySql, errMsg)) {
- fileStoreTables.add(table);
+ tables.add(table);
monitoredTables.addAll(tableInfo.identifiers());
} else {
excludedTables.addAll(tableInfo.identifiers());
@@ -261,7 +188,7 @@ public class MySqlSyncDatabaseAction extends ActionBase {
} catch (Catalog.TableNotExistException e) {
catalog.createTable(identifier, fromMySql, false);
table = (FileStoreTable) catalog.getTable(identifier);
- fileStoreTables.add(table);
+ tables.add(table);
monitoredTables.addAll(tableInfo.identifiers());
}
}
@@ -273,47 +200,31 @@ public class MySqlSyncDatabaseAction extends ActionBase {
MySqlSource<String> source =
MySqlActionUtils.buildMySqlSource(
- mySqlConfig,
+ cdcSourceConfig,
tableList(
mode,
- mySqlConfig.get(MySqlSourceOptions.DATABASE_NAME),
+ cdcSourceConfig.get(MySqlSourceOptions.DATABASE_NAME),
includingTables,
monitoredTables,
excludedTables));
+ return env.fromSource(source, WatermarkStrategy.noWatermarks(), sourceName());
+ }
- NewTableSchemaBuilder schemaBuilder = new NewTableSchemaBuilder(tableConfig, caseSensitive);
-
- TypeMapping typeMapping = this.typeMapping;
- MySqlRecordParser recordParser =
- new MySqlRecordParser(mySqlConfig, caseSensitive, typeMapping, metadataConverters);
- EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
- () ->
- new RichCdcMultiplexRecordEventParser(
- schemaBuilder,
- includingPattern,
- excludingPattern,
- tableNameConverter);
+ @Override
+ protected String sourceName() {
+ return "MySQL Source";
+ }
- String database = this.database;
- MultiTablesSinkMode mode = this.mode;
- new FlinkCdcSyncDatabaseSinkBuilder<RichCdcMultiplexRecord>()
- .withInput(
- env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL Source")
- .flatMap(recordParser)
- .name("Parse"))
- .withParserFactory(parserFactory)
- .withDatabase(database)
- .withCatalogLoader(catalogLoader())
- .withTables(fileStoreTables)
- .withMode(mode)
- .withTableOptions(tableConfig)
- .build();
+ @Override
+ protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
+ boolean caseSensitive = catalog.caseSensitive();
+ return new MySqlRecordParser(
+ cdcSourceConfig, caseSensitive, typeMapping, metadataConverters);
}
- private void validateCaseInsensitive(boolean caseSensitive) {
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database", database);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table prefix", tablePrefix);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table suffix", tableSuffix);
+ @Override
+ protected String jobName() {
+ return String.format("MySQL-Paimon Database Sync: %s", database);
}
private void logNonPkTables(List<Identifier> nonPkTables) {
@@ -376,19 +287,4 @@ public class MySqlSyncDatabaseAction extends ActionBase {
public List<Identifier> excludedTables() {
return excludedTables;
}
-
- @VisibleForTesting
- public Map<String, String> tableConfig() {
- return tableConfig;
- }
-
- // ------------------------------------------------------------------------
- // Flink run methods
- // ------------------------------------------------------------------------
-
- @Override
- public void run() throws Exception {
- build();
- execute(String.format("MySQL-Paimon Database Sync: %s", database));
- }
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java
index ef56f900e..30c4895bc 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java
@@ -60,8 +60,8 @@ public class MySqlSyncDatabaseActionFactory implements ActionFactory {
optionalConfigMap(params, CATALOG_CONF),
optionalConfigMap(params, MYSQL_CONF));
- action.withTableConfig(optionalConfigMap(params, TABLE_CONF))
- .ignoreIncompatible(Boolean.parseBoolean(params.get(IGNORE_INCOMPATIBLE)))
+ action.withTableConfig(optionalConfigMap(params, TABLE_CONF));
+ action.ignoreIncompatible(Boolean.parseBoolean(params.get(IGNORE_INCOMPATIBLE)))
.mergeShards(
!params.has(MERGE_SHARDS) || Boolean.parseBoolean(params.get(MERGE_SHARDS)))
.withTablePrefix(params.get(TABLE_PREFIX))
@@ -70,7 +70,7 @@ public class MySqlSyncDatabaseActionFactory implements ActionFactory {
.excludingTables(params.get(EXCLUDING_TABLES))
.withMode(MultiTablesSinkMode.fromString(params.get(MODE)));
if (params.has(METADATA_COLUMN)) {
- action.withMetadataKeys(Arrays.asList(params.get(METADATA_COLUMN).split(",")));
+ action.withMetadataColumns(Arrays.asList(params.get(METADATA_COLUMN).split(",")));
}
if (params.has(TYPE_MAPPING)) {