Qlik Sense - 根据配置的记录,实现自动抽取源DB表/视图
目录
场景描述
程序开发
〇、主体代码结构
一、常量声明Const
二、子程序声明
三、抽取数据主代码(关键)
四、数据合并(个性化处理、可以忽略)
场景描述
形成Qlik自动化抽数,未来需要增加、减少同构表,只需要在配置表中添加删除记录、并标记抽取方式(增量、全量等)即可,大大减少了工作量
- 根据一份配置表(excel表),自动去源数据库获取数据,并存储成qvd文件
- 支持多个数据源
- 支持全量和多种增量抽取方式(向源库获取同构表)
- 抽取动作均有日志,精确到每次抽取的每一张表
程序开发
〇、主题代码结构
- [Main]:系统常量设定,application自动生成,不做更改
- [Const]:程序常量声明,如DB连接名,文件根目录路径名等
- [SubProgs]:子程序声明,定义一些可复用的程序块
- [LoadData]:抽数主程序,获取源数据库数据,并存储到qvd中
- [CombineData]:其他个性化的数据处理,这里是对两个DB源,同名表进行union(concatenate)
一、常量声明Const
// 全局常量
Set DefaultDBConn = 'LsNodeA'; //默认源DB库连接名
Set SenseRootPath = 'lib://QlikSense'; //存储抽取日志、配置文件等根目录的连接名// 配置文件
Set ConfigPath = '$(SenseRootPath)/Config/';
Set LogFileName = 'ABCLoadLog.qvd'; //日志文件的qvd名称
Set DataPath = '$(SenseRootPath)/Data/'; // 每个view&table的qvd文件路径Set ViewType = 'Interface'; // 'logs' 数据表类型:日志表(logs)/接口表(Interface)// 前端配置文件,用于处理集合表达式
Set DBEffDays = 10;
二、子程序声明
定义可复用的子程序块,这块代码可以考虑以文件形式(txt)放在服务器,application去调用字块程序。特别是log 就能开发给其他application使用
// 创建日志文件(日志不存在时)
Sub CreateLogFile(p_ErpLogPath,p_LogFileName)LogTable:Load * Inline [ViewType,SourceViewName,StatusFlag,ExeTime,Comments,CreationDate];Store LogTable Into [$(p_ErpLogPath)$(p_LogFileName)](qvd);
End Sub;// 存储 table load 记录,增量数据需要根据该日志时间,来确定增量时间点
Sub StoreLogs(p_ViewType, p_DwViewName, p_ErpLogPath, p_LogFileName, p_StatusFlag, p_ThisExeTime)Let LogComments = Replace(ScriptErrorList,chr(10),';');ScriptErrorList = ''; // 清空错误列表Let LogCreationTime = Text(Now());LogTable:Load * From [$(p_ErpLogPath)$(p_LogFileName)](qvd); Concatenate (LogTable)Load * Inline [ViewType ,SourceViewName ,StatusFlag ,ExeTime ,Comments ,CreationDate'$(p_ViewType)','$(p_DwViewName)',$(p_StatusFlag),'$(p_ThisExeTime)','$(LogComments)','$(LogCreationTime)'];Store LogTable Into [$(p_ErpLogPath)$(p_LogFileName)](qvd);Drop Table LogTable;
End Sub;// 必需变量: ViewType(模块名), DwViewName(待存储的表名), ErpQvdPath(qvd保存路径), QvdName(保存名称)
// ErpLogPath(日志路径), LogFileName(日志名称), ThisExeTime(执行时间)
// 可选变量: PreviousErrorCount 如果遍历多个时上一次的错误计数
Sub StoreQvd(p_ViewType, p_DwViewName, p_ErpQvdPath, p_QvdName, p_PreviousErrorCount, p_ErpLogPath, p_LogFileName, p_ThisExeTime)// 保存Qvd文件 & 写入日志文件Let LogNotExists = IsNull(FileTime('$(p_ErpLogPath)$(p_LogFileName)')); If $(LogNotExists) = -1 ThenLogTable:Load * Inline [ViewType,SourceViewName,StatusFlag,ExeTime,Comments,CreationDate];Store LogTable Into [$(p_ErpLogPath)$(p_LogFileName)](qvd);Drop Table LogTable;End If;trace 'ScriptErrorCount=$(ScriptErrorCount)';trace 'p_PreviousErrorCount=$(p_PreviousErrorCount)';If (If(IsNull($(p_PreviousErrorCount)), $(ScriptErrorCount), $(p_PreviousErrorCount)-$(ScriptErrorCount))) = 0 ThenStore $(p_DwViewName) into [$(p_ErpQvdPath)$(p_QvdName).qvd](qvd);Call StoreLogs('$(p_ViewType)', '$(p_DwViewName)', '$(p_ErpLogPath)', '$(p_LogFileName)', 'S', '$(p_ThisExeTime)');ElseCall StoreLogs('$(p_ViewType)', '$(p_DwViewName)', '$(p_ErpLogPath)', '$(p_LogFileName)', 'E', '$(p_ThisExeTime)');End If; End Sub;
三、抽取数据主代码(关键)
抽取数据时,读取excel表,里面记录了要抽取的表/视图、以及相应的抽取方式,内容如下
- source_view_name:要抽取的表名/视图名(按该名字去源库执行select语句,所以名称一定不能错)
- qvd_name:存储成qvd文件时的文件名
- primary_key:表主键,当增量查询时必填
- etl_type:抽取方式
- IUD:增量 - 原表记录存在新增(Insert)、更新(Update)、删除(Delete)操作时,用【IUD】方式抽取
- IU :增量 - 原表记录只有新增(Insert)、更新(Update)操作时,用【IU】方式抽取(表记录不会被删除)
- I :增量 - 原表记录只有新增(Insert)操作时,用 【I】方式抽取
- ALL:全量抽取,完全覆盖上一次抽取的记录
- time_stamp_col:时间戳字段,增量更新时必填,通过时间戳确定数据变更时间,以确定增量记录
- owner:表/视图所属的db用户
- db_conn:在sense中定义的数据库连接名称
在sense editor中配置好连接名,即可写入以下代码,来对数据进行ETL
// 读取excel表配置
SourceViews:
LOADSOURCE_VIEW_NAME,QVD_NAME,PRIMARY_KEY,ETL_TYPE,TIME_STAMP_COL,OWNER,DB_CONN
FROM [$(ConfigPath)ABC_Tables.xlsx]
(ooxml, embedded labels, table is Intf);Let SourceViewsCnt = NoOfRows('SourceViews');Trace '==> DefaultDBConn = $(DefaultDBConn)';
// Load DB data
Let DBConn = '$(DefaultDBConn)'; // 配置默认DB连接
LIB CONNECT TO [$(DBConn)];For i = 0 to $(SourceViewsCnt)-1Let RecordErrorCount = ScriptErrorCount;// Let LogComments = 'Success';//Begin: 获取数据基础信息Let Owner = peek('OWNER',i,'SourceViews')&'.';Let DwViewName = peek('SOURCE_VIEW_NAME',i,'SourceViews');Let PrimaryKey = peek('PRIMARY_KEY',i,'SourceViews');Let QvdName = peek('QVD_NAME',i,'SourceViews');Let EtlType = peek('ETL_TYPE',i,'SourceViews');Let TimeStampCol = peek('TIME_STAMP_COL',i,'SourceViews');Let ViewDbConn = peek('DB_CONN',i,'SourceViews'); If '$(DBConn)' = '$(ViewDbConn)' ThenTrace '==> DB Connection is $(DBConn)';ElseDisConnect;Let DBConn = '$(ViewDbConn)';LIB CONNECT TO [$(DBConn)];End If;Trace ' => DwViewName = $(DwViewName); EtlType = $(EtlType)';Let QvdNotExist = IsNull(QvdCreateTime('$(DataPath)'&'$(QvdName)'&'.qvd'));Let LogNotExist = IsNull(QvdCreateTime('$(DataPath)'&'$(LogFileName)'));If $(LogNotExist) = -1 ThenCall CreateLogFile('$(DataPath)','$(LogFileName)');End If;//End: 获取数据基础信息Trace ' => QvdNotExist = $(QvdNotExist)';//Begin: 确定数据起止时间LastExeTime:Load Text(Date(Max(ExeTime),'YYYY/MM/DD hh:mm:ss')) AS ExeTime From [$(DataPath)$(LogFileName)](qvd)Where SourceViewName = '$(DwViewName)'And StatusFlag = 'S';Let LastExeTime = If(IsNull(FieldValue('ExeTime',1)),'1990/01/01 00:00:00',Text(FieldValue('ExeTime',1)));Let ThisExeTime = Text(Now()); //这里需要修改为当前数据库的实时时间Drop Table LastExeTime;//Begin: 确定数据起止时间Let NowTime = Text(Now());Trace '==> 处理表 $(DwViewName) 开始时间 $(NowTime) '; //Begin 根据配置,做数据增量全量判断If $(QvdNotExist) = 0 ThenIf (EtlType = 'IUD') Then$(DwViewName):Select p.*From $(Owner)$(DwViewName) p,(Select $(PrimaryKey) From $(Owner)$(DwViewName)Where $(TimeStampCol) > TO_DATE('$(LastExeTime)','YYYY/MM/DD HH24:MI:SS')And $(TimeStampCol) <= TO_DATE('$(ThisExeTime)','YYYY/MM/DD HH24:MI:SS')) tempWhere p.$(PrimaryKey) = temp.$(PrimaryKey);Concatenate ($(DwViewName))Load *From [$(DataPath)$(QvdName).qvd](qvd)Where Not Exists ($(PrimaryKey));Inner Join ($(DwViewName))Select $(PrimaryKey) From $(Owner)$(DwViewName);ElseIf (EtlType = 'IU') Then$(DwViewName):Select p.*From $(Owner)$(DwViewName) p,(Select $(PrimaryKey) From $(Owner)$(DwViewName)Where $(TimeStampCol) > TO_DATE('$(LastExeTime)','YYYY/MM/DD HH24:MI:SS')And $(TimeStampCol) <= TO_DATE('$(ThisExeTime)','YYYY/MM/DD HH24:MI:SS')) tempWhere p.$(PrimaryKey) = temp.$(PrimaryKey);Concatenate ($(DwViewName))Load *From [$(DataPath)$(QvdName).qvd](qvd)Where Not Exists ($(PrimaryKey));ElseIf (EtlType = 'I') Then$(DwViewName):Select p.*From $(Owner)$(DwViewName) p,(Select $(PrimaryKey) From $(Owner)$(DwViewName)Where $(TimeStampCol) > TO_DATE('$(LastExeTime)','YYYY/MM/DD HH24:MI:SS')And $(TimeStampCol) <= TO_DATE('$(ThisExeTime)','YYYY/MM/DD HH24:MI:SS')) tempWhere p.$(PrimaryKey) = temp.$(PrimaryKey);Concatenate ($(DwViewName))Load *From [$(DataPath)$(QvdName).qvd](qvd);ElseIf (EtlType = 'ALL') Then$(DwViewName):Select *From $(Owner)$(DwViewName);End If;Else$(DwViewName):Select *From $(Owner)$(DwViewName);End If;//End 根据配置,做数据增量全量判断Let NowTime = Text(Now());Trace '==> 处理表 $(DwViewName) 结束时间 $(NowTime) '; // 保存数据至Qvd & 写日志Call StoreQvd('$(ViewType)', '$(DwViewName)', '$(DataPath)', '$(QvdName)', '$(RecordErrorCount)', '$(DataPath)', '$(LogFileName)', '$(ThisExeTime)');Drop Table $(DwViewName);Next i;Set ErrorMode = 1;Drop Table SourceViews;//End: Load Data
DisConnect;
这段代码存在一点问题是:
ThisExeTime的取值现在是获取Sense服务器上时间,最好要改成从各数据源取得。
如:Oracle DB - [sysdate],MySql/PostgreSQL - [now()]
四、数据合并(个性化处理、可以忽略)
这里代码对应场景是 有N个数据源(DB),有一部分表的表结构在各个源库中是一模一样的,只是表名差别(业务上是同一类业务下的不同子分类,后台表分了库)。使用代码进行自动union,形成一张大表。
Set filePrefixSet = 'ABC_INTERFACE_BATCHES,ABC_INTERFACE_DETAILS,ABC_INTERFACE_REFUNDS';
Set ViewType = 'QlikInterface';
Let fileTypeCnt = SubStringCount('$(filePrefixSet)',',') + 1;For fileTypeIdx = 1 to fileTypeCntLet filePrefix = SubField('$(filePrefixSet)',',',fileTypeIdx);Let prefixLen = Len('$(DataPath)')+Len('$(filePrefix)')+2;Set fileCnt = 0;For Each file in FileList('$(DataPath)'&'$(filePrefix)'&'_*.qvd')Let fileCnt = fileCnt + 1;Let srcSysIdf = Replace(Mid('$(file)',prefixLen),'.qvd','');If fileCnt = 1 Then$(filePrefix):Load '$(srcSysIdf)' AS SRC_SYS_IDF,*From [$(file)](qvd);ElseConcatenate ($(filePrefix))Load '$(srcSysIdf)' AS SRC_SYS_IDF,*From [$(file)](qvd);End If;NextStore $(filePrefix) Into [$(DataPath)$(filePrefix).qvd](qvd);Drop Table $(filePrefix);
Next
Qlik Sense - 根据配置的记录,实现自动抽取源DB表/视图相关推荐
- SpringBoot自动装配源码解析
Spring Boot 自动装配原理 使用Spring Boot最方便的一点体验在于我们可以几零配置的搭建一个Spring Web项目,那么他是怎么做到不通过配置来对Bean完成注入的呢.这就要归功于 ...
- 一文解析:安装和运行Qlik Sense的系统要求及浏览器支持
在开始本文之前,请阅读这篇文章以了解什么是Qlik,以及Qlik的功能详细介绍:商业智能BI明星产品:一篇文章带你走进Qlik 前面几篇文章,我们已经详细介绍了Qlik灵活的功能.强大的性能.有趣的扩 ...
- Qlik Sense 创建 QVD 文件详解(Store)
文章目录 1 概述 2 语法 2.1 写入 QVD 2.2 查看 QVD 2.3 读取 QVD 3 扩展 3.1 QvdNoOfRecords:获取 QVD 文件记录数 1 概述 写脚本,将数据存储至 ...
- Qlik Sense Desktop 下载和安装
前言 现在经常要和Qlik Sense打交道,通常企业用的比较多的是Hub版,但是对于个人而言,Desktop是免费且不错的. 什么是Qlik Sense Desktop? Qlik Sense De ...
- 【Oracle RAC+DG实验】Oracle RAC+ASM+DataGuard配置实验记录+常见问题
[Oracle RAC+DG实验]Oracle RAC+ASM+DataGuard配置实验记录+常见问题 1.环境规划: ---RAC环境介绍(primary database) ...
- 配置spring整合jpa自动生成数据表
配置spring整合jpa自动生成数据表 applicationContext.xml <?xml version="1.0" encoding="UTF-8&qu ...
- 【clickhouse】 clickhouse配置查询记录query_log
1.概述 转载:clickhouse配置查询记录query_log 最近在使用clickhouse中,看到官方文档上可以配置query_log,但是文档上写的配置比较模糊,特此记录一下具体配置方法,以 ...
- Centos配置Jenkins实现Android自动打包并上传到蒲公英
本篇文章来自 徐永红 的投稿,给大家讲解Centos配置Jenkins实现Android自动打包并上传到蒲公英,希望对大家有所帮助. 徐永红 的博客地址: https://xuyonghong.cn ...
- manjaro i3安装配置全记录
manjaro i3安装配置全记录 安装 准备 安装踩坑记录 硬盘 安装过程 系统配置 准备 设置中文系统 安装必要配置 美化桌面 常用软件推荐 安装研发环境(java方向手动安装) 安装 准备 ma ...
最新文章
- C++ 字符串字母大小写转换
- 王贻芳院士:为什么中国要探究中微子实验?
- 京东程序员回应“被猝死”:我还活着,还在写代码
- 深度学习的40种应用
- phpmyadmin 无法载入 mysql 扩展_phpMyAdmin“无法载入mysql扩展,请检查PHP配置”解决方法...
- Linux各个文件夹的作用~~~非常实用!!
- router-link
- Manasa and Combinatorics
- ​2020启示:拼多多篇— —退潮后,你才发现人家是游泳健将
- 什么是.NET Framework
- 2022年1月舆情信息事件分析总结报告
- 微橙人人店 php,最新微橙人人店分销3.5.7整站源码,全开源任意二开,新增及修复多项功能,附升级补丁...
- android游戏源码 2048,Android实现2048小游戏
- 计算机专业英语 9次作业合集 从第九次到第一次
- BLE(15)—— Privacy
- git 解决push报错
- 小米已成国产手机标杆,海外出货量碾压华为OV
- 【攻略】如何用云服务器实现云搬砖
- 华为设备配置PIM-SM命令
- 广州3日2夜游行程~
热门文章
- css 高度塌陷_css中父元素高度塌陷是什么意思,如何解决?(附代码)
- FreeEIM 网站地图 A
- 51单片机用c语言在液晶1602上显示汉字,lcd1602与单片机连接图,基于51单片机的lcd1602液晶显示屏连接电路图...
- matlab电位图仿真实验,基于MATLAB的静电场描绘实验仿真
- 一个人越想赚钱,就越要改掉这3个习惯,否则注定穷一辈子
- jmp指令(0903)
- 面试笔记:面经-瓜子
- python requests瓜子二手车城市列表
- Invalid bound statement (not found): cn.jeefast.xiangmu.dao.AchDao.selectByI 解决
- Kafka架构篇 - 多副本机制