<properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <flink.version>1.13.1</flink.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>com.starrocks</groupId>
            <artifactId>flink-connector-starrocks</artifactId>
            <!-- for flink-1.13 -->
            <version>1.2.3_flink-1.13_2.11</version>
<!--            <version>flink-connector-starrocks-1.2.3_flink-1.13_2.12</version>-->
        </dependency>
<!--        <dependency>-->
<!--            <groupId>org.apache.flink</groupId>-->
<!--            <artifactId>flink-table-planner-blink_2.11</artifactId>-->
<!--            <version>${flink.version}</version>-->
<!--        </dependency>-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-sql-connector-kafka_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.20</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.3.0</version>
        </dependency>
        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-format-changelog-json</artifactId>
            <version>2.0.2</version>
        </dependency>
        <dependency>
            <groupId>com.ververica</groupId>
            <!-- add the dependency matching your database -->
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>2.0.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-sql-connector-elasticsearch7_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.poi</groupId>
            <artifactId>poi-ooxml</artifactId>
            <version>5.0.0</version>
        </dependency>
        <dependency>
            <groupId>com.github.houbb</groupId>
            <artifactId>pinyin</artifactId>
            <version>0.2.1</version>
        </dependency>
    </dependencies>
    <build>
        <finalName>${project.artifactId}.${git.branch}.${project.version}.${build.version}</finalName>
        <resources>
            <resource>
                <directory>src/main/resources</directory>
                <filtering>true</filtering>
                <excludes>
                    <exclude>static/**</exclude>
                </excludes>
                <includes>
                    <include>mybatis/**</include>
                    <include>bin/*.sh</include>
                    <include>*.properties</include>
                    <include>*.xml</include>
                    <include>*.json</include>
                </includes>
            </resource>
            <!-- fonts file cannot use filter as the data structure of byte file will
                be changed via filter -->
            <resource>
                <directory>src/main/resources</directory>
                <filtering>false</filtering>
                <includes>
                    <include>static/**</include>
                    <include>templates/**</include>
                </includes>
            </resource>
        </resources>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <configuration>
                    <skip>true</skip>
                    <testFailureIgnore>true</testFailureIgnore>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.iflytek.swk.MysqlTransElastic</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <descriptors>
                        <descriptor>src/assembly/distribution.xml</descriptor>
                    </descriptors>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-source-plugin</artifactId>
                <executions>
                    <execution>
                        <id>attach-sources</id>
                        <phase>package</phase>
                        <goals>
                            <goal>jar</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

 public static void main(String[] args) throws Exception {
        // 加载外部配置
        List<Props> propsList = new ArrayList<>();

        for (String arg : args) {
            if (FileUtil.isFile(arg)) {
                Props props = Props.create();
                props.load(FileUtil.getUtf8Reader(arg));
                propsList.add(props);
            } else if (FileUtil.isDirectory(arg)) {
                File dir = FileUtil.file(arg);
                for (File file : Objects.requireNonNull(dir.listFiles())) {
                    if (!StrUtil.equals(FileUtil.extName(file), "properties")) {
                        continue;
                    }
                    Props props = Props.create();
                    props.load(FileUtil.getUtf8Reader(file));
                    propsList.add(props);
                }
            } else {
                throw new FileNotFoundException("找不到配置文件");
            }
        }


        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings Settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env, Settings);
        env.getConfig().enableObjectReuse();//启用对象重用
        //1.1Checkpoint相关
        /*读取的是binlog中的数据，如果集群挂掉，尽量能实现断点续传功能。如果从最新的读取（丢数据）。如果从最开始读（重复数据）。理想状态：读取binlog中的数据读一行，保存一次读取到的（读取到的行）位置信息。而flink中读取行位置信息保存在Checkpoint中。使用Checkpoint可以把flink中读取（按行）的位置信息保存在Checkpoint中*/
        env.enableCheckpointing(60000L);//1m执行一次Checkpoint
        //设置Checkpoint的模式：精准一次
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //任务挂掉的时候是否清理checkpoint。使任务正常退出时不删除CK内容，有助于任务恢复。默认的是取消的时候清空checkpoint中的数据。RETAIN_ON_CANCELLATION表示取消任务的时候，保存最后一次的checkpoint。便于任务的重启和恢复，正常情况下都使用RETAIN
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置一个重启策略：默认的固定延时重启次数，重启的次数是Integer的最大值，重启的间隔是1s
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(64, 10000L));
        //join表state较大，需要使用rocksdb模式处理大的state
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        //设置并行数
        env.setParallelism(2);
        //设置一个状态后端 jobManager。如果使用的yarn集群模式，jobManager随着任务的生成而生成，任务挂了jobManager就没了。因此需要启动一个状态后端。只要设置checkpoint，尽量就设置一个状态后端。保存在各个节点都能读取的位置：hdfs中
        Configuration configuration = tenv.getConfig().getConfiguration();
        configuration.setString("table.exec.mini-batch.enabled", "true");
        configuration.setString("table.exec.mini-batch.allow-latency", "10s");
        configuration.setString("table.exec.mini-batch.size", "1000");
        configuration.setString("table.exec.hive.infer-source-parallelism.max", "100");
        configuration.setString("table.exec.hive.fallback-mapred-reader", "true");

        for (Props props : propsList) {
            String source = props.getStr("source");
            String sink = props.getStr("sink");
            // 执行语句
            if (StrUtil.isNotBlank(source)) {
                tenv.executeSql(source);
            }
            if (StrUtil.isNotBlank(sink)) {
                tenv.executeSql(sink);
            }
        }
        for (Props props : propsList) {
            String existSource = props.getStr("source[0]");
            if (StringUtils.isNotBlank(existSource)) {
                Map<String, List<String>> map = new HashMap<>();
                props.forEach((k, v) -> {
                    String key = StringUtils.split(k.toString(), "[")[0];
                    List<String> val = map.getOrDefault(key, new ArrayList<>());
                    val.add(StrUtil.toString(v));
                    map.putIfAbsent(key, val);
                });

                List<String> sourceList = map.get("source");
                // 执行语句
                for (String source : sourceList) {
                    tenv.executeSql(source);
                }


                List<String> sinkList = map.get("sink");
                // 执行语句
                for (String sink : sinkList) {
                    tenv.executeSql(sink);
                }
            }

        }
        // 数据视图
        for (Props props : propsList) {
            String view = props.getStr("view");
            if (StrUtil.isNotBlank(view)) {
                tenv.executeSql(view);
            }
        }
        // 数据转移sql
        for (Props props : propsList) {
            String sqlcdc = props.getStr("trans");
            if (StrUtil.isNotBlank(sqlcdc)) {
                tenv.executeSql(sqlcdc);
            }
        }

        //env.execute();

    }