flink-jobs-quickstart

/* MySQL */
/* Create database */
CREATE DATABASE test;
/* Create table */
USE test;
CREATE TABLE test_table (
  ID int NOT NULL,
  NAME varchar(255),
  CREATE_TIME datetime,
  PRIMARY KEY (ID)
);
/* Insert records */
INSERT INTO test_table(ID, NAME, CREATE_TIME) VALUES (1, 'Flink Jobs', '2022-09-20 15:13:24');
INSERT INTO test_table(ID, NAME, CREATE_TIME) VALUES (2, 'Flink CDC', '2022-09-20 15:24:01');
INSERT INTO test_table(ID, NAME, CREATE_TIME) VALUES (3, 'MySQL', '2022-09-20 17:12:16');
INSERT INTO test_table(ID, NAME, CREATE_TIME) VALUES (4, 'StarRocks', '2022-09-20 17:12:55');

/* StarRocks */
/* Create database */
CREATE DATABASE test;
/* Create table */
USE test;
CREATE TABLE test_table (
  ID int(11) NOT NULL,
  NAME varchar(765),
  CREATE_TIME datetime,
  EVENT_TIMESTAMP datetime COMMENT "事件时间",
  ETL_TIMESTAMP datetime COMMENT "清洗时间"
)
PRIMARY KEY(ID)
DISTRIBUTED BY HASH(ID) BUCKETS 3;

<?xml version="1.0" encoding="UTF-8"?>
<flink-jobs xmlns="http://www.10mg.cn/schema/flink-jobs"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.10mg.cn/schema/flink-jobs http://www.10mg.cn/schema/flink-jobs.xsd">
	<!-- configuration的内容支持Flink SQL中WITH参数的语法，使用“,”来分隔两个配置；也支持properties语法（使用换行分隔两个配置，暂不支持配置的值带换行符） -->
	<!-- configuration的内容可以在使用客户端提交任务之前根据规则统一指定，这样就不需要每个任务都配置了 -->
	<configuration><![CDATA[pipeline.name=test]]></configuration>
	<!-- 配置文件中并没有配置名称为test的数据源，因此会根据自动数据源的配置自动生成数据源。-->
	<!-- 根据自动数据源的配置这个数据源实际上指向StarRocks的test数据库 -->
	<data-sync from="source" to="test" table="test_table">
		<!-- server-id 可以使用客户端提交任务之前统一指定的参数，然后使用 server-id=#serverId或 server-id=:serverId来引用 -->
		<from-config><![CDATA[server-id=5500]]></from-config>
	</data-sync>
</flink-jobs>