需求描述：
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 方法获取配置。文章来源：https://www.toymoban.com/news/detail-773530.html
1）导入相关依赖
这里的依赖比较冗余，大家可以根据各自需求做删除或保留。文章来源地址https://www.toymoban.com/news/detail-773530.html
<?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());
}
}
};
}
}
到了这里，关于【Flink-Kafka-To-ClickHouse】使用 Flink 实现 Kafka 数据写入 ClickHouse的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
