flink 维表关联、翻译字典-预加载维表

作者：马育民 • • 阅读：10153

上接：flink 维表关联、翻译字典

介绍

启动服务时，将外部数据库（MySQL、redis）中的数据，全部查询出来，并放入到内存中（如：放入到Map）

适用场景

维表数据少，变更频率低，对变更及时性要求低

优点

  • 实现简单
  • 一次性访问外部数据库，将数据放入到内存，之后关联时不需要查询外部数据库，大大提高了效率，减轻外部数据库的压力

缺点

  • 因为数据存于内存，对计算节点的内存消耗很高，所以只适合 小数据量
  • 维表数据发生更新，Flink 是无法感知的，会出现维表数据不一致，针对这种情况我们可以采取 定时拉取维表数据（但是更新会有延时）

实现

定义类继承 RichMapFunction，在 open() 中读取维表数据加载到内存中，在 map() 方法中与维表数据进行关联。

例子-简单实现

以订单数据为例，根据下单用户的 城市 ID，去关联查询 城市名称

简单实现，查询MySQL数据库，并将数据放入到 Map 中

缺点：维表数据发生更新，Flink 是无法感知的，会出现维表数据不一致

建表

CREATE TABLE `t_city` (
    `cid` INT(11) NOT NULL,
    `cname` VARCHAR(64) NOT NULL COLLATE 'utf8mb4_general_ci',
    PRIMARY KEY (`cid`)
)
ENGINE=InnoDB
;

添加测试数据

INSERT INTO `t_city` (`cid`, `cname`) VALUES (1, '北京');
INSERT INTO `t_city` (`cid`, `cname`) VALUES (2, '黑龙江');
INSERT INTO `t_city` (`cid`, `cname`) VALUES (3, '吉林2');

依赖

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <flink.version>1.12.0</flink.version>
</properties>

<dependencies>
    <!--java-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-statebackend-rocksdb_2.12</artifactId>
        <version>1.12.0</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.47</version>
    </dependency>
</dependencies>

java 代码

package flink;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class 预加载 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> dss = env.fromElements(
                "1", // 1表示北京
                "2", // 2 表示黑龙江
                "3"); // 3 表示吉林

        SingleOutputStreamOperator soso = dss.map(new MyRichMapFunction());

        soso.print("soso::");
        env.execute();
    }
    static class MyRichMapFunction extends RichMapFunction<String,String>{
        // MySQL数据库连接
        static final String URL = "jdbc:mysql://localhost:3308/test?characterEncoding=UTF-8";
        // 数据库用户名
        static final String USERNAME = "root";
        // 数据库密码
        static final String PASSWORD = "";
        // 驱动
        static final String DRIVER = "com.mysql.jdbc.Driver";

        //数据库连接
        private Connection conn;

        private Map<String,String> cache;

        @Override
        public void open(Configuration parameters) throws Exception {
            cache = new HashMap<>();
            Class.forName(DRIVER);
            conn = DriverManager.getConnection(URL, USERNAME, PASSWORD);
            load(); // 立即查询数据库，并加数据放入到cache中
        }

        // 查询MySQL数据库，并将数据放入到 Map 中
        public void load() throws Exception {
            String sql = "select cid,cname from t_city";
            PreparedStatement statement = null;
            ResultSet rs = null;
            try {
                statement = conn.prepareStatement(sql);
                rs = statement.executeQuery();
                //全量更新维度数据到内存
                while (rs.next()) {
                    String id = rs.getString(1);
                    String name = rs.getString(2);
                    cache.put(id, name);
                    System.out.println(String.format("查询结果：%s-%s", id, name));
                }
            }finally {
                if(rs!=null) {
                    rs.close();
                }
                if(statement!=null) {
                    statement.close();
                }
            }
        }

        @Override
        public void close() throws Exception {
            // 关闭数据库连接
            if(conn!=null) {
                conn.close();
            }
        }

        @Override
        public String map(String value) throws Exception {
            String name = cache.get(value);
            if(name == null){
                name = "";
            }
            return name;
        }
    }
}

例子-定时加载

为了解决上面的缺点，这里使用 ScheduledExecutorService 类，定时拉取维表数据

本例从 kafka 读取数据

依赖

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <flink.version>1.12.0</flink.version>
</properties>

<dependencies>
    <!--java-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-statebackend-rocksdb_2.12</artifactId>
        <version>1.12.0</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.47</version>
    </dependency>
    <!-- kafka Dependency -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

java 代码

package flink;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class 预加载_定时 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        DataStreamSource<String> dss = env.addSource(fkc);

        SingleOutputStreamOperator soso = dss.map(new MyRichMapFunction());

        soso.print("soso--");
        env.execute();
    }
    static class MyRichMapFunction extends RichMapFunction<String,String>{
        // MySQL数据库连接
        static final String URL = "jdbc:mysql://localhost:3308/test?characterEncoding=UTF-8";
        // 数据库用户名
        static final String USERNAME = "root";
        // 数据库密码
        static final String PASSWORD = "";
        // 驱动
        static final String DRIVER = "com.mysql.jdbc.Driver";

        //数据库连接
        private Connection conn;

        //定时任务
        private ScheduledExecutorService scheduled;

        private Map<String,String> cache;

        @Override
        public void open(Configuration parameters) throws Exception {
            cache = new HashMap<>();
            Class.forName(DRIVER);
            conn = DriverManager.getConnection(URL, USERNAME, PASSWORD);
            load(); // 立即查询数据库，并加数据放入到cache中
            // 定时查询数据库，并加数据放入到cache中
            scheduled = Executors.newSingleThreadScheduledExecutor();
            scheduled.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        load(); // 定时调用load()方法查询数据库，并加数据放入到cache中
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            },5,5, TimeUnit.SECONDS);//延迟5秒执行第一次；然后每间隔5秒执行一次
        }

        // 查询MySQL数据库，并将数据放入到 Map 中
        public void load() throws Exception {
            SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
            System.out.println(sdf.format(new Date())+"  查询数据库");
            String sql = "select cid,cname from t_city";
            PreparedStatement statement = null;
            ResultSet rs = null;
            try {
                statement = conn.prepareStatement(sql);
                rs = statement.executeQuery();
                //全量更新维度数据到内存
                while (rs.next()) {
                    String id = rs.getString(1);
                    String name = rs.getString(2);
                    cache.put(id, name);
                    System.out.println(String.format("查询结果：%s-%s", id, name));
                }
            }finally {
                if(rs!=null) {
                    rs.close();
                }
                if(statement!=null) {
                    statement.close();
                }
            }
        }

        @Override
        public void close() throws Exception {
            // 关闭数据库连接
            if(conn!=null) {
                conn.close();
            }
            // 终止定时任务
            if(scheduled!=null){
                scheduled.shutdown();
            }
        }

        @Override
        public String map(String value) throws Exception {
            String name = cache.get(value);
            if(name == null){
                name = "";
            }
            return name;
        }
    }
}

启动zk

登录 hadoop1，执行下面命令：

/program/bin/zk.sh start

启动kafka

登录 hadoop1，执行下面命令：

/program/bin/kafka.sh start

启动 flink

在 idea 中运行 flink 程序

启动生产者、指定主题

登录 hadoop1，执行下面命令：

/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic kafka2flink

然后输入：

>1
>2
>3

执行结果

在 idea 控制台中打印结果如下：

09:48:20  查询数据库
查询结果：1-北京
查询结果：2-黑龙江
查询结果：3-吉林

soso--> 北京
soso--> 黑龙江
soso--> 吉林

修改MySQL数据

MySQL 修改数据，将 吉林 修改成 吉林2，过5秒后，自动查询并将数据放入到 Map 中，查询出的结果是 吉林2

原文出处：/show_1IX4mYGfHp88.html

吉ICP备15003043号