<!-- https://mvnrepository.com/artifact/cn.tenmg/flink-jobs-launcher -->
<dependency>
    <groupId>cn.tenmg</groupId>
    <artifactId>flink-jobs-launcher</artifactId>
    <version>${flink-jobs-launcher.version}</version>
</dependency>

# RPC configuration
jobmanager.rpc.address= 192.168.100.27
jobmanager.rpc.port=6123
# Рекомендуемый, но необязательный путь к файлу JAR, который flink-jobs-launcher отправляет на выполнение.
flink.jobs.default.jar=/yourpath/your-flink-jobs-app-1.0.0.jar
# Необязательный класс, который flink-jobs-launcher отправляет для выполнения. Вы также можете указать основной класс в файле JAR.
#flink.jobs.default.class=yourpackage.App

FlinkJobs flinkJobs = XMLConfigLoader.getInstance().load(ClassUtils.getDefaultClassLoader().getResourceAsStream("flink-jobs.xml"));
RestClusterClientFlinkJobsLauncher launcher = new RestClusterClientFlinkJobsLauncher();
FlinkJobsInfo flinkJobsInfo = launcher.launch(flinkJobs);
System.out.println("Flink job launched: " + JSON.toJSONString(flinkJobsInfo));// Запуск задания flink-jobs

FlinkJobs flinkJobs = XMLConfigLoader.getInstance()
    .load("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\r\n" + 
        "<flink-jobs xmlns=\"http://www.10mg.cn/schema/flink-jobs\"\r\n" + 
        "   xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"\r\n" + 
        "   xsi:schemaLocation=\"http://www.10mg.cn/schema/flink-jobs http://www.10mg.cn/schema/flink-jobs.xsd\"\r\n" + 
        "   jar=\"/opt/flink-jobs/flink-jobs-quickstart-1.1.4.jar\" serviceName=\"HelloWorldService\">\r\n" + 
            "</flink-jobs>");
RestClusterClientFlinkJobsLauncher launcher = new RestClusterClientFlinkJobsLauncher();
System.out.println("Flink job launched: " + JSON.toJSONString(flinkJobsInfo));// Запуск задания flink-jobs

System.out.println("Задание flink-jobs с идентификатором задания: " + flinkJobsInfo.getJobId() + " остановлено, путь сохранения: "
                + launcher.stop(flinkJobsInfo.getJobId()));// Остановка задания flink-jobs

<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">
</flink-jobs>

<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"
    jar="D:\Programs\flink-1.8.3\examples\batch\WordCount.jar">
</flink-jobs>

<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"
    jar="/yourPath/yourJar.jar" serviceName="yourServiceName">
</flink-jobs>

<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"
    jar="/yourPath/yourJar.jar">
    <!--Параметры задачи, некоторые общие параметры также могут быть указаны перед вызовом Java API, например системное время и т. д. -->
    <params>
        <param name="beginDate">2021-01-01</param>
        <param name="endDate">2021-07-01</param>
    </params>

    <!-- Использование источника данных с именем hivedb для создания каталога с именем hive -->
    <execute-sql dataSource="hivedb">
        <![CDATA[
            create catalog hive
        ]]>
    </execute-sql>
    <!--Загрузка модуля hive -->
    <execute-sql>
        <![CDATA[
            load module hive
        ]]>
    </execute-sql>
    <!--Использование модулей hive,core -->
    <execute-sql>
        <![CDATA[
            use modules hive,core
        ]]>
    </execute-sql>
    <!-- Использовать источник данных с именем pgdb для создания таблицы order_stats_daily (если имя исходной таблицы и указанное имя таблицы построения не совпадают, вы можете указать WITH ('table-name' = 'actrual_table_name') для указания) -->
    <execute-sql dataSource="pgdb">
        <![CDATA[
            CREATE TABLE order_stats_daily (
              stats_date DATE,
              `count` BIGINT,
              PRIMARY KEY (stats_date) NOT ENFORCED
            ) WITH ('sink.buffer-flush.max-rows' = '0')
        ]]>
    </execute-sql>
    <!-- Используйте каталог hive для запроса и сохраните результаты как временную таблицу tmp в default_catalog по умолчанию -->
    <sql-query saveAs="tmp" catalog="hive">
        <![CDATA[
            select cast(to_date(o.business_date) as date) stats_date, count(*) `count` from odc_order_info_par o where o.business_date >= :beginDate and o.business_date < :endDate group by cast(to_date(o.business_date) as date)
        ]]>
    </sql-query>
    <!-- Удалить существующую статистику order_stats_daily данных (FLINK SQL не поддерживает DELETE, здесь выполняется JDBC)-->
    <execute-sql dataSource="pgdb">
        <![CDATA[
            delete from order_stats_daily where stats_date >= :beginDate and stats_date < :endDate
        ]]>
    </execute-sql>
    <!-- Вставка данных. На самом деле Flink в конечном итоге выполнит Upsert синтаксис -->
    <execute-sql>
        <![CDATA[
            INSERT INTO order_stats_daily(stats_date,`count`) SELECT stats_date, `count` FROM tmp
        ]]> **Запуск потоковой обработки SQL**

Ниже приведён XML-файл конфигурации задачи для синхронизации гетерогенных баз данных с помощью Debezium:

#### Запуск задачи синхронизации данных

Ниже представлен файл конфигурации XML для выполнения задачи синхронизации разнородных баз данных с помощью Debezium:

```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-1.1.2.xsd">
    <data-sync table="od_order_info" to="data_skyline"
        from="kafka" topic="testdb.testdb.od_order_info">
        <!-- В случае, когда структура таблиц в источнике и целевой базе данных одинакова (имена и типы полей совпадают), интеллектуальный режим позволяет автоматически получать метаданные таблицы из целевой базы данных, что упрощает процесс синхронизации данных. -->
        <!-- Если структура таблиц в источнике и целевой базе данных отличается (например, различаются имена или типы полей), необходимо настроить информацию о различиях столбцов вручную, например, указать исходный тип и функцию преобразования: -->
        <column fromName="UPDATE_TIME" fromType="BIGINT">TO_TIMESTAMP(FROM_UNIXTIME(UPDATE_TIME/1000, 'yyyy-MM-dd HH:mm:ss'))</column>
        <!-- Кроме того, если интеллектуальный режим отключен, необходимо предоставить подробную информацию обо всех столбцах. -->
    </data-sync>
</flink-jobs>