Flink写入clickhouse案例

2022-02-17

字数统计: 1.2k字 | 阅读时长≈ 5分

Flink写入clickhouse案例

源码地址:

https://github.com/ShangBaiShuYao/flink-learning-from-zhisheng/blob/main/Flink-shangbaishuyao-RTDW-gmall-realtime/src/main/java/com/shangbaishuyao/gmall/realtime/app/DWS/KeywordStats4ProductApp.java

https://github.com/ShangBaiShuYao/flink-learning-from-zhisheng/blob/main/Flink-shangbaishuyao-RTDW-gmall-realtime/src/main/java/com/shangbaishuyao/gmall/realtime/utils/ClickHouseUtil.java

主程序:

package com.shangbaishuyao.gmall.realtime.app.DWS;

import com.shangbaishuyao.gmall.realtime.app.Function.KeywordUDTF;
import com.shangbaishuyao.gmall.realtime.bean.KeywordStats;
import com.shangbaishuyao.gmall.realtime.utils.ClickHouseUtil;
import com.shangbaishuyao.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Author: shangbaishuyao
 * Date: 2021/3/1
 * Desc:  从商品统计中获取关键词
 */
public class KeywordStats4ProductApp {
    public static void main(String[] args) throws Exception {
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);
        /*
        //CK相关设置
        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        StateBackend fsStateBackend = new FsStateBackend(
                "hdfs://hadoop202:8020/gmall/flink/checkpoint/ProvinceStatsSqlApp");
        env.setStateBackend(fsStateBackend);
        System.setProperty("HADOOP_USER_NAME","shangbaishuyao");
        */
        //TODO 1.定义Table流环境
        EnvironmentSettings settings = EnvironmentSettings
            .newInstance()
            .inStreamingMode()
            .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        //TODO 2.注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyze",  KeywordUDTF.class);
        tableEnv.createTemporarySystemFunction("keywordProductC2R",  KeywordProductC2RUDTF.class);

        //TODO 3.将数据源定义为动态表
        String groupId = "keyword_stats_app";
        String productStatsSourceTopic ="dws_product_stats";

        tableEnv.executeSql("CREATE TABLE product_stats (spu_name STRING, " +
            "click_ct BIGINT," +
            "cart_ct BIGINT," +
            "order_ct BIGINT ," +
            "stt STRING,edt STRING ) " +
            "  WITH ("+ MyKafkaUtil.getKafkaDDL(productStatsSourceTopic,groupId)+")");

        //TODO 6.聚合计数
        Table keywordStatsProduct = tableEnv.sqlQuery("select keyword,ct,source, " +
            "DATE_FORMAT(stt,'yyyy-MM-dd HH:mm:ss')  stt," +
            "DATE_FORMAT(edt,'yyyy-MM-dd HH:mm:ss') as edt, " +
            "UNIX_TIMESTAMP()*1000 ts from product_stats  , " +
            "LATERAL TABLE(ik_analyze(spu_name)) as T(keyword) ," +
            "LATERAL TABLE(keywordProductC2R( click_ct ,cart_ct,order_ct)) as T2(ct,source)");

        //TODO 7.转换为数据流
        DataStream<KeywordStats> keywordStatsProductDataStream =
            tableEnv.<KeywordStats>toAppendStream(keywordStatsProduct, KeywordStats.class);

        keywordStatsProductDataStream.print();
        //TODO 8.写入到ClickHouse
        keywordStatsProductDataStream.addSink(
            ClickHouseUtil.<KeywordStats>getJdbcSink(
                "insert into keyword_stats(keyword,ct,source,stt,edt,ts) values(?,?,?,?,?,?)"
            )
        );
        env.execute();
    }
}

clickhouse工具类:

package com.shangbaishuyao.gmall.realtime.utils;

import com.shangbaishuyao.gmall.realtime.bean.TransientSink;
import com.shangbaishuyao.gmall.realtime.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Author: 上白书妖
 * Date: 2021/2/23
 * Desc: 操作ClickHouse的工具类
 *
 * 其中flink-connector-jdbc 是官方通用的jdbcSink包。
 * 只要引入对应的jdbc驱动，flink可以用它应对各种支持jdbc的数据库，
 * 比如phoenix也可以用它。但是这个jdbc-sink只支持数据流对应一张数据表。
 * 如果是一流对多表，就必须通过自定义的方式实现了，比如之前的维度数据。
 * 虽然这种jdbc-sink只能一流对一表，但是由于内部使用了预编译器，
 * 所以可以实现批量提交以优化写入速度。
 *
 * A.JdbcSink.<T>sink( )的四个参数说明
 * 参数1： 传入Sql，格式如：insert into xxx values(?,?,?,?)
 * 参数2:  可以用lambda表达实现(jdbcPreparedStatement, t) -> t为数据对象，要装配到语句预编译器的参数中。
 * 参数3：设定一些执行参数，比如重试次数，批次大小。
 * 参数4：设定连接参数，比如地址，端口，驱动名。
 */
public class ClickHouseUtil {
    /**
     * 获取向Clickhouse中写入数据的SinkFunction
     *
     * @param sql
     * @param <T>
     * @return
     */
    public static <T> SinkFunction getJdbcSink(String sql) {
        SinkFunction<T> sinkFunction = JdbcSink.<T>sink(
            //要执行的SQL语句
            sql,
            //执行写入操作   就是将当前流中的对象属性赋值给SQL的占位符 insert into visitor_stats values(?,?,?,?,?,?,?,?,?,?,?,?)
            new JdbcStatementBuilder<T>() {
                //obj  就是流中的一条数据对象
                //我要想实现通用型的话,这个当前流的类型就不能固定,只能使用泛型.T. 可以用反射拿到属性.
                @Override
                public void accept(PreparedStatement ps, T obj) throws SQLException {
                    //insert into visitor_stats values(?,?,?,?,?,?,?,?,?,?,?,?)
                    //获取当前类中  所有的属性
                    Field[] fields = obj.getClass().getDeclaredFields();//这个可以获取私有的.
                    //跳过的属性计数    insert into visitor_stats values(?,?,?,?,跳过,?,?,?,?,?,?,?)
                    int skipOffset = 0;

                    for (int i = 0; i < fields.length; i++) {
                        Field field = fields[i];
                        //通过属性对象获取属性上是否有@TransientSink注解
                        TransientSink transientSink = field.getAnnotation(TransientSink.class);
                        //如果transientSink不为空，说明属性上有@TransientSink标记，那么在给?占位符赋值的时候，应该跳过当前属性
                        if (transientSink != null) {
                            skipOffset++;
                            continue;
                        }
                        //设置私有属性可访问
                        field.setAccessible(true);
                        try {
                            //获取属性值
                            Object o = field.get(obj);
                            ps.setObject(i + 1 - skipOffset, o);
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                }
            },
            //构建者设计模式，创建JdbcExecutionOptions对象，给batchSize属性赋值，执行执行批次大小
            new JdbcExecutionOptions.Builder().withBatchSize(5).build(),
            //构建者设计模式，JdbcConnectionOptions，给连接相关的属性进行赋值
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(GmallConfig.CLICKHOUSE_URL)
                .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                .build()
        );
        return sinkFunction;
    }
}

本文作者： xubatian
本文链接： http://xubatian.cn/Flink写入clickhouse案例/
版权声明： 本博客所有文章除特别声明外均为原创，采用 CC BY 4.0 CN协议许可协议。转载请注明出处:https://www.xubatian.cn/