首页 > 其他分享 >中电金信:技术实践|Flink维度表关联方案解析

中电金信:技术实践|Flink维度表关联方案解析

时间:2024-03-15 11:22:18浏览次数:30  
标签:city name 数据库 Flink 关联 金信 维表 维度

导语:Flink是一个对有界和无界数据流进行状态计算的分布式处理引擎和框架,主要用来处理流式数据。它既可以处理有界的批量数据集,也可以处理无界的实时流数据,为批处理和流处理提供了统一编程模型。

 

维度表可以看作是用户来分析数据的窗口,它区别于事实表业务真实发生的数据,通常用来表示业务属性,以便为分析者提供有用的信息。在实际场景中,由于数据是实时变化的,因此需要通过将维度表进行关联,来保证业务的时效性和稳定性。本文主要围绕Flink维度表关联方案进行论述,分析不同关联方案的作用和特点,与各位读者共飨。

 

维度表与事实表的关联是数据分析中常见的一种分析方式,在传统数仓系统中,由于数据是有界的,因此关联实现相对简单。但是在实时系统或实时数仓中,数据是无界的,关联时需要考虑的问题就会复杂很多,如数据迟到导致的关联结果不准确、缓存数据消耗资源过大等等。

 

在典型的实时系统中,维表数据一般来源于源系统的OLTP数据库中,采用CDC技术将维表数据实时采集到Kafka或其他消息队列,最后保存到HBase、Hudi、Redis等组件中供数据分析使用。一个比较常见的架构图如下:

​编辑

Flink维度表关联有多种方案,包括实时lookup数据库关联、预加载维表关联、广播维度表、Temporal Table Function Join等。每种方案都有各自的特点,需要结合实际情况综合判断,维表关联方案主要考虑的因素有如下几个方面:

 

■ 实现复杂度:实现维表关联复杂度越低越好

■ 数据库负载:随着事实表数据量增大,数据库吞吐量能否满足,数据库负载能否支撑

■ 维表更新实时性:维表更新后,新的数据能否及时被应用到

■ 内存消耗:是否占用太多内存

■ 横向扩展:随着数据量增大,能否横向扩展

■ 结果确定性:结果的正确性是否能够保证

 

01 实时lookup数据库关联

 

所谓实时lookup数据库关联,就是在用户自定义函数中通过关联字段直接访问数据库实现关联的方式。每条事实表数据都会根据关联键,到存储维度表的数据库中查询一次。

​编辑

 

实时lookup数据库关联的特点是实现简单,但数据库压力较大,无法支撑大数据量的维度数据查询,并且在查询时只能根据当时的维度表数据查询,如果事实表数据重放或延迟,查询结果的正确性无法得到保证,且多次查询结果可能不一致。

 

实时lookup数据库关联还可以再细分为三种方式:同步lookup数据库关联、异步lookup数据库关联和带缓存的数据库lookup关联。

 

 

1.1 同步lookup数据库关联

 

同步实时数据库lookup关联实现最简单,只需要在一个RichMapFunction或者RichFlat-MapFunction中访问数据库,处理好关联逻辑后将结果数据输出即可。上游每输入一条数据就会前往外部表中查询一次,等待返回后输出关联结果。

 

同步lookup数据库关联的参考代码如下:

 

创建类并继承RichMapFunction抽象类。

public class HBaseMapJoinFun extends RichMapFunction<Tuple2<String,String>,Tuple3<String,String,String>> {

 

在open方法中实现连接数据库(该数据库存储了维度表信息)。

public void open(Configuration parameters) throws Exception {
    org.apache.hadoop.conf.Configuration hconf= HBaseConfiguration.create();
    InputStream hbaseConf = DimSource.class.getClassLoader().getResourceAsStream("hbase-site.xml");
    InputStream hdfsConf = DimSource.class.getClassLoader().getResourceAsStream("hdfs-site.xml");
    InputStream coreConf = DimSource.class.getClassLoader().getResourceAsStream("core-site.xml");
    hconf.addResource(hdfsConf);
    hconf.addResource(hbaseConf);
    hconf.addResource(coreConf);
    if (User.isHBaseSecurityEnabled(hconf)){
        String userName = "dl_rt";
        String keyTabFile = "/opt/kerberos/kerberos-keytab/keytab";
        LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, userName, keyTabFile);
    }else {
        LOG.error("conf load error!");
    }
    connection = ConnectionFactory.createConnection(hconf);
}

 

在map方法中实现关联操作,并返回结果。

@Override
public Tuple3<String, String, String> map(Tuple2<String, String> stringStringTuple2) throws Exception 
    LOG.info("Search hbase data by key .");
    String row_key = stringStringTuple2.f1;
    String p_name = stringStringTuple2.f0;
    byte[] familyName = Bytes.toBytes("cf");
    byte[] qualifier = Bytes.toBytes("city_name");
    byte[] rowKey = Bytes.toBytes(row_key);
    table = connection.getTable(TableName.valueOf(table_name));
    Get get = new Get(rowKey);
    get.addColumn(familyName,qualifier);
    Result result = table.get(get);
    for (Cell cell : result.rawCells()){
        LOG.info("{}:{}:{}",Bytes.toString(CellUtil.cloneRow(cell)),Bytes.toString(CellUtil.cloneFamily(cell)),
            Bytes.toString(CellUtil.cloneQualifier(cell)),
            Bytes.toString(CellUtil.cloneValue(cell)));
    }
    String cityName = Bytes.toString(result.getValue(Bytes.toBytes("cf"),Bytes.toBytes("city_name")));
    return new Tuple3<String, String, String>(row_key,p_name,cityName);
}

 

在主类中调用。

//关联维度表
SingleOutputStreamOperator<Tuple3<String,String,String>> resultStream = dataSource.map(new HBaseMapJoinFun());
resultStream.print().setParallelism(1);

 

 

1.2 异步lookup数据库关联

 

异步实时数据库lookup关联需要借助AsyncIO来异步访问维表数据。AsyncIO可以充分利用数据库提供的异步Client库并发处理lookup请求,提高Task并行实例的吞吐量。

 

相较于同步lookup,异步方式可大大提高数据库查询的吞吐量,但相应的也会加大数据库的负载,并且由于查询只能查当前时间点的维度数据,因此可能造成数据查询结果的不准确。

 

​编辑

 

AsyncIO提供lookup结果的有序和无序输出,由用户自己选择是否保证event的顺序。

 

示例代码参考如下:

 

创建Join类并继承RichAsyncFunction抽象类。

public class HBaseAyncJoinFun extends RichAsyncFunction<Tuple2<String,String>, Tuple3<String,String,String>> {

 

在open方法中实现连接数据库(存储了维度表的信息)。

public void open(Configuration parameters) throws Exception {
    org.apache.hadoop.conf.Configuration hconf= HBaseConfiguration.create();
    InputStream hbaseConf = DimSource.class.getClassLoader().getResourceAsStream("hbase-site.xml");
    InputStream hdfsConf = DimSource.class.getClassLoader().getResourceAsStream("hdfs-site.xml");
    InputStream coreConf = DimSource.class.getClassLoader().getResourceAsStream("core-site.xml");
    hconf.addResource(hdfsConf);
    hconf.addResource(hbaseConf);
    hconf.addResource(coreConf);
    if (User.isHBaseSecurityEnabled(hconf)){
        String userName = "dl_rt";
        String keyTabFile = "/opt/kerberos/kerberos-keytab/keytab";
        LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, userName, keyTabFile);
    }else {
        LOG.error("conf load error!");
    }
    final ExecutorService threadPool = Executors.newFixedThreadPool(2,
        new ExecutorThreadFactory("hbase-aysnc-lookup-worker", Threads.LOGGING_EXCEPTION_HANDLER));
    try{
        connection = ConnectionFactory.createAsyncConnection(hconf).get();
        table=connection.getTable(TableName.valueOf(table_name),threadPool);
    }catch (InterruptedException | ExecutionException e){
        LOG.error("Exception while creating connection to HBase.",e);
        throw new RuntimeException("Cannot create connection to HBase.",e);
    }

 

在AsyncInvoke方法中实现异步关联,并返回结果。

@Override
public void asyncInvoke(Tuple2<String, String> input, ResultFuture<Tuple3<String, String, String>> resultFuture) throws Exception {
    LOG.info("Search hbase data by key .");
    String row_key = input.f1;
    String p_name = input.f0;
    byte[] familyName = Bytes.toBytes("cf");
    byte[] qualifier = Bytes.toBytes("city_name");
    byte[] rowKey = Bytes.toBytes(row_key);
    Get get = new Get(rowKey);
    get.addColumn(familyName,qualifier);
    CompletableFuture<Result> responseFuture = table.get(get);
    responseFuture.whenCompleteAsync(
        (result, throwable) -> {
            if (throwable != null){
                if (throwable instanceof TableNotFoundException){
                    LOG.error("Table '{}' not found", table_name,throwable);
                    resultFuture.completeExceptionally(
                        new RuntimeException("HBase table '"+table_name+"' not found.",throwable)
                    );
                }else {
                    LOG.error(String.format("HBase asyncLookup error,retry times = %d",1),throwable);
                    responseFuture.completeExceptionally(throwable);
                }
            }else{
                List list = new ArrayList<Tuple3<String, String, String>>();
                if (result.isEmpty()){
                    String cityName="";
                    list.add(new Tuple3<String,String,String>(row_key,p_name,cityName));
                    resultFuture.complete(list);
                }else{
                    String cityName = Bytes.toString(result.getValue(Bytes.toBytes("cf"),Bytes.toBytes("city_name")));
                    list.add(new Tuple3<String,String,String>(row_key,p_name,cityName));
                    resultFuture.complete(list);
                }
            }
        }
    );


}

 

在主方法中调用。

//异步关联维度表
DataStream<Tuple3<String,String,String>> unorderedResult = AsyncDataStream.unorderedWait(dataSource, new HBaseAyncJoinFun(),
    5000L, TimeUnit.MILLISECONDS,2).setParallelism(2);
unorderedResult.print();

 

此处使用unorderedWait方式,允许返回结果存在乱序。

 

 

1.3 带缓存的数据库lookup关联

 

带缓存的数据库lookup关联是对上述两种方式的优化,通过增加缓存机制来降低查询数据库的请求数量,而且缓存不需要通过 Checkpoint 机制持久化,可以采用本地缓存,例如Guava Cache可以比较轻松的实现。

 

此种方式的问题在于缓存的数据无法及时更新,可能会造成关联数据不正确的问题。

 

​编辑

 

02 预加载维表关联

 

预加载维表关联是在作业启动时就把维表全部加载到内存中,因此此种方式只适用于维度表数据量不大的场景。相较于lookup方式,预加载维表可以获得更好的性能。

 

预加载维表关联还可以再细分为四种方式:启动时预加载维表、启动时预加载分区维表、启动时预加载维表并定时刷新和启动时预加载维表并实时lookup数据库。

 

预加载维表的各种细分方案可根据实际应用场景进行结合应用,以此来满足不同的场景需求。

 

 

2.1 启动时预加载维表

 

启动时预加载维表实现比较简单,作业初始化时,在用户函数的open方法中读取数据库的维表数据放到内存中,且缓存的维表数据不作为State,每次重启时open方法都被再次执行,从而加载新的维表数据。

​编辑

 

此方法需要占用内存来存储维度表数据,不支持大数据量的维度表,且维度表加载入内存后不能实时更新,因此只适用于对维度表更新要求不高且数据量小的场景。

 

 

2.2 启动时预加载分区维表

 

对于维表比较大的情况,可以在启动预加载维表基础之上增加分区功能。简单来说就是将数据流按字段进行分区,然后每个Subtask只需要加在对应分区范围的维表数据。此种方式一定要自定义分区,不要用KeyBy。

​编辑

 

 

2.3 启动时预加载维表并定时刷新

 

预加载维度数据只有在Job启动时才会加载维度表数据,这会导致维度数据变更无法被识别,在open方法中初始化一个额外的线程来定时更新内存中的维度表数据,可以一定程度上缓解维度表更新问题,但无法彻底解决。

​编辑

 

示例代码参考如下:

public class ProLoadDimMap extends RichMapFunction<Tuple2<String,Integer>,Tuple2<String,String>> {
    private static final Logger LOG = LoggerFactory.getLogger(ProLoadDimMap.class.getName());


    ScheduledExecutorService executor = null;
    private Map<String,String> cache;


    @Override
    public void open(Configuration parameters) throws Exception {
        executor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    load();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },5,5, TimeUnit.MINUTES);//每隔 5 分钟拉取一次维表数据
    }


    @Override
    public void close() throws Exception {


    }
    @Override
    public Tuple2<String, String> map(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
        String username = stringIntegerTuple2.f0;
        Integer city_id = stringIntegerTuple2.f1;
        String cityName = cache.get(city_id.toString());
        return new Tuple2<String,String>(username,cityName);
    }


    public void load() throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:mysql://172.XX.XX.XX:XX06/yumd?useSSL=false&characterEncoding=UTF-8", "root", "Root@123");
        PreparedStatement statement = con.prepareStatement("select city_id,city_name from city_dim;");
        ResultSet rs = statement.executeQuery();
        //全量更新维度数据到内存
        while (rs.next()) {
            String cityId = rs.getString("city_id");
            String cityName = rs.getString("city_name");
            cache.put(cityId, cityName);
        }
        con.close();
    }
}

 

 

2.4 启动时预加载维表并实时lookup数据库

 

此种方案就是将启动预加载维表和实时look两种方式混合使用,将预加载的维表作为缓存给实时lookup使用,未命中则到数据库里查找。该方案可解决关联不上的问题。

​编辑

 

03 广播维度表

 

广播维度表方案是将维度表数据用流的方式接入Flink Job 程序,并将维度表数据进行广播,再与事件流数据进行关联,此种方式可以及时获取维度表的数据变更,但因数据保存在内存中,因此支持的维度表数据量较小。

 

示例代码参考如下:

 

首先将维度表进行广播。

//维度数据源
DataStream<Tuple2<Integer,String>> dimSource = env.addSource(new DimSource1());


// 生成MapStateDescriptor
MapStateDescriptor<Integer,String> dimState = new MapStateDescriptor<Integer, String>("dimState",
    BasicTypeInfo.INT_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO);
BroadcastStream<Tuple2<Integer,String>> broadcastStream = dimSource.broadcast(dimState);

 

实现BroadcastProcessFunction类的processElement方法处理事实流与广播流的关联,并返回关联结果。

SingleOutputStreamOperator<String> output = dataSource.connect(broadcastStream).process(
    new BroadcastProcessFunction<Tuple2<String, Integer>, Tuple2<Integer, String>, String>() {
        @Override
        public void processElement(Tuple2<String, Integer> input, ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {
            ReadOnlyBroadcastState<Integer,String> state = readOnlyContext.getBroadcastState(dimState);
            String name = input.f0;
            Integer city_id = input.f1;
            String city_name="NULL";
            if (state.contains(city_id)){
                city_name=state.get(city_id);
                collector.collect("result is : "+name+" ,"+city_id+" ,"+city_name);
            }
        }

 

实现BroadcastProcessFunction类的processBroadcastElement方法处理广播流数据,将新的维度表数据进行广播。

@Override
public void processBroadcastElement(Tuple2<Integer, String> input, Context context, Collector<String> collector) throws Exception {
    LOG.info("收到广播数据:"+input);
    context.getBroadcastState(dimState).put(input.f0,input.f1);
}

 

 

04 Temporal Table Function Join

 

 

Temporal Table Function Join仅支持在Flink SQL API中使用,需要将维度表数据作为流的方式传入Flink Job。该种方案可支持大数据量的维度表,且维度表更新及时,关联数据准确性更高,缺点是会占用状态后端和内存的资源,同时自行实现的代码复杂度过高。

 

Temporal Table是持续变化表上某一时刻的视图,Temporal Table Function是一个表函数,传递一个时间参数,返回Temporal Table这一指定时刻的视图。可以将维度数据流映射为Temporal Table,主流与这个Temporal Table进行关联,可以关联到某一个版本(历史上某一个时刻)的维度数据。

​编辑

 

示例代码参考如下:

public class TemporalFunTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, bsSettings);
        env.setParallelism(1);


        //定义主流
        DataStream<Tuple3<String,Integer,Long>> dataSource = env.addSource(new EventSource2())
            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String,Integer,Long>>(Time.seconds(0)){
            @Override
            public long extractTimestamp(Tuple3<String, Integer, Long> stringIntegerLongTuple3) {
                return stringIntegerLongTuple3.f2;
            }
        });


        //定义维度流
        DataStream<Tuple3<Integer, String, Long>> cityStream = env.addSource(new DimSource())
            .assignTimestampsAndWatermarks(
                //指定水位线、时间戳
                new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Integer, String, Long>>(Time.seconds(0)) {
                    @Override
                    public long extractTimestamp(Tuple3<Integer, String, Long> element) {
                        return element.f2;
                    }
                });
        //主流,用户流, 格式为:user_name、city_id、ts
        Table userTable = tableEnv.fromDataStream(dataSource,"user_name,city_id,ts.rowtime");
        //定义城市维度流,格式为:city_id、city_name、ts
        Table cityTable = tableEnv.fromDataStream(cityStream,"city_id,city_name,ts.rowtime");
        tableEnv.createTemporaryView("userTable", userTable);
        tableEnv.createTemporaryView("cityTable", cityTable);


        //定义一个TemporalTableFunction
        TemporalTableFunction dimCity = cityTable.createTemporalTableFunction("ts", "city_id");
        //注册表函数
        tableEnv.registerFunction("dimCity", dimCity);


        Table u = tableEnv.sqlQuery("select * from userTable");
        u.printSchema();
        tableEnv.toAppendStream(u, Row.class).print("user streaming receive : ");


        Table c = tableEnv.sqlQuery("select * from cityTable");
        c.printSchema();
        tableEnv.toAppendStream(c, Row.class).print("city streaming receive : ");


        //关联查询
        Table result = tableEnv
            .sqlQuery("select u.user_name,u.city_id,d.city_name,u.ts " +
                "from userTable as u " +
                ", Lateral table (dimCity(u.ts)) d " +
                "where u.city_id=d.city_id");
        //打印输出
        DataStream resultDs = tableEnv.toAppendStream(result, Row.class);
        resultDs.print("\t\t join result out:");
        env.execute("joinDemo");


    }
}

 

最后,总结各种维度表关联方案的特点如下:​编辑

 

 

标签:city,name,数据库,Flink,关联,金信,维表,维度
From: https://www.cnblogs.com/zhongdianjinxin/p/18075041

相关文章

  • 代码随想录算法训练营第七天 | 454.四数相加II 383. 赎金信 15. 三数之和 18. 四数之
    day7记录代码随想录第一题力扣454.四数相加II 给定四个包含整数的数组列表 A,B,C,D,计算有多少个元组(i,j,k,l) ,使得 A[i]+B[j]+C[k]+D[l]=0。为了使问题简单化,所有的A,B,C,D具有相同的长度 N,且0≤N≤500。所有整数的范围在-2^28到......
  • Flink实时写Hudi报NumberFormatException异常
    Flink实时写Hudi报NumberFormatException异常问题描述在Flink项目中,针对Hudi表xxxx_table的bucket_write操作由于java.lang.NumberFormatException异常而从运行状态切换到失败状态。异常信息显示在解析字符串"ddd7a1ec"为整数时出现了问题。报错如下:bucket_write:......
  • 【快捷部署】002_Flink
    Flink一键安装(本地模式)install-flink.sh脚本内容#!/bin/bash####变量###执行脚本的当前目录mydir=$(cd"$(dirname"$0")";pwd)echo$mydir#flink安装目录flink=/flink#检查点目录cp=$flink/checkpoints/#保留点目录sp=$flink/savepoints/#tasknumber数量ta......
  • 张量维度改变总结
    文章目录一、view()或reshape()二、unsqueeze()三、squeeze()四、transpose()五、torch.expand_dims一、view()或reshape()  view()或reshape():这两个函数可以用于改变张量的形状,但保持元素总数不变。它们可以接受一个新的形状作为参数,并返回一个新的张量......
  • MindSpore自定义算子中的张量维度问题
    技术背景在前面的几篇博客中,我们介绍了MindSpore框架下使用CUDA来定义本地算子的基本方法,以及配合反向传播函数的使用,这里主要探讨一下MindSpore框架对于CUDA本地算子的输入输出的规范化形式。测试思路MindSpore使用的CUDA算子规范化接口形式为:extern"C"intCustomOps(intn......
  • flink部署模式和运行模式
    flink部署模式部署模式:flink里面的计算程序运行的方式sessionsession模式一个flink集群可以跑多个计算任务,资源共享session模式下集群是提前启动的,然后向flink集群提交jobper-job(高版本已经不推荐了)per-job模式下,一个集群只跑一个计算任务,资源独立,集群的启动是跟随......
  • 代码随想录算法训练营第七天| 454. 四数相加 II 383. 赎金信
    454.四数相加IIhttps://leetcode.cn/problems/4sum-ii/description/、publicintfourSumCount(int[]nums1,int[]nums2,int[]nums3,int[]nums4){intres=0;HashMap<Integer,Integer>map=newHashMap<>();for(inti:nu......
  • centos7安装flink
    local模式环境说明,flink需要jdk,并且flin.2k1.17,需要的是jdk11,jdk17不行,实测jdk1.8也行下载flink包wgethttps://dlcdn.apache.org解压#解压tar-zxvfflink-1.17.2-bin-scala_2.12.tgz#进入flink目录cdflink-1.17.2修改配置文件viconf/flink-conf.yaml#允......
  • 中电金信:我们为什么需要单元化架构?
    导语:随着数字化时代的不断发展,金融企业计算机系统也经历了多次变革,从辅助类单机单库系统到全分布式的数字化企业架构。经过十几年的高速发展,金融企业的系统架构建设无论是在性能方面还是在复杂度方面都产生了巨大的变化。而在探索建设金融企业系统架构的道路上,很多金融企业都不约......
  • Flink CDC简介-flinkcdc-jian-jie
    FlinkCDC官方文档什么是FlinkCDC¶FlinkCDCConnectors是ApacheFlink的一组源连接器,使用变更数据捕获(CDC)从不同数据库中获取变更。FlinkCDCConnectors集成Debezium作为捕获数据变化的引擎。所以它可以充分发挥Debezium的能力。详细了解Debezium是什么。支......