Flink自定义函数（自定义分词函数UDTF）

小灰灰 2022-11-13 00:59 327阅读 0赞

# Flink自定义函数概述 #

自定义函数（UDF）是一种扩展开发机制，可以用来在查询语句里调用难以用其他方式表达的频繁使用或自定义的逻辑。  
函数分类:

1.  Scalar functions  
    类似于spark的udf
2.  Table functions  
    类似于 spark的udtf
3.  Aggregate functions  
    类似于spark的udaf
4.  Table aggregate functions
5.  Async table functins

## 自定义Table Function实现分词功能 ##

一个词条包括多个词语所以分词是指上是一种一对多的拆分，一拆多的情况，选择Table Function。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

参考: https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/functions/udfs.html

package com.gmall.realtime.udf;
    
    import com.gmall.realtime.util.MyKeyWordUtil;
    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;
    
    import java.util.List;
    
    //告诉flink，row有几列，每列的数据类型
    @FunctionHint(output = @DataTypeHint("row<k string>"))
    public class KeyWorldUdtf extends TableFunction<Row> { 
    
        public void eval(String text){ 
            final List<String> words = MyKeyWordUtil.analysizer(text);
            for (String word : words) { 
                //一进多出
                collect(Row.of(word));
            }
        }
    }

## 分词函数工具类 ##

**导入IK分词器依赖**

<dependency>
        <groupId>com.janeluo</groupId>
        <artifactId>ikanalyzer</artifactId>
        <version>2012_u6</version>
    </dependency>

**封装分词工具类**

package com.gmall.realtime.util;
    
    import org.wltea.analyzer.core.IKSegmenter;
    import org.wltea.analyzer.core.Lexeme;
    import java.io.IOException;
    import java.io.StringReader;
    import java.util.ArrayList;
    import java.util.List;
    
    public class MyKeyWordUtil { 
        public static List<String> analysizer(String text){ 
    
            final StringReader reader = new StringReader(text);
            
            //创建IK分词器
            final IKSegmenter ikSegmenter = new IKSegmenter(reader, true);
            try { 
                final List<String> result = new ArrayList<>();
                Lexeme next = ikSegmenter.next();
    
                while (next != null){ 
                    final String word = next.getLexemeText();
                    result.add(word);
                    next=ikSegmenter.next();
                }
                return result;
    
            } catch (IOException e) { 
                e.printStackTrace();
            }
            return null;
        }
    
        public static void main(String[] args) { 
            System.out.println(analysizer("我是中国人，我还有梦要去追，我还有要守护的人"));
        }
    
    }

## 主题宽表具体实现代码 ##

**在ClickHouse中建表**

use gmall2021;
    create table keyword_stats_2021 (
        stt DateTime,
        edt DateTime,
        keyword String ,
        source String ,
        ct UInt64 ,
        ts UInt64
    )engine =ReplacingMergeTree( ts)
            partition by  toYYYYMMDD(stt)
            order by  ( stt,edt,keyword,source );

**代码清单**

package com.gmall.realtime.app.dws;
    
    import com.gmall.realtime.app.BaseSqlAPP;
    import com.gmall.realtime.common.GmallConstant;
    import com.gmall.realtime.udf.KeyWorldUdtf;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    public class DWSKeyWordStatsApp extends BaseSqlAPP { 
        public static void main(String[] args) { 
            new DWSKeyWordStatsApp().init(30004,2,"DWSKeyWordStatsApp");
        }
    
        @Override
        public void run(StreamTableEnvironment tenv) { 
    
            tenv.executeSql("create table page_view (" +
                    " common map<string,string>, " +
                    " page map<string,string>, " +
                    " ts bigint, " +
                    " et as to_timestamp(from_unixtime(ts / 1000, 'yyyy-MM-dd HH:mm:ss')), " +
                    " watermark for et as et - interval '5' second " +
                    ")with(" +
                    " 'connector' = 'kafka', " +
                    " 'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                    " 'properties.group.id' = 'DWSKeyWordStatsApp', " +
                    " 'topic' = 'dwd_page_log', " +
                    " 'scan.startup.mode' = 'latest-offset'," +
                    " 'format' = 'json'" +
                    ")");
    
            //统计每个关键词出现的次数
    
            //1.过滤出需要的数据，item不是null，并且pageId是good_list
            final Table t1 = tenv.sqlQuery("select " +
                    " page['item'] keyword, " +
                    " et " +
                    " from page_view " +
                    " where page['item'] is not null " +
                    " and page['page_id']='good_list' ");
            tenv.createTemporaryView("t1",t1);
    
            //2.利用udtf函数进行炸裂 拆分
            //2.1 注册自定义函数
            tenv.createTemporaryFunction("ik_analyzer", KeyWorldUdtf.class);
    
            final Table t2 = tenv.sqlQuery("select " +
                    " kw, " +
                    " et " +
                    " from t1, " +
                    " lateral table(ik_analyzer(keyword)) as T(kw) ");
            tenv.createTemporaryView("t2",t2);
    
            //3. 开窗聚合
            final Table t3 = tenv.sqlQuery("select " +
                    " date_format(tumble_start(et,interval '10' second), 'yyyy-MM-dd HH:mm:ss') stt, " +
                    " date_format(tumble_end(et,interval '10' second), 'yyyy-MM-dd HH:mm:ss') edt, " +
                    " kw keyword, " +
                    " '" + GmallConstant.KEYWORD_SEARCH + "' source, " +
                    " count(*) ct, " +
                    " unix_timestamp()*1000 ts " +
                    " from t2 " +
                    " group by " +
                    " tumble(et, interval '10' second), kw");
    
    
            //4.建表，关联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");
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221022/17051a2de7ea46468f12fdd598347a79.png