自动同步整个 MySQL/Oracle 数据库以进行数据分析

<dependency><groupId>org.apache.doris</groupId><artifactId>flink-doris-connector-1.15</artifactId><!--artifactId>flink-doris-connector-1.16</artifactId--><!--artifactId>flink-doris-connector-1.17</artifactId--><version>1.4.0</version></dependency>

<FLINK_HOME>/bin/flink run \-Dexecution.checkpointing.interval=10s \-Dparallelism.default=1 \-c org.apache.doris.flink.tools.cdc.CdcTools \lib/flink-doris-connector-1.16-1.4.0.jar \mysql-sync-database \--database test_db \--mysql-conf hostname=127.0.0.1 \--mysql-conf username=root \--mysql-conf password=123456 \--mysql-conf database-name=mysql_db \--including-tables "tbl|test.*" \--sink-conf fenodes=127.0.0.1:8030 \--sink-conf username=root \--sink-conf password=123456 \--sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \--sink-conf sink.label-prefix=label1 \--table-conf replication_num=1

CREATE TABLE flink_doris_source (
   name STRING,
   age INT,
   score DECIMAL(5,2)
  ) 
   WITH (
     'connector' = 'doris',
     'fenodes' = '127.0.0.1:8030',
     'table.identifier' = 'database.table',
     'username' = 'root',
     'password' = 'password',
     'doris.filter.query' = 'age=18'
);

SELECT * FROM flink_doris_source;

CREATE TABLE fact_table (
 `id` BIGINT,
 `name` STRING,
 `city` STRING,
 `process_time` as proctime()
) WITH (
 'connector' = 'kafka',
...
);

create table dim_city(
 `city` STRING,
 `level` INT ,
 `province` STRING,
 `country` STRING
) WITH (
 'connector' = 'doris',
 'fenodes' = '127.0.0.1:8030',
 'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
 'lookup.jdbc.async' = 'true',
 'table.identifier' = 'dim.dim_city',
 'username' = 'root',
 'password' = ''
);

SELECT a.id, a.name, a.city, c.province, c.country,c.level 
FROM fact_table a
LEFT JOIN dim_city FOR SYSTEM_TIME AS OF a.process_time AS c
ON a.city = c.city

CREATE TABLE doris_sink (   name STRING,   age INT,   score DECIMAL(5,2)  )    WITH (     'connector' = 'doris',     'fenodes' = '127.0.0.1:8030',     'table.identifier' = 'database.table',     'username' = 'root',     'password' = '',     'sink.label-prefix' = 'doris_label',     //json write in     'sink.properties.format' = 'json',     'sink.properties.read_json_by_line' = 'true');