[FEATURE][TOOLS-CDC]:add flink Command line support #105

xxsc0529 · 2024-11-12T02:30:15Z

Summary

When synchronizing data to OceanBase, the flink CLI is supported. close #101

Solution Description

Define program entries, receive parameters, and parse them. Use Oceanbase Catalog to obtain the information required by Flink and create the corresponding sink。Here's how it works.
`CREATE TABLE test_history (
itemid BIGINT NOT NULL,
clock INT DEFAULT 0 NOT NULL,
value DOUBLE PRECISION DEFAULT 0.0000 NOT NULL,
ns INT DEFAULT 0 NOT NULL,
PRIMARY KEY (itemid, clock, ns)
) ;

CREATE TABLE test_history_uint (
itemid BIGINT NOT NULL,
clock INT DEFAULT 0 NOT NULL,
value DECIMAL(20,0) DEFAULT 0 NOT NULL,
ns INT DEFAULT 0 NOT NULL,
PRIMARY KEY (itemid, clock, ns)
) ;

CREATE TABLE test_history_str (
itemid BIGINT NOT NULL,
clock INT DEFAULT 0 NOT NULL,
value VARCHAR(255) DEFAULT '' NOT NULL,
ns INT DEFAULT 0 NOT NULL,
PRIMARY KEY (itemid, clock, ns)
);The command line is as follows./bin/flink run
-Dexecution.checkpointing.interval=10s
-Dparallelism.default=1
-c com.oceanbase.connector.flink.tools.cdc.CdcTools
lib/flink-connector-oceanbase-tools-cdc-1.4-SNAPSHOT.jar
mysql-sync-database
--database test_db
--mysql-conf hostname=xxxx
--mysql-conf port=3306
--mysql-conf username=root
--mysql-conf password=xxxx
--mysql-conf database-name=test_db
--including-tables "tbl1|test.*"
--sink-conf username=xxxx@xxxx
--sink-conf password=xxxx
--sink-conf url=jdbc:mysql://xxxx:xxxx
--sink-conf sink.label-prefix=label
--table-conf replication_num=1`
The sink-config configuration is the configuration of the OceanBase database

whhe · 2024-11-28T09:20:07Z

...ls-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java

+        }
+        sb.append(";");
+
+        System.out.println("Generated DDL: " + sb);


Please use logger but not System.out, check it in other files as well.

whhe · 2024-11-28T09:32:16Z

...-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java

+
+            return new DataChangeRecord(tableInfo, type, values);
+        } catch (IOException e) {
+            log.error("Failed to parse rowData JSON: {}", rowDataStr, e);


Why not throw the exception out?

whhe · 2024-11-28T09:34:40Z

...eanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/FieldSchema.java

+package com.oceanbase.connector.flink.tools.catalog;
+
+public class FieldSchema {
+    private String name;


In what scenarios will these fields be modified? How about make these fields final and remove all setters?

whhe · 2024-11-28T09:37:09Z

...ls-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java

+        sb.append(", ");
+    }
+
+    public static String identifier(String name) {


Better to use OceanBaseDialect.quoteIdentifier

whhe · 2024-11-28T09:41:24Z

...eanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/TableSchema.java

+import java.util.Map;
+
+public class TableSchema {
+    public static final String DORIS_TABLE_REGEX = "^[a-zA-Z][a-zA-Z0-9-_]*$";


Unrelated field.

whhe · 2024-11-28T09:41:42Z

...eanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/TableSchema.java

+    private List<String> distributeKeys = new ArrayList<>();
+    private Map<String, String> properties = new HashMap<>();
+
+    private Integer tableBuckets;


What is tableBuckets used for?

whhe · 2024-11-28T09:49:03Z

...-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/SourceSchema.java

+    public abstract String convertToOceanBaseType(
+            String fieldType, Integer precision, Integer scale);
+
+    public abstract String getCdcTableName();


Seems it is not necessary while there is already a method getTableIdentifier here.

whhe · 2024-11-28T09:50:03Z

...ctor-oceanbase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/CdcTools.java

+        boolean createTableOnly = params.has(DatabaseSyncConfig.CREATE_TABLE_ONLY);
+        boolean ignoreDefaultValue = params.has(DatabaseSyncConfig.IGNORE_DEFAULT_VALUE);
+        boolean ignoreIncompatible = params.has(DatabaseSyncConfig.IGNORE_INCOMPATIBLE);
+        boolean singleSink = params.has(DatabaseSyncConfig.SINGLE_SINK);


What is it used for?

whhe · 2024-11-28T09:51:36Z

...base-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/cdc/DatabaseSyncConfig.java

+
+public class DatabaseSyncConfig {
+
+    public static final String MYSQL_SYNC_DATABASE = "mysql-sync-database";


There should be a short description for every config key.

whhe · 2024-11-28T09:54:20Z

There should be at least a test case for mysql cdc, please add one on the github workflow.

yuanoOo

Thanks for your contribution, I left some comments.

yuanoOo · 2024-11-28T10:23:12Z

...-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java

+                return data -> Date.valueOf(LocalDate.ofEpochDay((int) data));
+            case TIME_WITHOUT_TIME_ZONE:
+                return data -> Time.valueOf(LocalTime.ofNanoOfDay((int) data * 1_000_000L));
+            case TIMESTAMP_WITHOUT_TIME_ZONE:


Why not support LocalZonedTimestampType, TimestampType?

yuanoOo · 2024-11-28T10:25:03Z

...-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java

+                return data -> ((TimestampData) data).toTimestamp();
+            case DECIMAL:
+                return data -> ((DecimalData) data).toBigDecimal();
+            case ARRAY:


Why not also support convert Map, Row type to string

yuanoOo · 2024-11-28T10:26:36Z

...-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java

+    }
+
+    @Override
+    protected SerializationRuntimeConverter createNotNullConverter(LogicalType type) {


We'd better add a test for this method, covering all types.

yuanoOo · 2024-11-29T03:07:01Z

...-cdc/src/main/java/com/oceanbase/connector/flink/table/OceanBaseJsonSerializationSchema.java

+public class OceanBaseJsonSerializationSchema extends AbstractRecordSerializationSchema<String> {
+
+    private static final long serialVersionUID = 1L;
+    private static final Logger log =


It is best to use uppercase letters to be consistent with other places.

yuanoOo · 2024-11-29T03:13:50Z

...ls-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java

+        return tableSchema;
+    }
+
+    private static List<String> buildDistributeKeys(


What does DistributeKeys mean in oceanbase, oceanbase only has index keys and partition keys.

yuanoOo · 2024-11-29T03:14:41Z

...ls-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSchemaFactory.java

+            buildColumn(sb, field, keys.contains(entry.getKey()));
+        }
+
+        sb = sb.deleteCharAt(sb.length() - 1); // 删除最后一个逗号


Do not use Chinese.

yuanoOo · 2024-11-29T03:16:26Z

...ase-tools-cdc/src/main/java/com/oceanbase/connector/flink/tools/catalog/OceanBaseSystem.java

+
+/** OceanBase System Operate. */
+@Public
+public class OceanBaseSystem implements Serializable {


Better to change the name, it's confusing.

yuanoOo · 2024-12-02T06:25:00Z