DWS层:商品行为关键词主题宽表

叁歲伎倆 2022-11-13 12:22 13阅读 0赞

# DWS层:商品行为关键词主题宽表 #

从商品主题获得，商品关键词与点击次数、订单次数、添加购物次数的统计表。

## 重构DWSProductStatsApp ##

重构DWSProductStatsApp, 增加最终的数据到`Kafka`的代码(因为flink没有直接读取clickhousse的方法，借助kafka中转)

private void sink2Kafka(SingleOutputStreamOperator<ProductStats> resultStream) { 
        resultStream.map(JSON::toJSONString).addSink(MyKafkaUtil.getKafkaSink("dws_product_stats"));
    }

## 自定义UDTF函数 ##

实现点击次数、订单次数、添加购物次数的统计

import com.atguigu.realtime.common.GmallConstant;
    import org.apache.flink.table.annotation.DataTypeHint;
    import org.apache.flink.table.annotation.FunctionHint;
    import org.apache.flink.table.functions.TableFunction;
    import org.apache.flink.types.Row;
    
    
    @FunctionHint(output = @DataTypeHint("ROW<source string, ct bigint>"))
    public class KeyWordProductUdtf extends TableFunction<Row> { 
        public void eval(Long clickCount, Long cartCount, Long orderCount) { 
            
            if(clickCount > 0){ 
                collect(Row.of(GmallConstant.KEYWORD_CLICK, clickCount));
            }
            
            if(cartCount > 0){ 
                collect(Row.of(GmallConstant.KEYWORD_CART, cartCount));
            }
            
            if(orderCount > 0){ 
                collect(Row.of(GmallConstant.KEYWORD_ORDER, orderCount));
            }
        }
    }

## 具体实现代码 ##

数据仍然写入到关键词主题表: keyword\_stats\_2021

package com.gmall.realtime.app.dws;
    
    import com.gmall.realtime.app.BaseSqlAPP;
    import com.gmall.realtime.udf.KeyWordProductUdtf;
    import com.gmall.realtime.udf.KeyWorldUdtf;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    public class DWSProductKeyWordStatsApp  extends BaseSqlAPP { 
    
        public static void main(String[] args) { 
            new DWSProductKeyWordStatsApp().init(30006,2,"DWSProductKeyWordStatsApp");
        }
    
        @Override
        public void run(StreamTableEnvironment tenv) { 
    
            tenv.executeSql("create table product_stats(" +
                    " spu_name string, " +
                    " click_ct bigint, " +
                    " cart_ct bigint, " +
                    " order_ct bigint, " +
                    " stt string, " +
                    " edt string " +
                    ")with(" +
                    " 'connector' = 'kafka'," +
                    " 'topic' = 'dws_product_stats'," +
                    " 'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092'," +
                    " 'properties.group.id' = 'DWSProductKeyWordStatsApp'," +
                    " 'scan.startup.mode' = 'latest-offset'," +
                    " 'format' = 'json'" +
                    ")");
    
            tenv.createFunction("ik_anlyzer", KeyWorldUdtf.class);
            final Table t1 = tenv.sqlQuery(" select " +
                    " keyword, " +
                    " click_ct , " +
                    " cart_ct , " +
                    " order_ct , " +
                    " stt , " +
                    " edt " +
                    " from product_stats, " +
                    " lateral table (ik_anlyzer(spu_name)) as T(keyword) ");
            tenv.createTemporaryView("t1",t1);
    
    
            //1.按照keyword，stt，edt进行聚合
            final Table t2 = tenv.sqlQuery("select " +
                    " keyword, " +
                    " stt , " +
                    " edt , " +
                    " sum(click_ct) click_ct , " +
                    " sum(cart_ct) cart_ct , " +
                    " sum(order_ct) order_ct " +
                    " from t1 " +
                    " group by keyword, stt, edt ");
            tenv.createTemporaryView("t2",t2);
    
    
            //2.列转行，一行的三个指标，转成三行
            tenv.createFunction("keyword_product", KeyWordProductUdtf.class);
            final Table t3 = tenv.sqlQuery("select " +
                    " stt, " +
                    " edt, " +
                    " keyword, " +
                    " source, " +
                    " ct, " +
                    " unix_timestamp()*1000 ts " +
                    " from t2 , " +
                    " lateral table(keyword_product(click_ct, cart_ct, order_ct)) as T(source, ct)");
    
    
            // 数据写入到ClickHouse
            tenv.executeSql("create table keyword_stats_2021(" +
                    " stt string," +
                    " edt string," +
                    " keyword string," +
                    " source string," +
                    " ct bigint," +
                    " ts bigint," +
                    " PRIMARY KEY (stt, edt, keyword, source) NOT ENFORCED" +
                    ")with(" +
                    " 'connector' = 'clickhouse', " +
                    " 'url' = 'clickhouse://hadoop162:8123', " +
                    " 'database-name' = 'gmall2021', " +
                    " 'table-name' = 'keyword_stats_2021'," +
                    " 'sink.batch-size' = '100', " +
                    " 'sink.flush-interval' = '1000', " +
                    " 'sink.max-retries' = '3' " +
                    ")");
    
            t3.executeInsert("keyword_stats_2021");
    
    
        }
    }