百度问答官网,网站如何做排名优化,网站访问量咋做,东莞最新网站建设软件ClickHouse官网文档
Flink 读取 ClickHouse 数据两种驱动
ClickHouse 官方提供Clickhouse JDBC.【建议使用】第3方提供的Clickhouse JDBC. ru.yandex.clickhouse.ClickHouseDriver ru.yandex.clickhouse.ClickHouseDriver.现在是没有维护 ClickHouse 官方提供Clickhouse JDBC…ClickHouse官网文档
Flink 读取 ClickHouse 数据两种驱动
ClickHouse 官方提供Clickhouse JDBC.【建议使用】第3方提供的Clickhouse JDBC. ru.yandex.clickhouse.ClickHouseDriver ru.yandex.clickhouse.ClickHouseDriver.现在是没有维护 ClickHouse 官方提供Clickhouse JDBC的包名com.clickhouse.jdbc.* 有些版本com.clickhouse.jdbc.* 包含了 ru.yandex.clickhouse.ClickHouseDriver. 因此加载包的时候一定要注意导入的包名 引入依赖 !-- clickhouse jdbc driver --dependencygroupIdcom.clickhouse/groupIdartifactIdclickhouse-jdbc/artifactId/dependency使用的是 0.3 这个版本该版本就包含上述3方CH jdbc包 !-- CH JDBC版本推荐使用 0.3, 0.4的版本是要 JDK 17 --clickhouse-jdbc.version0.3.2-patch11/clickhouse-jdbc.version自定义Source
测试表映射实体类该表仅有一个name字段
Data
NoArgsConstructor
AllArgsConstructor
public class CHTestPO {private String name;}Flink Clickhouse Source
public class ClickHouseSource implements SourceFunctionCHTestPO {private final String URL;private final String SQL;public ClickHouseSource(String URL, String SQL) {this.URL URL;this.SQL SQL;}Overridepublic void run(SourceContextCHTestPO output) throws Exception {// Properties是持久化的属性集 Properties的key和value都是字符串Properties properties new Properties();ClickHouseDataSource clickHouseDataSource new ClickHouseDataSource(URL, properties);// 使用 try-with-resource 方式关闭JDBC连接 无需手动关闭try (ClickHouseConnection conn clickHouseDataSource.getConnection()) {// clickhouse 通过游标的方式读取数据Statement stmt conn.createStatement();ResultSet rs stmt.executeQuery(SQL);while (rs.next()) {String name rs.getString(1);output.collect(new CHTestPO(name));}}}Overridepublic void cancel() {}
}自定义Sink
需额外引入依赖 !-- Flink-Connector-Jdbc --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-jdbc_${scala.binary.version}/artifactId/dependencyJava 对sql语句处理的两个对象
PreparedStatement对象能够对预编译之后的sql语句进行处理【SQL 语句预编译通过占位符?实现可以防止sql注入】Statement对象只能对静态的sql语句进行处理
核心代码
/*** 使用 Flink-jdbc-connector 批量写入 sql语句的预编译 写入 Clickhouse*/
public class ClickHouseJdbcSinkT {private final SinkFunctionT sink;private final static String NA null;public ClickHouseJdbcSink(String sql, int batchSize, String url) {sink JdbcSink.sink(sql,// 对sql语句进行预编译new ClickHouseJdbcStatementBuilderT(),// 设置批量插入数据new JdbcExecutionOptions.Builder().withBatchSize(batchSize).build(),// 设置ClickHouse连接配置new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl(url).build());}public SinkFunctionT getSink() {return this.sink;}/*** 对预编译之后的sql语句进行占位符替换** param ps: PreparedStatement对象 下标从 1 开始* param fields: clickhouse表PO对象的属性字段* param object: clickhouse表PO对象的属性字段所对应的数据类型*/public static void setPreparedStatement(PreparedStatement ps,Field[] fields,Object object) throws IllegalAccessException, SQLException {// 遍历 Field[]for (int i 1; i fields.length; i) {// 取出每个Field实例Field field fields[i - 1];// 指示反射的对象在使用时应该取消 Java 语言访问检查field.setAccessible(true);// 通过Field实例的get方法返回指定的对象Object o field.get(object);if (o null) {ps.setNull(i, 0);continue;}// 这里统一设为字符型String fieldValue o.toString();// 变量和常量的比较通常将常量放前可以避免空指针if (!NA.equals(fieldValue) !.equals(fieldValue)) {// 替换对应位置的占位符ps.setObject(i, fieldValue);} else {ps.setNull(i, 0);}}}}对sql语句进行预编译
Slf4j
public class ClickHouseJdbcStatementBuilderT implements JdbcStatementBuilderT {Overridepublic void accept(PreparedStatement preparedStatement, T t) throws SQLException {/* *********************** Java通过反射获取类的字段** 1. getDeclaredFields()获取所有的字段不会获取父类的字段* 2. getFields(): 只能会public字段获取包含父类的字段** *********************/Field[] fields t.getClass().getDeclaredFields();// 将获取到的字段替换sql预编译之后的占位符。try {ClickHouseJdbcSink.setPreparedStatement(preparedStatement, fields, t);} catch (IllegalAccessException e) {log.error(sql 预编译失败, e);e.printStackTrace();}}
}ClickHouse读写工具类 public class ClickHouseUtil {private static final String URL;static {ParameterTool parameterTool ParameterUtil.getParameters();URL parameterTool.get(clickhouse.url);}/*** 读取clickhouse*/public static DataStreamCHTestPO read(StreamExecutionEnvironment env, String sql) {return env.addSource(new ClickHouseSource(URL, sql));}/*** 批量写入ClickHouse*/public static T DataStreamSinkT batchWrite(DataStreamT dataStream,String sql,int batchSize) {//生成 SinkFunctionClickHouseJdbcSinkT clickHouseJdbcSink new ClickHouseJdbcSinkT(sql, batchSize, URL);return dataStream.addSink(clickHouseJdbcSink.getSink());}}测试一下
public class ClickHouseUtilTest {DisplayName(测试Flinkjdbc游标读取Clickhouse)Testvoid testRead() throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 设置并行度1env.setParallelism(1);// 从default数据库的user表中读取数据String sql select * from default.user;DataStreamCHTestPO ds ClickHouseUtil.read(env, sql);// 打印数据流中的元素ds.print(clickhouse);// 执行程序env.execute();}DisplayName(测试Flink-Connector-jdbc预编译批量写入Clickhouse)Testvoid testBatchWrite() throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 设置并行度1env.setParallelism(1);CHTestPO po new CHTestPO();po.setName(Lucy);CHTestPO po1 new CHTestPO();po1.setName(Jack);DataStreamCHTestPO ds env.fromCollection(Arrays.asList(po, po1));// 定义将数据写入ClickHouse数据库的SQL语句String sql insert into default.user(name) values(?);// 调用ClickHouseUtil的batchWrite方法将数据流ds中的数据批量写入ClickHouse数据库ClickHouseUtil.batchWrite(ds, sql, 2);// 执行程序env.execute();}
}此时表中仅一行记录 读取没有问题 写入没有问题
