问题导读： 业务背景 常见的实时数仓方案 为什么选择Flink+Clickhouse 实时数仓1.0 踩过的“坑” 实践展望-实时数仓2.0 一、业务背景
二、常见的实时数仓方案
三、为什么选择Flink+Clickhouse batch (B)：离线计算 Analytical（A）：交互式分析 Servering (S)：高并发的在线服务 Transaction (T)：事务隔离机制
1.存储架构 2.查询架构
CREATE TABLE jc_bi.ads_journal_recent_1year
(
`pk_detail` String,
...
`datasource` String,
`synctime` String
)
ENGINE = MySQL('10.100.x.xx:3306',
'jc_bi',
'ads_journal',
'xxx',
'xxx');
--副本表
CREATE TABLE jc_bi.ads_journal_replica
(
`pk_detail` String,
...
`datasource` String,
`synctime` String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/ads_journal_replica',
'{replica}')
PARTITION BY datasource
ORDER BY (unitcode,...)
SETTINGS index_granularity = 8192;
--Clickhouse分布式表，与副本表ads_journal_replica对应
CREATE TABLE jc_bi.ads_journal_dist
(
`pk_detail` String,
...
`datasource` String,
`synctime` String
)
ENGINE = Distributed('cluster_3shards_1replicas',
'jc_bi',
'ads_journal_replica',
rand() % 3);
--以视图合并Mysql引擎和distributed引擎的两张表
CREATE VIEW jc_bi.v_ads_journal
(
`pk_detail` String,
...
`datasource` String,
`synctime` String
) AS
SELECT *
FROM jc_bi.jc_bi.ads_journal_recent_1year
UNION ALL
SELECT *
FROM jc_bi.ads_journal_dist;
复制代码 四、实时数仓1.0
create external table cdmd.dim_bd_accsubj_mapping_hbase(
rowkey String,
pk_accsubj String,
...
mod 复制代码 五、踩过的“坑” 1.cdc插件选型 2.Idea调试Flink代码在开启checkpoint的情况下，触发报错时不输出异常信息且不断重启 3. hive映射Hbase后数据类型不对应问题 4.hdfs配置高可用后namenode节点切换导致hdfs地址失效
5. fastjson的坑 6.实时数据乱序导致计算结果错误 7.关联维度等待数据导致数据阻塞问题
public static byte[] getCellBytes(Connection connection, String tableName, String rowkey, String columnFamily, String column) {
//获取表的连接
Table table = null;
try {
table = connection.getTable(TableName.valueOf("dim", tableName));
} catch (IOException e) {
e.printStackTrace();
log.error("查询hbase中维表过程中创建table出错！！！");
}
//创建get对象
Get get = new Get(Bytes.toBytes(rowkey));
Result result = null;
if ("T_GL_VOUCHERENTRY_Data".equals(tableName)) {
long time = System.currentTimeMillis();
byte[] value;
do {
try {
result = table.get(get);
} catch (IOException e) {
e.printStackTrace();
log.error("查询hbase中维表过程中get result出错！！！");
}
value = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
} while ( value == null && System.currentTimeMillis() <= time + 500); //未拿到维度数据则等待500ms
return value;
} else {
try {
result = table.get(get);
} catch (IOException e) {
e.printStackTrace();
log.error("查询hbase中维表过程中get result出错！！！");
}
byte[] value = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
return value;
}
} 复制代码
DataStream<DWD_GL_DETAIL> joinDs1 = LangChaoFctDs.coGroup(LangChaoVoucherDim)
.where(new KeySelector<FCT_GL_DETAIL, String>() {
@Override
public String getKey(FCT_GL_DETAIL value) throws Exception {
String pk_voucher = value.getPk_voucher();
return pk_voucher;
}
})
.equalTo(new KeySelector<DIM_BD_VOUCHER, String>() {
@Override
public String getKey(DIM_BD_VOUCHER value) throws Exception {
String pk_voucher = value.getPk_voucher();
return pk_voucher;
}
})
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.apply(new RichCoGroupFunction<FCT_GL_DETAIL, DIM_BD_VOUCHER, DWD_GL_DETAIL>() {
Connection connection;
Gson gson;
@Override
public void open(Configuration parameters) throws Exception {
connection = HbaseConnection.createConnection();
gson = new Gson();
}
@Override
public void coGroup(Iterable<FCT_GL_DETAIL> first, Iterable<DIM_BD_VOUCHER> second, Collector<DWD_GL_DETAIL> out) throws Exception {
for (FCT_GL_DETAIL fct_gl_detail : first) {
String pk_voucher = fct_gl_detail.getPk_voucher();
String datasource = fct_gl_detail.getDatasource();
DWD_GL_DETAIL dwdpojo = new DWD_GL_DETAIL();
DWD_GL_DETAIL dwd_gl_detail = handleDwdLc(connection, gson, fct_gl_detail, dwdpojo);
Boolean ishave = false;
for (DIM_BD_VOUCHER dim_bd_voucher : second) {
//省略中间处理过程
ishave = true;
}
//如果维度流中拿不到维度数据则从hbase中查询
if (!ishave && pk_voucher != null) {
String rowkey1 = Md5Util.getMD5Str(pk_voucher).substring(0, 8) + datasource;
//省略往hbase查询凭证维度代码
}
out.collect(dwd_gl_detail);
}
}
@Override
public void close() throws Exception {
if (connection != null) {
connection.close();
}
}
}); 复制代码 六、实践展望-实时数仓2.0 1.统一存储 2.统一计算 3.统一查询 CREATE TABLE gl_detail_ck
( `PK_DETAIL` STRING,
...
`TABLENAME` STRING,
`OPERATION` STRING,
`DATASOURCE` STRING,
`SCN` BIGINT )
WITH ('connector' = 'jdbc',
'url' = 'jdbc:clickhouse://:/',
'table-name' = 'gl_detail',
'username' = '', 'password' = '' ); 复制代码
CREATE TABLE jc_bi.gl_detail (
`pk_detail` String,
...
`sign` Int8,
`version` UInt32 )
ENGINE = VersionedCollapsingMergeTree(sign,version)
PARTITION BY datasource
ORDER BY pk_detail
SETTINGS index_granularity = 8192; 复制代码
{
"sql":"update 'GL_DETAIL' set 'ERRMESSAGE' = NULL where PK_DETAIL=********,、、、",
"TABLENAME":"GL_DETAIL",
"OPERATION":"UPDATE",
"DATASOURCE":"NC",
"SCN":"91685778"
}
复制代码 public void parse(String sqlRedo) throws JSQLParserException {
//通过jsqlparser开源sql解析框架对Sql进行解析获取Satement
Statement stmt = CCJSqlParserUtil.parse(sqlRedo);
LinkedHashMap<String,String> afterDataMap = new LinkedHashMap<>();
LinkedHashMap<String,String> beforeDataMap = new LinkedHashMap<>();
parseUpdateStmt((Update) stmt, beforeDataMap, afterDataMap, sqlRedo);
}
private static void parseUpdateStmt(Update update, LinkedHashMap<String,String> beforeDataMap, LinkedHashMap<String,String> afterDataMap, String sqlRedo){
Iterator<Expression> iterator = update.getExpressions().iterator();
//通过获取更新字段的迭代器填充afterDataMap
for (Column c : update.getColumns()){
afterDataMap.put(cleanString(c.getColumnName()), cleanString(iterator.next().toString()));
}
//通过where语句来获取未修改的字段的值
if(update.getWhere() != null){
update.getWhere().accept(new ExpressionVisitorAdapter() {
@Override
public void visit(final EqualsTo expr){
String col = cleanString(expr.getLeftExpression().toString());
if(afterDataMap.containsKey(col)){
String value = cleanString(expr.getRightExpression().toString());
beforeDataMap.put(col, value);
} else {
String value = cleanString(expr.getRightExpression().toString());
beforeDataMap.put(col, value);
afterDataMap.put(col, value);
}
}
});
}else{
LOG.error("where is null when LogParser parse sqlRedo, sqlRedo = {}, update = {}", sqlRedo, update.toString());
}
} 复制代码 {
"scn":91685778,
"type":"UPDATE",
"schema":"nc",
"table":"GL_DETAIL",
"ts":6797472127529390080,
"opTime":91945745,
"after_TS":"2021-05-10 18:50:53"、、、、,
"before_TS":"2021-05-10 18:50:23"、、、、
}
复制代码 Table sqlParsertest =
tEnv.sqlQuery(
"select "+ "*"+"from (\n" +
"select "+ before_column + " , -1 AS sign ,abs(hashcode("+before_column_string+")) AS version from sqlParsertest where type = 'UPDATE'\n" +
"union all\n" +
"select "+ after_column + ", 1 AS sign , abs(hashcode("+after_column_string+")) AS version from sqlParsertest where type = 'UPDATE' " +
"union all\n" +
"select " +after_column +" ,1 AS sign, abs(hashcode(" +after_column_string +")) AS version from sqlParsertest where type = 'INSERT'\n" +
"union all\n" +
"select "+ before_column+",-1 AS sign, abs(hashcode( "+before_column_string +")) AS version from sqlParsertest where type = 'DELETE' "+
") \n"
);
tEnv.executeSql( " insert into gl_detail_ck select * from " + sqlParsertest); 复制代码
4.实时数据仓库2.0架构
https://mp.weixin.qq.com/s/1pI9OOxH_gkF1wj2nGTTmw 最新经典文章，欢迎关注公众号
/2 
京ICP备2020039040号
简书网举报电话：021-34700000
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡
