需求描述：

1、数据从 Kafka 写入 ClickHouse。

2、相关配置存放于 Mysql 中，通过 Mysql 进行动态读取。

3、此案例中的 Kafka 是进行了 Kerberos 安全认证的，如果不需要自行修改。

4、先在 ClickHouse 中创建表然后动态获取 ClickHouse 的表结构。

5、Kafka 数据为 Json 格式，通过 FlatMap 扁平化处理后，根据表结构封装到 Row 中后完成写入。

6、写入时转换成临时视图模式，利用 Flink-Sql 实现数据写入。

7、本地测试时可以编辑 resources.flink_backup_local.yml 通过 ConfigTools.initConf 方法获取配置。

1）导入相关依赖

这里的依赖比较冗余，大家可以根据各自需求做删除或保留。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>gaei.cn.x5l</groupId>
    <artifactId>kafka2ch</artifactId>
    <version>1.0.0</version>

    <properties>
        <hbase.version>2.3.3</hbase.version>
        <hadoop.version>3.1.1</hadoop.version>
        <spark.version>3.0.2</spark.version>
        <scala.version>2.12.10</scala.version>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.14.0</flink.version>
        <scala.binary.version>2.12</scala.binary.version>
        <target.java.version>1.8</target.java.version>
        <maven.compiler.source>${target.java.version}</maven.compiler.source>
        <maven.compiler.target>${target.java.version}</maven.compiler.target>
        <log4j.version>2.17.2</log4j.version>
        <hadoop.version>3.1.2</hadoop.version>
        <hive.version>3.1.2</hive.version>
    </properties>

    <dependencies>
        <!-- 基础依赖  开始-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <!-- 基础依赖  结束-->
        <!-- TABLE  开始-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
            <version>1.14.0</version>
            <scope>provided</scope>
        </dependency>


        <!-- 使用 hive sql时注销，其他时候可以放开 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-cep_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- TABLE  结束-->
        <!-- sql  开始-->
        <!-- sql解析 开始 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <!-- sql解析 结束 -->
        <!-- sql连接 kafka -->
<!--        <dependency>-->
<!--            <groupId>org.apache.flink</groupId>-->
<!--            <artifactId>flink-sql-connector-kafka_${scala.binary.version}</artifactId>-->
<!--            <version>${flink.version}</version>-->
<!--        </dependency>-->
        <!-- sql  结束-->
        <!-- 检查点 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-state-processor-api_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <!-- 有状态的函数依赖 开始 -->
        <!--        <dependency>-->
        <!--            <groupId>org.apache.flink</groupId>-->
        <!--            <artifactId>statefun-sdk-java</artifactId>-->
        <!--            <version>3.0.0</version>-->
        <!--        </dependency>-->
        <!-- 有状态的函数依赖 结束 -->
        <!-- 连接Kafka -->
<!--        <dependency>-->
<!--            <groupId>org.apache.flink</groupId>-->
<!--            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>-->
<!--            <version>${flink.version}</version>-->
<!--        </dependency>-->
        <dependency>
            <groupId>commons-lang</groupId>
            <artifactId>commons-lang</artifactId>
            <version>2.5</version>
            <scope>compile</scope>
        </dependency>
        <!-- DataStream 开始 -->
        <!--        <dependency>-->
        <!--            <groupId>org.apache.flink</groupId>-->
        <!--            <artifactId>statefun-flink-datastream</artifactId>-->
        <!--            <version>3.0.0</version>-->
        <!--        </dependency>-->
        <!-- DataStream 结束 -->
        <!-- 本地监控任务 开始 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <!-- 本地监控任务 结束 -->
        <!-- DataStream 开始 -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>${log4j.version}</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>${log4j.version}</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>${log4j.version}</version>
            <scope>runtime</scope>
        </dependency>
        <!-- hdfs -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.3.1</version>

            <!--            <exclusions>-->
            <!--                <exclusion>-->
            <!--                    <groupId>org.apache.curator</groupId>-->
            <!--                    <artifactId>curator-client</artifactId>-->
            <!--                </exclusion>-->
            <!--            </exclusions>-->

        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.curator/curator-client -->
        <!--        <dependency>-->
        <!--            <groupId>org.apache.curator</groupId>-->
        <!--            <artifactId>curator-client</artifactId>-->
        <!--            <version>5.3.0</version>-->
        <!--        </dependency>-->

        <!-- 重点，容易被忽略的jar -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-auth</artifactId>
            <version>${hadoop.version}</version>


        </dependency>
        <!-- rocksdb_2 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- 其他 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.1.23</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.18</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>gaei.cn.x5l.bigdata.common</groupId>
            <artifactId>x5l-bigdata-common</artifactId>
            <version>1.3-SNAPSHOT</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-core</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-api</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-slf4j-impl</artifactId>
                </exclusion>
            </exclusions>
        </dependency>


        <!--        <dependency>-->
        <!--            <groupId>org.apache.flink</groupId>-->
        <!--            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>-->
        <!--            <version>${flink.version}</version>-->
        <!--        </dependency>-->
        <!--   将  flink-connector-kafka_2.12  改为  flink-sql-connector-kafka_2.12 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-sql-connector-kafka_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>


        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-clickhouse</artifactId>
            <version>1.14.3-SNAPSHOT</version>
            <!--<systemPath>${project.basedir}/lib/flink-connector-clickhouse-1.12.0-SNAPSHOT.jar</systemPath>-->
            <!--<scope>system</scope>-->
        </dependency>

        <dependency>
            <groupId>gaei.cn.x5l</groupId>
            <artifactId>tsp-gb-decode</artifactId>
            <version>1.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.jyaml</groupId>
            <artifactId>jyaml</artifactId>
            <version>1.3</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.44</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>gaei.cn.x5l.flink.common</groupId>
            <artifactId>x5l-flink-common</artifactId>
            <version>1.4-SNAPSHOT</version>
        </dependency>


    </dependencies>

    <build>
        <plugins>

            <!-- Java Compiler -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>${target.java.version}</source>
                    <target>${target.java.version}</target>
                </configuration>
            </plugin>

            <!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
            <!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.0.0</version>
                <executions>
                    <!-- Run shade goal on package phase -->
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>org.apache.flink:force-shading</exclude>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>org.apache.logging.log4j:*</exclude>
                                    <exclude>org.apache.flink:flink-runtime-web_2.11</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the META-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.owp.flink.kafka.KafkaSourceDemo</mainClass>
                                </transformer>
                                <!-- flink sql 需要  -->
                                <!-- The service transformer is needed to merge META-INF/services files -->
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                <!-- ... -->
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>

        <pluginManagement>
            <plugins>
                <!-- This improves the out-of-the-box experience in Eclipse by resolving some warnings. -->
                <plugin>
                    <groupId>org.eclipse.m2e</groupId>
                    <artifactId>lifecycle-mapping</artifactId>
                    <version>1.0.0</version>
                    <configuration>
                        <lifecycleMappingMetadata>
                            <pluginExecutions>
                                <pluginExecution>
                                    <pluginExecutionFilter>
                                        <groupId>org.apache.maven.plugins</groupId>
                                        <artifactId>maven-shade-plugin</artifactId>
                                        <versionRange>[3.0.0,)</versionRange>
                                        <goals>
                                            <goal>shade</goal>
                                        </goals>
                                    </pluginExecutionFilter>
                                    <action>
                                        <ignore/>
                                    </action>
                                </pluginExecution>
                                <pluginExecution>
                                    <pluginExecutionFilter>
                                        <groupId>org.apache.maven.plugins</groupId>
                                        <artifactId>maven-compiler-plugin</artifactId>
                                        <versionRange>[3.1,)</versionRange>
                                        <goals>
                                            <goal>testCompile</goal>
                                            <goal>compile</goal>
                                        </goals>
                                    </pluginExecutionFilter>
                                    <action>
                                        <ignore/>
                                    </action>
                                </pluginExecution>
                            </pluginExecutions>
                        </lifecycleMappingMetadata>
                    </configuration>
                </plugin>

            </plugins>
        </pluginManagement>

    </build>

</project>

2）代码实现

2.1.resources

2.1.1.appconfig.yml

mysql.url: "jdbc:mysql://1.1.1.1:3306/test?useSSL=false&useUnicode=true&characterEncoding=UTF8&connectTimeout=60000&socketTimeout=60000"
mysql.username: "test"
mysql.password: "123456"
mysql.driver: "com.mysql.jdbc.Driver"

2.1.2.log4j.properties

log4j.rootLogger=info, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

2.1.3.log4j2.xml

<?xml version="1.0" encoding="UTF-8"?>
<configuration monitorInterval="5">
    <Properties>
        <property name="LOG_PATTERN" value="%date{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n" />
        <property name="LOG_LEVEL" value="ERROR" />
    </Properties>

    <appenders>
        <console name="console" target="SYSTEM_OUT">
            <PatternLayout pattern="${LOG_PATTERN}"/>
            <ThresholdFilter level="${LOG_LEVEL}" onMatch="ACCEPT" onMismatch="DENY"/>
        </console>
        <File name="log" fileName="tmp/log/job.log" append="false">
            <PatternLayout pattern="%d{HH:mm:ss.SSS} %-5level %class{36} %L %M - %msg%xEx%n"/>
        </File>
    </appenders>

    <loggers>
        <root level="${LOG_LEVEL}">
            <appender-ref ref="console"/>
            <appender-ref ref="log"/>
        </root>
    </loggers>
</configuration>

2.1.4.flink_backup_local.yml

clickhouse:
  connector: 'clickhouse'
  database-name: 'dwd'
  driver: 'ru.yandex.clickhouse.ClickHouseDriver'
  jdbcurl: 'jdbc:clickhouse://10.1.1.1:8123/dwd?socket_timeout=480000'
  password: 'X8v@123456!%$'
  reissueInterval: 3
  sink.batch-size: '200000'
  sink.flush-interval: '3000000'
  sink.ignore-delete: 'true'
  sink.max-retries: '3'
  sink.partition-key: 'toYYYYMMDD(sample_date_time)'
  sink.partition-strategy: 'balanced'
  table-name: 'test_local'
  url: 'clickhouse://10.1.1.1:8123,10.1.1.1:8123,10.1.1.1:8123,10.1.1.1:8123,10.1.1.1:8123'
  username: 'test'
hdfs:
  checkPointPath: 'hdfs://nameserver/user/flink/rocksdbcheckpoint'
  checkpointTimeout: 360000
  checkpointing: 300000
  maxConcurrentCheckpoints: 1
  minPauseBetweenCheckpoints: 10000
  restartInterval: 60
  restartStrategy: 3
kafka-consumer:
  prop:
    auto.offset.reset: 'earliest'
    bootstrap.servers: 'kfk01:9092,kfk02:9092,kfk03:9092'
    enable.auto.commit: 'false'
    fetch.max.bytes: '52428700'
    group.id: 'test'
    isKerberized: '1'
    keytab: 'D:/keytab/test.keytab'
    krb5Conf: 'D:/keytab/krb5.conf'
    max.poll.interval.ms: '300000'
    max.poll.records: '1000'
    principal: 'test@PRE.TEST.COM'
    security_protocol: 'SASL_PLAINTEXT'
    serviceName: 'kafka'
    session.timeout.ms: '600000'
    useTicketCache: 'false'
  topics: 'topicA,topicB'
kafka-producer:
  defaultTopic: 'kafka2hive_error'
  prop:
    acks: 'all'
    batch.size: '1048576'
    bootstrap.servers: 'kfk01:9092,kfk02:9092,kfk03:9092'
    compression.type: 'lz4'
    key.serializer: 'org.apache.kafka.common.serialization.StringSerializer'
    retries: '3'
    value.serializer: 'org.apache.kafka.common.serialization.StringSerializer'

2.2.utils

2.2.1.DBConn

import java.sql.*;

public class DBConn {


    private static final String driver = "com.mysql.jdbc.Driver";		//mysql驱动
    private static Connection conn = null;

    private static PreparedStatement ps = null;
    private static ResultSet rs = null;
    private static final CallableStatement cs = null;

    /**
     * 连接数据库
     * @return
     */
    public static Connection conn(String url,String username,String password) {
        Connection conn = null;
        try {
            Class.forName(driver);  //加载数据库驱动
            try {
                conn = DriverManager.getConnection(url, username, password);  //连接数据库
            } catch (SQLException e) {
                e.printStackTrace();
            }
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return conn;
    }

    /**
     * 关闭数据库链接
     * @return
     */
    public static void close() {
        if(conn != null) {
            try {
                conn.close();  //关闭数据库链接
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}

2.2.2.CommonUtils

@Slf4j
public class CommonUtils {
    public static StreamExecutionEnvironment setCheckpoint(StreamExecutionEnvironment env) throws IOException {
//        ConfigTools.initConf("local");
        Map hdfsMap = (Map) ConfigTools.mapConf.get("hdfs");
        env.enableCheckpointing(((Integer) hdfsMap.get("checkpointing")).longValue(), CheckpointingMode.EXACTLY_ONCE);//这里会造成offset提交的延迟
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(((Integer) hdfsMap.get("minPauseBetweenCheckpoints")).longValue());
        env.getCheckpointConfig().setCheckpointTimeout(((Integer) hdfsMap.get("checkpointTimeout")).longValue());
        env.getCheckpointConfig().setMaxConcurrentCheckpoints((Integer) hdfsMap.get("maxConcurrentCheckpoints"));
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                (Integer) hdfsMap.get("restartStrategy"), // 尝试重启的次数，不宜过小，分布式任务很容易出问题（正常情况），建议3-5次
                Time.of(((Integer) hdfsMap.get("restartInterval")).longValue(), TimeUnit.SECONDS) // 延时
        ));
        //设置可容忍的检查点失败数，默认值为0表示不允许容忍任何检查点失败
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);

        //设置状态后端存储方式
        env.setStateBackend(new RocksDBStateBackend((String) hdfsMap.get("checkPointPath"), true));
//        env.setStateBackend(new FsStateBackend((String) hdfsMap.get("checkPointPath"), true));
//        env.setStateBackend(new HashMapStateBackend(());
        return env;
    }

    public static FlinkKafkaConsumer<ConsumerRecord<String, String>> getKafkaConsumer(Map<String, Object> kafkaConf) throws IOException {
        String[] topics = ((String) kafkaConf.get("topics")).split(",");
        log.info("监听的topic: {}", topics);
        Properties properties = new Properties();
        Map<String, String> kafkaProp = (Map<String, String>) kafkaConf.get("prop");
        for (String key : kafkaProp.keySet()) {
            properties.setProperty(key, kafkaProp.get(key).toString());
        }

        if (!StringUtils.isBlank((String) kafkaProp.get("isKerberized")) && "1".equals(kafkaProp.get("isKerberized"))) {
            System.setProperty("java.security.krb5.conf", kafkaProp.get("krb5Conf"));
            properties.put("security.protocol", kafkaProp.get("security_protocol"));
            properties.put("sasl.jaas.config", "com.sun.security.auth.module.Krb5LoginModule required "
                    + "useTicketCache=" + kafkaProp.get("useTicketCache") + " "

                    + "serviceName=\"" + kafkaProp.get("serviceName") + "\" "
                    + "useKeyTab=true "
                    + "keyTab=\"" + kafkaProp.get("keytab").toString() + "\" "
                    + "principal=\"" + kafkaProp.get("principal").toString() + "\";");
        }

        properties.put("key.serializer", "org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArrayDeserializer");
        properties.put("value.serializer", "org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.ByteArrayDeserializer");

        FlinkKafkaConsumer<ConsumerRecord<String, String>> consumerRecordFlinkKafkaConsumer = new FlinkKafkaConsumer<ConsumerRecord<String, String>>(Arrays.asList(topics), new KafkaDeserializationSchema<ConsumerRecord<String, String>>() {
            @Override
            public TypeInformation<ConsumerRecord<String, String>> getProducedType() {
                return TypeInformation.of(new TypeHint<ConsumerRecord<String, String>>() {
                });
            }

            @Override
            public boolean isEndOfStream(ConsumerRecord<String, String> stringStringConsumerRecord) {
                return false;
            }

            @Override
            public ConsumerRecord<String, String> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                return new ConsumerRecord<String, String>(
                        record.topic(),
                        record.partition(),
                        record.offset(),
                        record.timestamp(),
                        record.timestampType(),
                        record.checksum(),
                        record.serializedKeySize(),
                        record.serializedValueSize(),
                        new String(record.key() == null ? "".getBytes(StandardCharsets.UTF_8) : record.key(), StandardCharsets.UTF_8),
                        new String(record.value() == null ? "{}".getBytes(StandardCharsets.UTF_8) : record.value(), StandardCharsets.UTF_8));
            }
        }, properties);
        return consumerRecordFlinkKafkaConsumer;
    }
}

2.2.3.RemoteConfigUtil

public class RemoteConfigUtil {
    private static final Logger log = LoggerFactory.getLogger(RemoteConfigUtil.class);
    private static Connection conn = null;
    private static PreparedStatement ps = null;
    private static ResultSet rs = null;
    public static Map<String, Object> mapConf;

    public RemoteConfigUtil() {
    }

    public static Map<String, Object> getByAppNameAndConfigName(String appName, String ConfigName) throws SQLException {
        if (mapConf != null && mapConf.size() > 0) {
            return mapConf;
        } else {
            Map<String, String> ymlMap = LocalConfigUtil.getYmlMap("/appconfig");
            String username = (String)ymlMap.get("mysql.username");
            String password = (String)ymlMap.get("mysql.password");
            String url = (String)ymlMap.get("mysql.url");
            String driver = (String)ymlMap.get("mysql.driver");
            Connection conn = JdbcUtil.getConnection(url, username, password, driver);
            PreparedStatement preparedStatement = null;

            Map var14;
            try {
                String sql = "select config_context from base_app_config where app_name = '%s' and config_name = '%s'";
                preparedStatement = conn.prepareStatement(String.format(sql, appName, ConfigName));
                ResultSet rs = preparedStatement.executeQuery();

                String config_context;
                for(config_context = ""; rs.next(); config_context = rs.getString("config_context")) {
                }
                rs.close();
                log.info("配置信息config_context: {}", config_context);
                if (StringUtils.isNotBlank(config_context)) {
                    System.out.println(JSONObject.toJSONString(JSONObject.parseObject(config_context), new SerializerFeature[]{SerializerFeature.PrettyFormat}));
                }
                mapConf = (Map)JSON.parseObject(config_context, Map.class);
                var14 = mapConf;
            } finally {
                if (preparedStatement != null) {
                    preparedStatement.close();
                }

                if (conn != null) {
                    conn.close();
                }
            }

            return var14;
        }
    }
}

2.2.4.ClickhouseUtil

public class ClickhouseUtil {
    public ClickhouseUtil() {
    }

    public static List<SchemaPo> getSchemaPoList(Map<String, Object> chMapConf) throws SQLException {
        List schemaPos = new ArrayList();
        Connection connection = null;

        try {
            String jdbcurl = (String) chMapConf.get("jdbcurl");
            String driver = (String) chMapConf.get("driver");
            String userName = (String) chMapConf.get("username");
            String password = (String) chMapConf.get("password");
            String databaseName = (String) chMapConf.get("database-name");
            String tableName = (String) chMapConf.get("table-name");
            connection = JdbcUtil.getConnection(jdbcurl, userName, password, driver);
            DatabaseMetaData metaData = connection.getMetaData();
            ResultSet colRet = metaData.getColumns((String) null, databaseName, tableName, "%");
            System.out.println("表字段信息：");

            while (colRet.next()) {
                String columnName = colRet.getString("COLUMN_NAME");
                String columnType = colRet.getString("TYPE_NAME");
                schemaPos.add(new SchemaPo(columnName, columnType));
                System.out.println(columnName + "   " + columnType);
            }
        } finally {
            try {
                if (connection != null) {
                    connection.close();
                }
            } catch (SQLException var18) {
                var18.printStackTrace();
            }

        }

        return schemaPos;
    }

    public static String getCreateSinkTableSql(Map<String, Object> clickhouse, String sinkTableName, List<SchemaPo> schemaPos) {
        StringBuilder sinkTableSql = new StringBuilder();
        String userName = (String) clickhouse.get("username");
        String password = (String) clickhouse.get("password");
        String connector = (String) clickhouse.get("connector");
        String databaseName = (String) clickhouse.get("database-name");
        String url = (String) clickhouse.get("url");
        String tableName = (String) clickhouse.get("table-name");
        String sinkBatchSize = (String) clickhouse.get("sink.batch-size");
        String sinkFlushInterval = (String) clickhouse.get("sink.flush-interval");
        String sinkMaxRetries = (String) clickhouse.get("sink.max-retries");
        String sinkPartitionStrategy = (String) clickhouse.get("sink.partition-strategy");
        String sinkPartitionKey = (String) clickhouse.get("sink.partition-key");
        String sinkIgnoreDelete = (String) clickhouse.get("sink.ignore-delete");
        sinkTableSql.append(String.format("CREATE TABLE %s (\n", sinkTableName));
        int i = 0;
        Iterator var17 = schemaPos.iterator();

        while (var17.hasNext()) {
            SchemaPo schemaPo = (SchemaPo) var17.next();
            ++i;
            String signal = schemaPo.getSignal();
            String type = schemaPo.getType();
            if ("UInt64".equalsIgnoreCase(type)) {
                type = "BIGINT";
            } else if ("Map(String,String)".equalsIgnoreCase(type)) {
                type = "Map<String,String>";
            } else if ("Datetime".equalsIgnoreCase(type)) {
                type = "Timestamp(0)";
            } else {
                type = "String";
            }

            sinkTableSql.append(String.format("    `%s` %s", signal, type));
            sinkTableSql.append(i == schemaPos.size() ? ")" : ",\n");
        }

        sinkTableSql.append("WITH(\n");
        sinkTableSql.append(String.format("'connector' = '%s',\n", connector));
        sinkTableSql.append(String.format("'url' = '%s',\n", url));
        sinkTableSql.append(String.format("'username' = '%s',\n", userName));
        sinkTableSql.append(String.format("'password' = '%s',\n", password));
        sinkTableSql.append(String.format("'url' = '%s',\n", url));
        sinkTableSql.append(String.format("'database-name' = '%s',\n", databaseName));
        sinkTableSql.append(String.format("'table-name' = '%s',\n", tableName));
        sinkTableSql.append(String.format("'sink.batch-size' = '%s',\n", sinkBatchSize));
        sinkTableSql.append(String.format("'sink.flush-interval' = '%s',\n", sinkFlushInterval));
        sinkTableSql.append(String.format("'sink.max-retries' = '%s',\n", sinkMaxRetries));
        sinkTableSql.append(String.format("'sink.partition-strategy' = 'hash',\n"));
        sinkTableSql.append(String.format("'sink.partition-key' = 'sample_date_time',\n"));
        sinkTableSql.append(String.format("'sink.ignore-delete' = '%s'\n", sinkIgnoreDelete));
        sinkTableSql.append(" )");
        return sinkTableSql.toString();
    }


    //转换成ck需要的格式
    public static Row convertRow(Map<String, String> resultMap, List<SchemaPo> schemaPos) {
        Row row = new Row(schemaPos.size());
        for (int i = 0; i < schemaPos.size(); i++) {
            SchemaPo schemaPo = schemaPos.get(i);
            String valueStr = resultMap.get(schemaPo.getSignal());
            if (StringUtils.isBlank(valueStr)) {
                row.setField(i, null);
                continue;
            }
            if ("UInt64".equalsIgnoreCase(schemaPo.getType())) {
                Long svalue = Long.valueOf(valueStr);
                row.setField(i, Math.abs(svalue));
            } else if ("Int64".equalsIgnoreCase(schemaPo.getType())) {
                Long svalue = Long.valueOf(valueStr);
                row.setField(i, Math.abs(svalue));
            } else if ("Int32".equalsIgnoreCase(schemaPo.getType())) {
                Integer svalue = Integer.valueOf(valueStr);
                row.setField(i, svalue);
            } else if ("datetime".equalsIgnoreCase(schemaPo.getType())) {
                try {
                    Date svalue = (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")).parse(valueStr);
                    Timestamp timestamp = new Timestamp(svalue.getTime());
                    row.setField(i, timestamp);
                } catch (Exception ex) {
                    System.out.println(ex.getMessage());
                    System.out.println(Arrays.toString(ex.getStackTrace()));
                }
            } else {
                row.setField(i, valueStr);
            }
        }
        return row;
    }

}

2.3.flatmap

2.3.1.FlatMapFunction

public interface FlatMapFunction {
    public FlatMapFunction<ConsumerRecord<String, String>, Row> newInstance(List<SchemaPo> schemaPos);
}

2.4.sink

2.4.1.ClickHouseCatalog

public class ClickHouseCatalog extends AbstractCatalog {
    private static final Logger LOG = LoggerFactory.getLogger(ClickHouseCatalog.class);
    public static final String DEFAULT_DATABASE = "default";
    private final String baseUrl;
    private final String username;
    private final String password;
    private final boolean ignorePrimaryKey;
    private final Map<String, String> properties;
    private ClickHouseConnection connection;

    public ClickHouseCatalog(String catalogName, Map<String, String> properties) {
        this(catalogName, (String)properties.get("database-name"), (String)properties.get("url"), (String)properties.get("username"), (String)properties.get("password"), properties);
    }

    public ClickHouseCatalog(String catalogName, @Nullable String defaultDatabase, String baseUrl, String username, String password) {
        this(catalogName, defaultDatabase, baseUrl, username, password, Collections.emptyMap());
    }

    public ClickHouseCatalog(String catalogName, @Nullable String defaultDatabase, String baseUrl, String username, String password, Map<String, String> properties) {
        super(catalogName, defaultDatabase == null ? "default" : defaultDatabase);
        Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl), "baseUrl cannot be null or empty");
        Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(username), "username cannot be null or empty");
        Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(password), "password cannot be null or empty");
        this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
        this.username = username;
        this.password = password;
        this.ignorePrimaryKey = properties.get("catalog.ignore-primary-key") == null || Boolean.parseBoolean((String)properties.get("catalog.ignore-primary-key"));
        this.properties = Collections.unmodifiableMap(properties);
    }

    public void open() throws CatalogException {
        try {
            Properties configuration = new Properties();
            configuration.putAll(this.properties);
            configuration.setProperty(ClickHouseQueryParam.USER.getKey(), this.username);
            configuration.setProperty(ClickHouseQueryParam.PASSWORD.getKey(), this.password);
            configuration.setProperty("socket_timeout", "600000");
            String jdbcUrl = ClickHouseUtil.getJdbcUrl(this.baseUrl, this.getDefaultDatabase());
            BalancedClickhouseDataSource dataSource = new BalancedClickhouseDataSource(jdbcUrl, configuration);
            dataSource.actualize();
            this.connection = dataSource.getConnection();
            LOG.info("Created catalog {}, established connection to {}", this.getName(), jdbcUrl);
        } catch (Exception var4) {
            throw new CatalogException(String.format("Opening catalog %s failed.", this.getName()), var4);
        }
    }

    public synchronized void close() throws CatalogException {
        try {
            this.connection.close();
            LOG.info("Closed catalog {} ", this.getName());
        } catch (Exception var2) {
            throw new CatalogException(String.format("Closing catalog %s failed.", this.getName()), var2);
        }
    }

    public Optional<Factory> getFactory() {
        return Optional.of(new ClickHouseDynamicTableFactory());
    }

    public synchronized List<String> listDatabases() throws CatalogException {
        try {
            PreparedStatement stmt = this.connection.prepareStatement("SELECT name from `system`.databases");
            Throwable var2 = null;

            try {
                ResultSet rs = stmt.executeQuery();
                Throwable var4 = null;

                try {
                    List<String> databases = new ArrayList();

                    while(rs.next()) {
                        databases.add(rs.getString(1));
                    }

                    return databases;
                } catch (Throwable var31) {
                    var4 = var31;
                    throw var31;
                } finally {
                    if (rs != null) {
                        if (var4 != null) {
                            try {
                                rs.close();
                            } catch (Throwable var30) {
                                var4.addSuppressed(var30);
                            }
                        } else {
                            rs.close();
                        }
                    }

                }
            } catch (Throwable var33) {
                var2 = var33;
                throw var33;
            } finally {
                if (stmt != null) {
                    if (var2 != null) {
                        try {
                            stmt.close();
                        } catch (Throwable var29) {
                            var2.addSuppressed(var29);
                        }
                    } else {
                        stmt.close();
                    }
                }

            }
        } catch (Exception var35) {
            throw new CatalogException(String.format("Failed listing database in catalog %s", this.getName()), var35);
        }
    }

    public CatalogDatabase getDatabase(String databaseName) throws DatabaseNotExistException, CatalogException {
        if (this.listDatabases().contains(databaseName)) {
            return new CatalogDatabaseImpl(Collections.emptyMap(), (String)null);
        } else {
            throw new DatabaseNotExistException(this.getName(), databaseName);
        }
    }

    public boolean databaseExists(String databaseName) throws CatalogException {
        Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName));
        return this.listDatabases().contains(databaseName);
    }

    public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists) throws DatabaseAlreadyExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade) throws DatabaseNotEmptyException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists) throws DatabaseNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public synchronized List<String> listTables(String databaseName) throws DatabaseNotExistException, CatalogException {
        if (!this.databaseExists(databaseName)) {
            throw new DatabaseNotExistException(this.getName(), databaseName);
        } else {
            try {
                PreparedStatement stmt = this.connection.prepareStatement(String.format("SELECT name from `system`.tables where database = '%s'", databaseName));
                Throwable var3 = null;

                try {
                    ResultSet rs = stmt.executeQuery();
                    Throwable var5 = null;

                    try {
                        List<String> tables = new ArrayList();

                        while(rs.next()) {
                            tables.add(rs.getString(1));
                        }

                        return tables;
                    } catch (Throwable var32) {
                        var5 = var32;
                        throw var32;
                    } finally {
                        if (rs != null) {
                            if (var5 != null) {
                                try {
                                    rs.close();
                                } catch (Throwable var31) {
                                    var5.addSuppressed(var31);
                                }
                            } else {
                                rs.close();
                            }
                        }

                    }
                } catch (Throwable var34) {
                    var3 = var34;
                    throw var34;
                } finally {
                    if (stmt != null) {
                        if (var3 != null) {
                            try {
                                stmt.close();
                            } catch (Throwable var30) {
                                var3.addSuppressed(var30);
                            }
                        } else {
                            stmt.close();
                        }
                    }

                }
            } catch (Exception var36) {
                throw new CatalogException(String.format("Failed listing tables in catalog %s database %s", this.getName(), databaseName), var36);
            }
        }
    }

    public List<String> listViews(String databaseName) throws DatabaseNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
        if (!this.tableExists(tablePath)) {
            throw new TableNotExistException(this.getName(), tablePath);
        } else {
            Map<String, String> configuration = new HashMap(this.properties);
            configuration.put("url", this.baseUrl);
            configuration.put("database-name", tablePath.getDatabaseName());
            configuration.put("table-name", tablePath.getObjectName());
            configuration.put("username", this.username);
            configuration.put("password", this.password);
            String databaseName = tablePath.getDatabaseName();
            String tableName = tablePath.getObjectName();

            try {
                DistributedEngineFullSchema engineFullSchema = ClickHouseUtil.getAndParseDistributedEngineSchema(this.connection, tablePath.getDatabaseName(), tablePath.getObjectName());
                if (engineFullSchema != null) {
                    databaseName = engineFullSchema.getDatabase();
                    tableName = engineFullSchema.getTable();
                }
            } catch (Exception var6) {
                throw new CatalogException(String.format("Failed getting engine full of %s.%s.%s", this.getName(), databaseName, tableName), var6);
            }

            return new CatalogTableImpl(this.createTableSchema(databaseName, tableName), this.getPartitionKeys(databaseName, tableName), configuration, "");
        }
    }

    private synchronized TableSchema createTableSchema(String databaseName, String tableName) {
        try {
            PreparedStatement stmt = this.connection.prepareStatement(String.format("SELECT * from `%s`.`%s` limit 0", databaseName, tableName));
            Throwable var4 = null;

            TableSchema var24;
            try {
                ClickHouseResultSetMetaData metaData = (ClickHouseResultSetMetaData)stmt.getMetaData().unwrap(ClickHouseResultSetMetaData.class);
                Method getColMethod = metaData.getClass().getDeclaredMethod("getCol", Integer.TYPE);
                getColMethod.setAccessible(true);
                List<String> primaryKeys = this.getPrimaryKeys(databaseName, tableName);
                TableSchema.Builder builder = TableSchema.builder();

                for(int idx = 1; idx <= metaData.getColumnCount(); ++idx) {
                    ClickHouseColumnInfo columnInfo = (ClickHouseColumnInfo)getColMethod.invoke(metaData, idx);
                    String columnName = columnInfo.getColumnName();
                    DataType columnType = ClickHouseTypeUtil.toFlinkType(columnInfo);
                    if (primaryKeys.contains(columnName)) {
                        columnType = (DataType)columnType.notNull();
                    }

                    builder.field(columnName, columnType);
                }

                if (!primaryKeys.isEmpty()) {
                    builder.primaryKey((String[])primaryKeys.toArray(new String[0]));
                }

                var24 = builder.build();
            } catch (Throwable var21) {
                var4 = var21;
                throw var21;
            } finally {
                if (stmt != null) {
                    if (var4 != null) {
                        try {
                            stmt.close();
                        } catch (Throwable var20) {
                            var4.addSuppressed(var20);
                        }
                    } else {
                        stmt.close();
                    }
                }

            }

            return var24;
        } catch (Exception var23) {
            throw new CatalogException(String.format("Failed getting columns in catalog %s database %s table %s", this.getName(), databaseName, tableName), var23);
        }
    }

    private List<String> getPrimaryKeys(String databaseName, String tableName) {
        if (this.ignorePrimaryKey) {
            return Collections.emptyList();
        } else {
            try {
                PreparedStatement stmt = this.connection.prepareStatement(String.format("SELECT name from `system`.columns where `database` = '%s' and `table` = '%s' and is_in_primary_key = 1", databaseName, tableName));
                Throwable var4 = null;

                try {
                    ResultSet rs = stmt.executeQuery();
                    Throwable var6 = null;

                    try {
                        List<String> primaryKeys = new ArrayList();

                        while(rs.next()) {
                            primaryKeys.add(rs.getString(1));
                        }

                        return primaryKeys;
                    } catch (Throwable var33) {
                        var6 = var33;
                        throw var33;
                    } finally {
                        if (rs != null) {
                            if (var6 != null) {
                                try {
                                    rs.close();
                                } catch (Throwable var32) {
                                    var6.addSuppressed(var32);
                                }
                            } else {
                                rs.close();
                            }
                        }

                    }
                } catch (Throwable var35) {
                    var4 = var35;
                    throw var35;
                } finally {
                    if (stmt != null) {
                        if (var4 != null) {
                            try {
                                stmt.close();
                            } catch (Throwable var31) {
                                var4.addSuppressed(var31);
                            }
                        } else {
                            stmt.close();
                        }
                    }

                }
            } catch (Exception var37) {
                throw new CatalogException(String.format("Failed getting primary keys in catalog %s database %s table %s", this.getName(), databaseName, tableName), var37);
            }
        }
    }

    private List<String> getPartitionKeys(String databaseName, String tableName) {
        try {
            PreparedStatement stmt = this.connection.prepareStatement(String.format("SELECT name from `system`.columns where `database` = '%s' and `table` = '%s' and is_in_partition_key = 1", databaseName, tableName));
            Throwable var4 = null;

            try {
                ResultSet rs = stmt.executeQuery();
                Throwable var6 = null;

                try {
                    List<String> partitionKeys = new ArrayList();

                    while(rs.next()) {
                        partitionKeys.add(rs.getString(1));
                    }

                    return partitionKeys;
                } catch (Throwable var33) {
                    var6 = var33;
                    throw var33;
                } finally {
                    if (rs != null) {
                        if (var6 != null) {
                            try {
                                rs.close();
                            } catch (Throwable var32) {
                                var6.addSuppressed(var32);
                            }
                        } else {
                            rs.close();
                        }
                    }

                }
            } catch (Throwable var35) {
                var4 = var35;
                throw var35;
            } finally {
                if (stmt != null) {
                    if (var4 != null) {
                        try {
                            stmt.close();
                        } catch (Throwable var31) {
                            var4.addSuppressed(var31);
                        }
                    } else {
                        stmt.close();
                    }
                }

            }
        } catch (Exception var37) {
            throw new CatalogException(String.format("Failed getting partition keys of %s.%s.%s", this.getName(), databaseName, tableName), var37);
        }
    }

    public boolean tableExists(ObjectPath tablePath) throws CatalogException {
        try {
            return this.databaseExists(tablePath.getDatabaseName()) && this.listTables(tablePath.getDatabaseName()).contains(tablePath.getObjectName());
        } catch (DatabaseNotExistException var3) {
            return false;
        }
    }

    public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) throws TableNotExistException, TableAlreadyExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath) throws TableNotExistException, TableNotPartitionedException, CatalogException {
        return Collections.emptyList();
    }

    public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws TableNotExistException, TableNotPartitionedException, PartitionSpecInvalidException, CatalogException {
        return Collections.emptyList();
    }

    public List<CatalogPartitionSpec> listPartitionsByFilter(ObjectPath tablePath, List<Expression> filters) throws TableNotExistException, TableNotPartitionedException, CatalogException {
        return Collections.emptyList();
    }

    public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws PartitionNotExistException, CatalogException {
        throw new PartitionNotExistException(this.getName(), tablePath, partitionSpec);
    }

    public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException {
        throw new UnsupportedOperationException();
    }

    public void createPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, CatalogPartition partition, boolean ignoreIfExists) throws TableNotExistException, TableNotPartitionedException, PartitionSpecInvalidException, PartitionAlreadyExistsException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void dropPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) throws PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void alterPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, CatalogPartition newPartition, boolean ignoreIfNotExists) throws PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public List<String> listFunctions(String dbName) throws DatabaseNotExistException, CatalogException {
        return Collections.emptyList();
    }

    public CatalogFunction getFunction(ObjectPath functionPath) throws FunctionNotExistException, CatalogException {
        throw new FunctionNotExistException(this.getName(), functionPath);
    }

    public boolean functionExists(ObjectPath functionPath) throws CatalogException {
        return false;
    }

    public void createFunction(ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists) throws FunctionAlreadyExistException, DatabaseNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void alterFunction(ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists) throws FunctionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists) throws FunctionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) throws TableNotExistException, CatalogException {
        return CatalogTableStatistics.UNKNOWN;
    }

    public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) throws TableNotExistException, CatalogException {
        return CatalogColumnStatistics.UNKNOWN;
    }

    public CatalogTableStatistics getPartitionStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws PartitionNotExistException, CatalogException {
        return CatalogTableStatistics.UNKNOWN;
    }

    public CatalogColumnStatistics getPartitionColumnStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws PartitionNotExistException, CatalogException {
        return CatalogColumnStatistics.UNKNOWN;
    }

    public void alterTableStatistics(ObjectPath tablePath, CatalogTableStatistics tableStatistics, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void alterTableColumnStatistics(ObjectPath tablePath, CatalogColumnStatistics columnStatistics, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException, TablePartitionedException {
        throw new UnsupportedOperationException();
    }

    public void alterPartitionStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, CatalogTableStatistics partitionStatistics, boolean ignoreIfNotExists) throws PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }

    public void alterPartitionColumnStatistics(ObjectPath tablePath, CatalogPartitionSpec partitionSpec, CatalogColumnStatistics columnStatistics, boolean ignoreIfNotExists) throws PartitionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }
}

2.5.Kafka2ClickHouse

2.5.1.Kafka2chApp

public class Kafka2chApp {
    private static final Logger log = LoggerFactory.getLogger(Kafka2chApp.class);
    private static String SINK_TABLE = "sinkTable";
    private static String KAFKA_TEMP_VIEW = "kafkaTempView";

    /**
     * @param appName            mysql配置表对应字段
     * @param configName         mysql配置表对应字段
     * @throws Exception
     */
    public static void run(String appName, String configName, FlatMapFunction FlatMapFunction) throws Exception {
        log.info("Kafka2chApp.run传参appName:{}, configName:{}", appName, configName);

        // 获得数据库中的配置
        Map<String, Object> mapConf = RemoteConfigUtil.getByAppNameAndConfigName(appName, configName);
        if (mapConf == null || mapConf.size() == 0) return;
        Map<String, Object> clickhouseConf = (Map<String, Object>) mapConf.get("clickhouse");
        Map<String, Object> kafkaConsumerConf = (Map<String, Object>) mapConf.get("kafka-consumer");
        Map<String, Object> hdfsConf = (Map<String, Object>) mapConf.get("hdfs");
        // long beforeTime2Dropout = System.currentTimeMillis() - (Long) mapConf.get("before2DropoutHourStep") * 3600;
        // long after2DropoutTime = System.currentTimeMillis();


        // 初始化TableEnv & 获得流
        StreamExecutionEnvironment streamEnv = StreamEnv.getStreamEnv(hdfsConf);
        
        streamEnv.setParallelism(ckP);
        
        StreamTableEnvironment tableEnv = TableEnv.getTableEnv();

        // 处理
        List<SchemaPo> schemaPos = ClickhouseUtil.getSchemaPoList(clickhouseConf);

        TypeInformation[] types = getTypeInformationArray(schemaPos);

        // TypeInformation[] types = (schemaPos);
        String[] fieldNames = SchemaPoUtil.getFieldLists(schemaPos);

        FlatMapFunction<ConsumerRecord<String, String>, Row> flatMapFunction = x5lFlatMapFunction.newInstance(schemaPos);
        DataStreamSource<ConsumerRecord<String, String>> stream;
        SingleOutputStreamOperator<Row> infos;

        stream = streamEnv.addSource(CommonUtils.getKafkaConsumer(kafkaConsumerConf));

        System.out.println("Source 设置并行度为"+streamEnv.getParallelism());
        }

        infos = stream.flatMap(flatMapFunction);

        infos = infos.map(e -> e,new RowTypeInfo(types, fieldNames));
        System.out.println("map 设置并行度为"+streamEnv.getParallelism());
        }

        // 创建kafka数据临时视图
        tableEnv.createTemporaryView(KAFKA_TEMP_VIEW, infos);

        // 创建存放kafka数据的clickhouse映射表
        // String createSinkTableSql = ClickhouseUtil.getCreateSinkTableSql(clickhouseConf, SINK_TABLE, schemaPos);

        Map<String, String> props = new HashMap<>();
        props.put(ClickHouseConfig.DATABASE_NAME, (String) clickhouseConf.get("database-name"));
        props.put(ClickHouseConfig.URL, (String) clickhouseConf.get("url"));
        props.put(ClickHouseConfig.USERNAME, (String) clickhouseConf.get("username"));
        props.put(ClickHouseConfig.PASSWORD, (String) clickhouseConf.get("password"));
        props.put(ClickHouseConfig.SINK_FLUSH_INTERVAL, (String) clickhouseConf.get("sink.flush-interval"));
        props.put(ClickHouseConfig.SINK_BATCH_SIZE, (String) clickhouseConf.get("sink.batch-size"));

        Catalog cHcatalog = new ClickHouseCatalog("clickhouse", props);

        tableEnv.registerCatalog("clickhouse", cHcatalog);
        tableEnv.useCatalog("clickhouse");

        // Arrays.stream(tableEnv.listCatalogs()).forEach(e -> System.out.println("catalog: " + e));
        // Arrays.stream(tableEnv.listDatabases()).forEach(e -> System.out.println("database: " + e));

        // System.out.println(tableEnv.listTables().length);
        // Arrays.stream(tableEnv.listTables()).forEach(e -> System.out.println("table: " + e));

        // tableEnv.executeSql(createSinkTableSql);
        // System.out.println(tableEnv.executeSql("select * from " + KAFKA_TEMP_VIEW).getTableSchema());
        //拼接sql
        String insertSql = "insert into `" + clickhouseConf.get("table-name") + "` select * from default_catalog.default_database." + KAFKA_TEMP_VIEW;
        // System.out.println("insertSql: " + insertSql);
        // log.info("insertSql: ", insertSql);


        //执行sql
        tableEnv.executeSql(insertSql);

        // 测试打印infos结果
        /*infos.print();
        streamEnv.executeAsync();*/
    }
    
    public static TypeInformation[] getTypeInformationArray(List<SchemaPo> schemaPos) {
        // String[] fieldNames = new String[columnTypeMap.size()];
        TypeInformation[] types = new TypeInformation[schemaPos.size()];
        int i = 0;

        for (SchemaPo po : schemaPos) {
            if ("String".equalsIgnoreCase(po.getType())) {
                types[i] = Types.STRING;
            } else if ("Int64".equalsIgnoreCase(po.getType())) {
                types[i] = Types.LONG;

            } else if ("UInt64".equalsIgnoreCase(po.getType())) {
                types[i] = Types.LONG;

            } else if ("Int32".equalsIgnoreCase(po.getType())) {
                types[i] = Types.INT;
            } else if ("Int8".equalsIgnoreCase(po.getType())) {
                types[i] = Types.INT;
            } else if ("datetime".equalsIgnoreCase(po.getType())) {
                types[i] = Types.SQL_TIMESTAMP;
            } else if ("Map(String,String)".equalsIgnoreCase(po.getType())) {
                types[i] = Types.MAP(Types.STRING, Types.STRING);
            } else {
                types[i] = Types.STRING;
            }
            i++;
        }
        return types;
    }

}

2.5.2.Kafka2Ck-ODS

public class Kafka2Ck-ODS implements FlatMapFunction {
    private static Logger logger = Logger.getLogger(Kafka2Ck-ODS.class);

    public static void main(String[] args) throws Exception {
        Kafka2chApp.run(Kafka2Ck-ODS.class.getName(), args[0], new Kafka2Ck-ODS());
    }

    @Override
    public FlatMapFunction<ConsumerRecord<String, String>, Row> newInstance(List<SchemaPo> schemaPos) {

        return new FlatMapFunction<ConsumerRecord<String, String>, Row>() {
            @Override
            public void flatMap(ConsumerRecord<String, String> record, Collector<Row> out) throws Exception {
                // System.out.println("record ===> " +record); // 测试
                String value = record.value();
                try {
                    HashMap<String, Object> infoMap = JSON.parseObject(value, HashMap.class);

                    // 处理dataListMap中的数据
                    for (Map.Entry<String, String> entry : dataListMap.entrySet()) {
                        String key = entry.getKey();
                        String value1 = entry.getValue();
                        resultMap.put(key.toLowerCase(), value1);
                    }

                    Row row = TableEnv.getRowBySchemaPo1(resultMap, schemaPos);
                    out.collect(row);
                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.printf("数据异常，原因是%s,topic为%s,key为%s,value为%s%n", e.getMessage(), record.topic(), record.key(), record.value());
                }
            }
        };
    }
}