一：DIM层设计要点
（1）DIM层的设计依据是维度建模理论，该层存储维度模型的维度表。
（2）DIM层的数据存储在 HBase 表中。
DIM 层表是用于维度关联的，要通过主键去获取相关维度信息，这种场景下 K-V 类型数据库的效率较高。常见的 K-V 类型数据库有 Redis、HBase，而 Redis 的数据常驻内存，会给内存造成较大压力，因而选用 HBase 存储维度数据。
（3）DIM层表名的命名规范为**dim_**表名
二：DIM层大概实操流程
2.1 读取数据
Kafka---topic_db(包含所有的46张业务表)
2.2 过滤数据
过滤出所需要的维表数据
过滤条件:在代码中给定十几张维表的表名
问题:如果增加维表,需要修改代码-重新编译-打包-上传、重启任务
优化1:不修改代码、只重启任务
配置信息中保存需要的维表信息,配置信息只在程序启动的时候加载一次
优化2:不修改代码、不重启任务
方向:让程序在启动以后还可以获取配置信息中增加的内容
具体实施:
1) 定时任务:每隔一段时间加载一次配置信息
将定时任务写在Open方法
2) 监控配置信息:一旦配置信息增加了数据,可以立马获取到
(1) MySQLBinlog:FlinkCDC监控直接创建流
a.将配置信息处理成广播流:缺点 -> 如果配置信息过大,冗余太多
b.按照表名进行KeyBy处理:缺点 -> 有可能产生数据倾斜
(2) 文件:Flume->Kafka->Flink消费创建流
2.3 写出数据
将数据写出到Phoenix、JdbcSink、自定义Sink
三：配置表
本层的任务是将业务数据直接写入到不同的 HBase 表中。那么如何让程序知道流中的哪些数据是维度数据？维度数据又应该写到 HBase 的哪些表中？为了解决这个问题，我们选择在 MySQL 中构建一张配置表，通过 Flink CDC 将配置表信息读取到程序中。
3.1 配置表设计
1）字段解析
我们将为配置表设计五个字段
- source_table：作为数据源的业务数据表名
- sink_table：作为数据目的地的 Phoenix 表名
- sink_columns：Phoenix 表字段
- sink_pk：Phoenix 表主键
- sink_extend：Phoenix 建表扩展，即建表时一些额外的配置语句
将** source_table 作为配置表的主键**，可以通过它获取唯一的目标表名、字段、主键和建表扩展，从而得到完整的 Phoenix 建表语句。
** 数据格式：**
{"before":null,"after":
{"source_table":"aa","sink_table":"bb","sink_columns":"cc","sink_pk":"id","sink_extend":"xxx"},"source":
{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":165251303
9549,"snapshot":"false","db":"gmall-211126-
config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0
,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1652513039551,"transaction":null}
2）在Mysql中创建数据库建表并开启Binlog
（1）创建数据库 gmall_config ，注意:和 gmall 业务库区分开
[atguigu@hadoop102 db_log]$ mysql -uroot -p000000 -e"create database gmall_config charset
utf8 default collate utf8_general_ci"
（2）在 gmall_config 库中创建配置表 table_process
CREATE TABLE `table_process` (
`source_table` varchar(200) NOT NULL COMMENT '来源表',
`sink_table` varchar(200) DEFAULT NULL COMMENT '输出表',
`sink_columns` varchar(2000) DEFAULT NULL COMMENT '输出字段',
`sink_pk` varchar(200) DEFAULT NULL COMMENT '主键字段',
`sink_extend` varchar(200) DEFAULT NULL COMMENT '建表扩展',
PRIMARY KEY (`source_table`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
（3）在MySQL配置文件中增加 gmall_config 开启Binlog
[axing@hadoop107 ~]$ sudo vim /etc/my.cnf
（4）为了方便测试，目前就插入两张表名数据，作为维度表
四：实操流程
4.1 接收Kafka数据，过滤空值数据
对Maxwell抓取的数据进行ETL，有用的部分保留，没用的过滤掉。
4.2 动态拆分维度表功能
由于Maxwell是把全部数据统一写入一个Topic中, 这样显然不利于日后的数据处理。所以需要把各个维度表拆开处理。
在实时计算中一般把维度数据写入存储容器，一般是方便通过主键查询的数据库比如HBase,Redis,MySQL等。
这样的配置不适合写在配置文件中，因为这样的话，业务端随着需求变化每增加一张维度表表，就要修改配置重启计算程序。
所以这里需要一种动态配置方案，把这种配置长期保存起来，一旦配置有变化，实时计算可以自动感知。这种可以有三个方案实现：
一种是用Zookeeper存储，通过Watch感知数据变化；
另一种是用mysql数据库存储，周期性的同步；
再一种是用mysql数据库存储，使用广播流。
这里选择第三种方案，主要是MySQL对于配置数据初始化和维护管理，使用FlinkCDC读取配置信息表，将配置流作为广播流与主流进行连接。
4.3 把流中的数据保存到对应的维度表
维度数据保存到HBase的表中。
五：具体代码实现
5.1 接收Kafka数据，过滤空值数据
1**）创建 KafkaUtil **工具类
和 Kafka 交互要用到 Flink 提供的 FlinkKafkaConsumer、FlinkKafkaProducer 类，为了提高模板代码的复用性，将其封装到 KafkaUtil 工具类中。
此处从 Kafka 读取数据，创建 getKafkaConsumer(String topic, String groupId) 方法
public class KafkaUtil {
static String BOOTSTRAP_SERVERS = "hadoop102:9092, hadoop103:9092, hadoop104:9092";
static String DEFAULT_TOPIC = "default_topic";
public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId) {
Properties prop = new Properties();
prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(topic,
new KafkaDeserializationSchema<String>() {
@Override
public boolean isEndOfStream(String nextElement) {
return false;
}
@Override
public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
if(record != null && record.value() != null) {
return new String(record.value());
}
return null;
}
@Override
public TypeInformation<String> getProducedType() {
return TypeInformation.of(String.class);
}
}, prop);
return consumer;
}
}
2****）主程序
package com.atguigu.app.dim;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.TableProcessFunction;
import com.atguigu.bean.TableProcess;
import com.atguigu.utils.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
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.util.Collector;
public class DimApp {
public static void main(String[] args) throws Exception {
//1.获取执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);//生产环境，并行度应设置为kafka主题的分区数
/*
//生产环境下使用：
//1.1 开启checkpoint
env.enableCheckpointing(5*6000L, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointTimeout(10*6000L);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
//1.2 设置状态后端
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop107:8020/211126/ck");
System.setProperty("HADOOP_USER_NAME","atguigu");
*/
//2.读取kafka topic_db主题数据创建主流
String topic ="topic_db";
String groupId = "dim_app_211126";
DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
//3.过滤掉非JSON数据以及保留新增、变化以及初始化数据并将数据转换为JSON格式
SingleOutputStreamOperator<JSONObject> filterJsonDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
@Override
public void flatMap(String value, Collector<JSONObject> collector) throws Exception {
try {
//将数据装换为JSON格式
JSONObject jsonObject = JSON.parseObject(value);
//获取数据中的操作类型字段
String type = jsonObject.getString("type");
//保留新增、变化、以及初始化数据
if ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type)) {
collector.collect(jsonObject);
}
} catch (Exception e) {
System.out.println("脏数据：" + value);//或者写入侧输出流
}
}
});
//4.使用FlinkCDC读取mysql配置信息表创建配置流
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
.hostname("hadoop107")
.port(3306)
.username("root")
.password("000000")
.databaseList("gmall-config")
.tableList("gmall-config.table_process")
.startupOptions(StartupOptions.initial())
.deserializer(new JsonDebeziumDeserializationSchema())
.build();
DataStreamSource<String> mysqlSourceDS = env.fromSource(mySqlSource,
WatermarkStrategy.noWatermarks(),
"MysqlSource"
);
//5.将配置流处理为广播流
MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("map-state", String.class, TableProcess.class);
BroadcastStream<String> broadcastStream = mysqlSourceDS.broadcast(mapStateDescriptor);
//6.连接主流和广播流
BroadcastConnectedStream<JSONObject, String> connectedStream = filterJsonDS.connect(broadcastStream);
//7.处理连接流，根据配置信息处理主流数据
SingleOutputStreamOperator<JSONObject> dimDS = connectedStream.process(new TableProcessFunction(mapStateDescriptor));
//8.将数据写出到Phoenix
dimDS.print(">>>>>");
//9.启动任务
env.execute();
}
}
5.2 根据MySQL的配置表，动态进行分流
1）导入依赖
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>2.1.0</version>
</dependency>
<dependency>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix-spark</artifactId>
<version>5.0.0-HBase-2.0</version>
<exclusions>
<exclusion>
<groupId>org.glassfish</groupId>
<artifactId>javax.el</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- 如果不引入 flink-table 相关依赖，则会报错：
Caused by: java.lang.ClassNotFoundException:
org.apache.flink.connector.base.source.reader.RecordEmitter
引入以下依赖可以解决这个问题（引入某些其它的 flink-table相关依赖也可）
-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.12</artifactId>
<version>1.13.0</version>
</dependency>
2****）创建配置表实体类
package com.atguigu.gmall.realtime.bean;
import lombok.Data;
@Data
public class TableProcess {
//来源表
String sourceTable;
//输出表
String sinkTable;
//输出字段
String sinkColumns;
//主键字段
String sinkPk;
//建表扩展
String sinkExtend;
}
3）编写操作读取配置表形成广播流
// TODO 6. FlinkCDC 读取配置流并广播流
// 6.1 FlinkCDC 读取配置表信息
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
.hostname("hadoop102")
.port(3306)
.databaseList("gmall_config") // set captured database
.tableList("gmall_config.table_process") // set captured table
.username("root")
.password("000000")
.deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
.startupOptions(StartupOptions.initial())
.build();
// 6.2 封装为流
DataStreamSource<String> mysqlDSSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource");
// 6.3 广播配置流
MapStateDescriptor<String, TableProcess> tableConfigDescriptor = new MapStateDescriptor<String, TableProcess>("table-process-state", String.class, TableProcess.class);
BroadcastStream<String> broadcastDS = mysqlDSSource.broadcast(tableConfigDescriptor);
// TODO 7. 连接流
BroadcastConnectedStream<JSONObject, String> connectedStream = filterDS.connect(broadcastDS);
4）定义一个项目中常用的配置常量类GmallConfig
package com.atguigu.gmall.realtime.common;
public class GmallConfig {
// Phoenix库名
public static final String HBASE_SCHEMA = "GMALL2022_REALTIME";
// Phoenix驱动
public static final String PHOENIX_DRIVER = "org.apache.phoenix.jdbc.PhoenixDriver";
// Phoenix连接参数
public static final String PHOENIX_SERVER = "jdbc:phoenix:hadoop102,hadoop103,hadoop104:2181";
}
5）自定义函数****MyBroadcastFunction
（1）定义类MyBroadcastFunction
package com.atguigu.gmall.realtime.app.func;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
public class MyBroadcastFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
private MapStateDescriptor<String, TableProcess> tableConfigDescriptor;
public MyBroadcastFunction(MapStateDescriptor<String, TableProcess> tableConfigDescriptor) {
this.tableConfigDescriptor = tableConfigDescriptor;
}
@Override
public void processElement(JSONObject jsonObj, ReadOnlyContext readOnlyContext, Collector<JSONObject> out) throws Exception {
}
@Override
public void processBroadcastElement(String jsonStr, Context context, Collector<JSONObject> out) throws Exception {
}
}
（2）自定义函数MyBroadcastFunction-open
// 定义Phoenix的连接
private Connection conn;
@Override
public void open(Configuration parameter) throws Exception {
super.open(parameter);
Class.forName(GmallConfig.PHOENIX_DRIVER);
conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
}
（3）自定义函数MyBroadcastFunction-processBroadcastElement
@Override
public void processBroadcastElement(String jsonStr, Context context, Collector<JSONObject> out) throws Exception {
JSONObject jsonObj = JSON.parseObject(jsonStr);
BroadcastState<String, TableProcess> tableConfigState = context.getBroadcastState(tableConfigDescriptor);
String op = jsonObj.getString("op");
if ("d".equals(op)) {
TableProcess before = jsonObj.getObject("before", TableProcess.class);
String sourceTable = before.getSourceTable();
tableConfigState.remove(sourceTable);
} else {
TableProcess config = jsonObj.getObject("after", TableProcess.class);
String sourceTable = config.getSourceTable();
String sinkTable = config.getSinkTable();
String sinkColumns = config.getSinkColumns();
String sinkPk = config.getSinkPk();
String sinkExtend = config.getSinkExtend();
tableConfigState.put(sourceTable, config);
checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
}
}
（4）自定义函数MyBroadcastFunction-checkTable
在 Phoenix 建表之前要先创建命名空间 GMALL2022_REALTIM。
0: jdbc:phoenix:> create schema GMALL2022_REALTIME;
checkTable() 方法如下
/**
* Phoenix 建表函数
*
* @param sinkTable 目标表名 eg. test
* @param sinkColumns 目标表字段 eg. id,name,sex
* @param sinkPk 目标表主键 eg. id
* @param sinkExtend 目标表建表扩展字段 eg. ""
* eg. create table if not exists mydb.test(id varchar primary key, name varchar, sex varchar)...
*/
private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
// 封装建表 SQL
StringBuilder sql = new StringBuilder();
sql.append("create table if not exists " + GmallConfig.HBASE_SCHEMA
+ "." + sinkTable + "(\n");
String[] columnArr = sinkColumns.split(",");
// 为主键及扩展字段赋默认值
if (sinkPk == null) {
sinkPk = "id";
}
if (sinkExtend == null) {
sinkExtend = "";
}
// 遍历添加字段信息
for (int i = 0; i < columnArr.length; i++) {
sql.append(columnArr[i] + " varchar");
// 判断当前字段是否为主键
if (sinkPk.equals(columnArr[i])) {
sql.append(" primary key");
}
// 如果当前字段不是最后一个字段，则追加","
if (i < columnArr.length - 1) {
sql.append(",\n");
}
}
sql.append(")");
sql.append(sinkExtend);
String createStatement = sql.toString();
// 为数据库操作对象赋默认值，执行建表 SQL
PreparedStatement preparedSt = null;
try {
preparedSt = conn.prepareStatement(createStatement);
preparedSt.execute();
} catch (SQLException sqlException) {
sqlException.printStackTrace();
System.out.println("建表语句\n" + createStatement + "\n执行异常");
} finally {
if (preparedSt != null) {
try {
preparedSt.close();
} catch (SQLException sqlException) {
sqlException.printStackTrace();
throw new RuntimeException("数据库操作对象释放异常");
}
}
}
}
（5）自定义函数MyBroadcastFunction-processElement()
@Override
public void processElement(JSONObject jsonObj, ReadOnlyContext readOnlyContext, Collector<JSONObject> out) throws Exception {
ReadOnlyBroadcastState<String, TableProcess> tableConfigState = readOnlyContext.getBroadcastState(tableConfigDescriptor);
// 获取配置信息
String sourceTable = jsonObj.getString("table");
TableProcess tableConfig = tableConfigState.get(sourceTable);
if (tableConfig != null) {
JSONObject data = jsonObj.getJSONObject("data");
String sinkTable = tableConfig.getSinkTable();
// 根据 sinkColumns 过滤数据
String sinkColumns = tableConfig.getSinkColumns();
filterColumns(data, sinkColumns);
// 将目标表名加入到主流数据中
data.put("sinkTable", sinkTable);
out.collect(data);
}
}
（6）自定义函数MyBroadcastFunction-filterColumns()，校验字段，过滤掉多余的字段
private void filterColumns(JSONObject data, String sinkColumns) {
Set<Map.Entry<String, Object>> dataEntries = data.entrySet();
dataEntries.removeIf(r -> !sinkColumns.contains(r.getKey()));
}
（7）主程序DimSinkApp中调用MyBroadcastFunction提取维度数据
// TODO 8. 处理维度表数据
SingleOutputStreamOperator<JSONObject> dimDS = connectedStream.process(
new MyBroadcastFunction(tableConfigDescriptor)
);
5.3 保存维度到HBase(Phoenix)
1****）程序流程分析
DimSink 继承了RickSinkFunction，这个function得分两条时间线：
一条是任务启动时执行open操作（图中紫线），我们可以把连接的初始化工作放在此处一次性执行；
另一条是随着每条数据的到达反复执行invoke()（图中黑线）,在这里面我们要实现数据的保存，主要策略就是根据数据组合成sql提交给hbase。
*2）创建 Phoeni*****xUtil 工具类，在其中创建insertValues()****方法
package com.atguigu.gmall.realtime.util;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import java.sql.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
public class PhoenixUtil {
/**
* Phoenix 表数据导入方法
*
* @param conn 连接对象
* @param sinkTable 写入数据的 Phoenix 目标表名
* @param data 待写入的数据
*/
public static void insertValues(Connection conn, String sinkTable, JSONObject data) {
// 获取字段名
Set<String> columns = data.keySet();
// 获取字段对应的值
Collection<Object> values = data.values();
// 拼接字段名
String columnStr = StringUtils.join(columns, ",");
// 拼接字段值
String valueStr = StringUtils.join(values, "','");
// 拼接插入语句
String sql = "upsert into " + GmallConfig.HBASE_SCHEMA
+ "." + sinkTable + "(" +
columnStr + ") values ('" + valueStr + "')";
// 为数据库操作对象赋默认值
PreparedStatement preparedSt = null;
// 执行 SQL
try {
preparedSt = conn.prepareStatement(sql);
preparedSt.execute();
// 提交事务
conn.commit();
} catch (SQLException sqlException) {
sqlException.printStackTrace();
throw new RuntimeException("数据库操作对象获取或执行异常");
} finally {
if (preparedSt != null) {
try {
preparedSt.close();
} catch (SQLException sqlException) {
sqlException.printStackTrace();
throw new RuntimeException("数据库操作对象释放异常");
}
}
}
}
}
3）****MyPhoenixSink
自定义 SinkFunction 子类 MyPhoenixSink，在其中调用 Phoenix 工具类的 insertValues(String sinkTable, JSONObject data) 方法，将维度数据写出到 Phoenix 的维度表中。为了提升效率，减少频繁创建销毁连接带来的性能损耗，创建连接池。
（1）添加德鲁伊连接池依赖
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.1.16</version>
</dependency>
（2）连接池创建工具类
package com.atguigu.gmall.realtime.util;
import com.alibaba.druid.pool.DruidDataSource;
public class DruidDSUtil {
private static DruidDataSource druidDataSource;
public static DruidDataSource createDataSource() {
// 创建连接池
druidDataSource = new DruidDataSource();
// 设置驱动全类名
druidDataSource.setDriverClassName(GmallConfig.PHOENIX_DRIVER);
// 设置连接 url
druidDataSource.setUrl(GmallConfig.PHOENIX_SERVER);
// 设置初始化连接池时池中连接的数量
druidDataSource.setInitialSize(5);
// 设置同时活跃的最大连接数
druidDataSource.setMaxActive(20);
// 设置空闲时的最小连接数，必须介于 0 和最大连接数之间，默认为 0
druidDataSource.setMinIdle(1);
// 设置没有空余连接时的等待时间，超时抛出异常，-1 表示一直等待
druidDataSource.setMaxWait(-1);
// 验证连接是否可用使用的 SQL 语句
druidDataSource.setValidationQuery("select 1");
// 指明连接是否被空闲连接回收器（如果有）进行检验，如果检测失败，则连接将被从池中去除
// 注意，默认值为 true，如果没有设置 validationQuery，则报错
// testWhileIdle is true, validationQuery not set
druidDataSource.setTestWhileIdle(true);
// 借出连接时，是否测试，设置为 false，不测试，否则很影响性能
druidDataSource.setTestOnBorrow(false);
// 归还连接时，是否测试
druidDataSource.setTestOnReturn(false);
// 设置空闲连接回收器每隔 30s 运行一次
druidDataSource.setTimeBetweenEvictionRunsMillis(30 * 1000L);
// 设置池中连接空闲 30min 被回收，默认值即为 30 min
druidDataSource.setMinEvictableIdleTimeMillis(30 * 60 * 1000L);
return druidDataSource;
}
}
（3）MyPhoenixSink 函数
package com.atguigu.gmall.realtime.app.func;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.SQLException;
public class MyPhoenixSink extends RichSinkFunction<JSONObject> {
private DruidDataSource druidDataSource;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// 创建连接池
druidDataSource = DruidDSUtil.createDataSource();
}
@Override
public void invoke(JSONObject jsonObj, Context context) throws Exception {
// 获取目标表表名
String sinkTable = jsonObj.getString("sinkTable");
// 获取 id 字段的值
String id = jsonObj.getString("id");
// 清除 JSON 对象中的 sinkTable 字段
// 以便可将该对象直接用于 HBase 表的数据写入
jsonObj.remove("sinkTable");
// 获取连接对象
DruidPooledConnection conn = druidDataSource.getConnection();
try {
PhoenixUtil.insertValues(conn, sinkTable, jsonObj);
} catch (Exception e) {
System.out.println("维度数据写入异常");
e.printStackTrace();
} finally {
try {
// 归还数据库连接对象
conn.close();
} catch (SQLException sqlException) {
System.out.println("数据库连接对象归还异常");
sqlException.printStackTrace();
}
}
}
}
4）主程序 DimSinkApp** ******中调用 ****MyPhoenixSink
// TODO 9. 将数据写入 Phoenix 表
dimDS.addSink(new MyPhoenixSink());
6）测试
（1）启动HDFS、ZK、Kafka、Maxwell、HBase
（2）运行 IDEA 中的 DimSinkApp
（3）执行 mysql_to_kafka_init.sh 脚本
mysql_to_kafka_init.sh all
（4）通过phoenix查看hbase的schema以及表情况
附：整个流程的步骤以及所需要的进程
数据流：web/app -> nginx -> 业务服务器 -> Mysql(binlog) -> Maxwell -> Kafka(ODS) -> FlinkApp -> Phoenix
程序：Mock -> Mysql(binlog) -> Maxwell -> Kafka(Zk) -> DimApp(FlinkCDC/Mysql) -> Phoenix(HBase/ZK/HDFS)
/**
* 需要启动的进程：
* dfs -> zookeeper -> kafka -> maxwell -> hbase -> phoenix(客户端)：bin/sqlline.py
*/
本文转载自: https://blog.csdn.net/JiaXingNashishua/article/details/127771558
版权归原作者 大数据阿嘉 所有， 如有侵权，请联系我们删除。
版权归原作者 大数据阿嘉 所有， 如有侵权，请联系我们删除。