2024年7月

前言

最近看了
《中国数据库前世今生》纪录片
，感触颇深，也是一直在思考到底该用何种方式起笔*回顾这段筚路蓝缕却又充满民族自豪感的历程。大概构思了一*左右吧，我想，或许还是应该从那个计算机技术在国内刚刚萌芽的年代开始讲起，那时的一切都显得那么原始而纯粹，一群怀揣梦想的科研人员，在资源匮乏、条件艰苦的环境中，凭借着对知识的渴望和对国家的忠诚，开启了中国数据库从无到有的艰难探索。

资料有限，未尽之处，敬请谅解。

萌芽阶段—信息技术的曙光与基础建设（50~70年代）

或许有朋友会很奇怪，为何我要从50年代开始写起呢？我给出的答案是”
虽然这个时代看似与数据库技术的起步相隔甚远，但是，却早已悄然埋下了中国信息技术革命的种子，奠定了后*数据库发展的基础土壤。
“

根据目前收集到的资料，中国的信息化最早应该起步于
1956
年。

1956年，在***总理的亲自主持下，制定了我国《十二年科学技术发展规划》，选定计算机、电子学、半导体、自动化四项，作为科学规划的四项紧急措施。根据《十二年科学技术发展规划》，开始了我国计算机事业的创建工作。罗沛霖院士说：“为什么会成为‘紧急措施’，原因是在科学规划制定中有所忽略，直到***总理发现，首批*华的18位苏联国家级专家中居然有6位是通信、电子、计算机专家，才指出了《十二年规划》偏重传统产业的问题。”1957年7月，清华大学首批计算机专业有杨天行等几十人毕业。同期，北京大学创办了计算数学专业和软件训练班。哈军工、哈工大等相继培养了计算机专业本科生、研究生。总体概括，我国在上世纪50年代，培养计算机学科人才总数达300余人，后*大多成为我国信息化的栋梁之材。

而那时，国内对于计算机的认知几乎是一片空白，不知道什么是计算技术，也不明白何为数据，大部分学者压根就没有见过计算机。

1958
年8月1日，在前苏联的帮助下，第一台计算机
103机
交付使用，据考证，当时这台计算机运行速度仅有每秒30次，但这仍然
标志着我国第一台电子计算机的诞生
。

103机是中国制造的第一种通用数字电子计算机，为电子管小型计算机。该型机是在苏联提供的M-3小型电子计算机图纸的基础上研制的，研制工作由中国科学院计算技术研究所和北京有线电厂（国营738厂）承担。1958年8月，第一台103机在中科院计算所初步调试成功，这标志了中国第一台计算机的诞生。103机最早使用磁鼓存储器，后又增配了磁芯存储器。使用磁鼓存储器时，103机运算速度约为每秒30次；使用磁芯存储器时，运算速度约为每秒1500至2000次。130机研制项目项目负责人为莫根生、张梓昌，骨干有董占球、王行刚等人，另有苏联专家指导。1959年起103机被用于一些科学计算，如人民大会堂主席台的力学结构计算便是由103机完成的。

image

时隔一年多，
1959
年9月，根据前苏联有关计算机技术资料制成的
104
大型通用电子计算机通过试运算，运算速度提升到每秒1万次，
正式宣告我国第一台大型通用电子计算机试制成功
。

在
1960
年4月，我国第一台
自行设计
的小型通用电子计算机
107机
研制成功；
1964
年，我国第一台自行设计的大型通用数字电子管计算机
119机
研制成功，标志着我国独立自主发展计算机事业进入了一个新阶段。同时，研究人员还自主研制了当时具有国际水平的我国最早的实用高级程序设计语言BCY，并在119机上实现了其编译系统。利用BCY语言及其编译系统，在119机上完成了东方红一号卫星的轨道模拟计算。

在研制第一代电子管计算机的同时，我国也开始着手研制第二代晶体管计算机。

1964
年11月，研制成功我国第一台晶体管通用电子计算机
441-B
机。441-B机是在当时外部条件十分困难的情况下使用国产半导体元器件研制的。
1965
年又研制成功我国第一台大型晶体管通用数字计算机109乙机，广泛应用于国民经济和国防部门。

为服务“两弹一星”工程，在109乙机的基础上，
1967
年又推出在技术上更加先进、更加成熟的109丙机。这台机器服务国防事业长达15年，为我国核武器研制作出了卓越贡献，被誉为“功勋计算机”。

回望从1956年到1966年的社会主义革命和建设时期，在计算机认知几乎为零的背景下，仅仅数年间，我国不仅拥有了自己的计算机，还踏入了自行设计的门槛，不可不谓之“
奇迹
”了，这，也为后续数据库技术的孕育奠定了极为重要的物质与技术基础。

起步阶段—数据库概念的引入及初步探索（70~80年代）

70年代，
国外
数据库的发展可谓是如日中天。

关系型数据库之父埃德加·考特(Edgar Frank Codd)在美国发表《用于大型共享数据库的关系数据模型》这篇划时代的论文，不仅彻底颠覆了数据管理的传统模式，还为全球数据库技术的发展指明了新的方向。

到70年代末期，国外的数据库技术基本完成了从实验室到科技商用产品的市场化历程，数据库从网络型和层次型数据库过渡到
关系型
数据库。

但是，由于冷战的余波，西方国家对包括数据库技术在内的高新技术实施严格出口管制，企图遏制社会主义国家的技术进步。

甲骨文的创始人Ellison在1995年初次访华，据说Ellison到中国后的第一项工作就是拍摄一部有关Oracle一体机的宣传片，于是时任甲骨文中国总裁的冯星君就安排了20名北京的小学生共同参与宣传，可本*约好的早上8点正式开机，但直到9点Ellison他老人家都还没有起床。当时北京的冬天格外的寒冷，最低气温达到零下二十度，这些小学生在没有暖气的大巴上急得哇哇大哭，没办法冯星君只好一再催促Ellison赶快*救场，不过直到快到中午十二点钟，这位数据库帝国的掌门人才姗姗*迟，这种无奈的背后，其实也折射出我国当时在数据库是没有任何底气和信心的，就像现在的ASML光刻机一样，被卡着脖子，就只有任人摆布的份了。

在这样的时代背景下，1977年11月，以人民大学
萨师煊教授
为代表的老一辈科学家、教育家以一种强烈的责任心和敏锐的学术洞察力，意识到新兴数据库技术的潜在价值，他们在安徽黄山组织了一次小范围的数据库技术研讨会，拉开了我国数据库研究的序幕。虽然参会人员只有50余位，但这次会议就像一点星星之火，开始在中国的土地上闪烁着数据库的点点光芒。

我国对数据库技术的研究与应用始于20世纪70年代中后期，其标志是1977年11月在安徽省黄山召开的数据库技术研讨会。会议共录用7篇论文。这次会议后*被确认为
第一届全国数据库学术会议（NDBC）
。萨师煊教授参加了这次会议并发表了论文。萨师煊教授担任数据库学组组长期间，在他的领导下自1982年起，每年都要举办一次全国数据库学术会议，为数据库工作者交流学术成就和开发经验，检阅工作成果提供了讲坛。更重要的是形成了一个“团结、执着、和谐、潇洒”的优良学风，为推动我国数据库技术的持续发展打下了基础。

黄山会议上，中国计算机学会软件专业委员会决定下设成立数据库学组,虽然仅仅是一个三级学科组织，但它却迈出了对于中国数据库而言具有里程碑意义的一步。正是这个数据库学组的成立，
被视为国产数据库研究的起源
。

image

黄山会议召开后的第二年，
1979
年，萨师煊将自己的讲稿汇集成《数据库系统简介》和《数据库方法》并发表。这也是我国
最早的数据库学术论文
，对我国数据库研究和普及起到了启蒙作用，而人民大学也被业界赞誉为“中国数据库的发源地”。

1982
年，萨师煊教授率先在中国人民大学开设“
数据库系统概论
”课程。课程不仅传授了数据库的基本原理、设计方法与管理技术，更重要的是，它激发了一代学子对数据库领域的浓厚兴趣与探索热情，为我国培养了最早一批数据库专业人才。这些学生后*大多成为了中国数据库技术研究与应用领域的中坚力量，对中国数据库产业的发展产生了深远影响。

1983
年，以萨师煊教授、王珊教授等为代表的专家学者出版了我国第一部数据库教材《
数据库系统概论
》。这是国内第一部系统阐明数据库原理、技术和理论的教材，一直被大多数院校计算机专业和信息专业采用。

时至今日，笔者读大学时所用到的数据库课程教材，仍然是萨师煊教授所著。

image

随着数据库基础理论体系的完善，中国的数据库领域也逐渐从理论探索迈向实践应用的新阶段。

然而，由于技术能力有限，自主研发受到了较大阻力，所以最初的方向是偏向于直接
外*引进
的。

据笔者考证，在80年代初期，我国开始尝试引进
DBASE II
，这是有明确记录的较早引入的数据库系统之一。DBASE II在当时因其易用性和适应个人电脑环境的能力而被广泛采纳，对中国的数据库技术应用和发展起到了推动作用，但
这也标志着中国数据库市场即将步入一个以外企产品为主导的时期
。

image

混沌阶段—国外数据库的商战和中国数据库的诞生（80~90年代）

80~90年代，是
混乱但也辉煌的十年
。

混乱是因为在这段时间里，全球数据库市场经历了前所未有的动荡与竞争。随着
1979
年1月1日，中美正式建交，国际上，以Oracle、IBM的DB2等为代表的西方数据库巨头，纷纷涌入中国市场，展开了激烈的商战。这些国际巨头不仅带*了先进的数据库技术，也通过资本运作、市场策略和品牌影响力，对我国市场进行了深度渗透和抢占。本土企业面临巨大的挑战，市场竞争环境变得错综复杂，规则多变，小规模的数据库服务商在技术和资金双重压力下，生存空间被急剧压缩，整个行业处于一种高度不确定和混沌的状态。

而辉煌，则是因为在这样的国际竞争压力之下，国内的科技企业和研究机构并未退缩，反而激发出了前所未有的创新活力和自主发展的决心，在政府项目、金融、电信等关键领域逐步实现了从无到有的突破，为中国信息化建设提供了坚实的基础。

下面，我们按照时间线*依次梳理，
【国外】代表国外数据库，【国内】代表国内数据库
。

【国内】
1988
年，华中科技大学数据库与多媒体技术研究所，成功研发
我国第一个自主版权的国产数据库管理系统原型CRDS
，这可以看作是 DM 的起源，这一里程碑式的成就不仅填补了国内在自主数据库技术研发上的空白，还极大地激励了国内科技界对于信息技术核心领域的自主研发热情。

1984年4月，冯裕才编写的《数据库系统基础》教材出版，这本教材让冯裕才前期积累的理论知识得以沉淀，也为后面设计和研发数据库系统原型奠定了基础。1988年，冯裕才带领团队做出了我国第一个自主版权的国产数据库管理系统原型CRDS，它构建了达梦乃至整个中国数据库的“代码根”。

【国外】
1989
年，*自中国台湾的冯星君将
Oracle
带入到中国，并依靠代理开始售卖Oracle产品，2500元一套，要求客户带软盘现拷，而且没有说明书。

由于技术领先且在国际上有着较好的口碑，Oracle开始疯狂扩大业务规模，可以说几乎所有中国本土的数据库公司背后都曾有过他的身影。

【国内】
1990
年，东软完成OpenBASE 1.0的开发，12月27日通过冶金工业部科学技术司组织的技术鉴定。其后以软件包的形式在日本软件市场上销售，取得了良好的经济效益。
OpenBASE是我国第一个具有自主版权的商品化数据库管理系统
。

OpenBASE是东软集团有限公司软件产品事业部推出的我国第一个自主知识产权的商品化数据库管理系统，该产品由东软集团有限公司软件产品事业部研发并持有版权。10多年*，OpenBASE已逐渐形成了以大型通用关系型数据库管理系统为基础的产品系列，包括： OpenBASE多媒体数据库管理系统，OpenBASE Web应用服务器、OpenBASE Mini嵌入式数据库系统、 OpenBASE Secure安全数据库系统等。

截止目前，OpenBASE系列仍然在不断迭代更新，紧随数据库技术的最新发展趋势，持续融入云计算、大数据、人工智能等前沿技术，以满足日益增长的多元化数据管理和分析需求。

image

【国外】眼看Oracle赚的盆满钵满，其他数据库巨头自然也是眼馋的紧。

1991
年12月，
Sybase
进入中国大陆，随后投资230万美元正式设立赛贝斯软件。

1992
年，
IBM
正式进入中国，并启动了“发展中国”的大战略，协助中国全面开放。带*了
DB2
和
informix
； 同年，
Microsoft微软
在北京设立代办处，开始了在中国长达三十余年的深植。

image

【国内】面对国外数据库产商对本土市场的大肆侵占，我国科研人员也并未选择坐以待毙，而是开始了一系列卓有成效的创新行动，亲手创造了属于我们的
辉煌十年
。

1991
年，华中科技大学数据库与多媒体技术研究所团队先后完成了军用地图数据库 MDB、知识数据库 KDB、图形数据库 GDB、以及语言数据库 ADB。

1992
年，华中理工大学
达梦数据库研究所
成立，这也是
国内第一个数据库研究所
。

1993
年，该研究所研制的多用户数据库管理系统通过了鉴定，标志着
达梦数据库 1.0 版本
的诞生。

同年，中国航天科技集团联合浙江大学正式开展工程数据库管理系统（OSCAR）技术攻关和产品研制工作，
航天国产数据库启程
。

1995
年5月，邮电部电信总局提出开发和建设"市内电话业务计算机综合管理系统"，即”
九七工程
“，并于同年 7 月下发了一系列的技术和业务规范，要求全国县以上的邮电局在 1997 年底前实施
「九七工程」
。

1996
年，DM2（第二代达梦数据库系统）研制成功；同年，东软正式推出产品OpenBASE 3.0，开始标志着
我国具有自主版权的数据库系统软件产品正式走向市场
。

1997
年，中国电力财务公司华中分公司
财务应用系统首次使用国产数据库 DM2
，随后，在全国 76 家分子公司上线使用；东软开发基于Internet/Intranet多媒体综合信息服务体系结构及其支撑平台的OpenBASE，同时入选国家863计划重大目标产品。

【国外】然而，现实却总是会给我们当头一棒。

1997
年，Oracle拿下东三省邮电管理局“九七工程”5期工程的大单，由此在中国数据库领域，尤其电信领域站稳脚步。

1998
年，自九七工程落下帷幕后，中国数据库的行业格局也就在这一时期形成了：
金融行业用 IBM DB2 数据库，Informix 数据库，在电信行业则是 Oracle 的天下。

至此，
中国数据库市场正式进入被外企产品主导的时期
。

image

【国内】即便如此，面对围剿，中国本土的数据库企业并没有选择放弃，反而激发出了更强的斗志和创新精神，他们深刻认识到，要在强手如林的市场中脱颖而出，就必须走出一条差异化竞争的道路，
即自主可控，打响品牌第一枪
！

1999
年6月18日，
中国第一家数据库公司北京人大金仓信息技术股份有限公司
创立，他的诞生，不仅填补了国内数据库品牌的空白，更是向世界宣告了中国数据库行业自主发展的决心。

后*，经过数年的学术和研究成果转化，人大金仓研制开发出了具有自主知识产权的大型通用关系型数据库管理系统
KingbaseES
，标志着
数据库科研成果成功从实验室走向市场
。此后多家国产数据库厂商相继问世，开启了我国数据库“产学研用”的可持续发展之路。

image

1999
年 8月24日，根据数据库领域学术研究和应用发展的需要，在兰州大学召开的第十六届全国数据库学术会议上，
中国计算机学会数据库专业委员会
正式成立，标志着中国数据库领域进入了一个
组织化、专业化
发展的新阶段。

回顾上面这些年，我们不难看出，虽然这个阶段市场是被国外数据库产品所主导，但是我们的先辈们仍然凭借着坚韧不拔的精神和对技术自主的执着追求，依旧坚守着国产数据库的自主创新之路。他们不仅在夹缝中求生存，更是在逆境中寻求创新与突破，从模仿学习到自主研发，每一步都凝聚着汗水与智慧。

正是在这个过程中，涌现出了一批批优秀的国产数据库企业，比如上文说到的人大金仓，以及在下文要提到的达梦数据库、南大通用、神舟通用等，他们不仅仅局限于复制国外成功模式，而是更加注重
结合中国国情和行业需求
，开发出更适合本土应用场景的特色功能和服务，而这，
也吹响了国产数据库企业的第一次反攻号角。

发展阶段—国产数据库的百花齐放与稳健成长（00~10年代）

进入21世纪的头十年，信息技术的全球化浪潮汹涌澎湃，互联网的普及和电子商务的兴起极大地推动了数据量的爆炸式增长，为数据库行业带*了前所未有的机遇与挑战。

面对国际数据库巨头的强势地位，国产数据库企业一方面加快技术引进消化吸收的步伐，缩短与国际先进水平的差距；另一方面，针对国内市场的特定需求，如中文处理、多用户并发、以及特定行业应用等，进行本土化创新，逐步在某些细分领域形成特色优势。

期间，大量优秀的国产数据库企业、数据库产品纷纷涌现。

这个时期的数据库大部分都是依托大学、国有产业之类，研发的数据库也都大部分用在银行，国家政府机关，行政企事业单位等。*满足国家对自研数据库项目使用的一些要求和建议。

2000
年11月13日，武汉华工
达梦数据库
有限公司（武汉达梦数据库股份有限公司）成立，同年推出产品DM3。

2003
年，北京
神舟航天
软件技术有限公司成立，开始承担国家863项目，培育数据库产品，推动了国产数据库在政府、军工等关键领域的应用。

2004
年5月，天津
南大通用
数据技术公司成立，专注于数据库管理系统研发，成为国产数据库领域的又一生力军，主要产品为
GBASE
。

2004
年，人大金仓完成具有自主知识产权的KingbaseES V4版本的发布，进一步提升了国产数据库在企业级应用中的竞争力。

2005
年：“十五”期间，中国航天科技集团的北京神舟航天软件技术有限公司研发成功
神舟数据库管理系统
（OSCAR），神舟OSCAR数据库成功应用神舟六号飞船和长征二号Ｆ型运载火箭的研制、生产和管理。

2006
年3月：南大发布GBase 8g通用数据库。

2006
年下，达梦数据库推出DM5，实现了从DM3到DM5的重大技术跨越，支持了更多企业级特性，如高级安全性、高可用性及大数据处理能力，荣获第十届软博会金奖。

2008
年11月：天津
神舟通用
数据技术有限公司成立，主要产品为
神通大型通用数据库
。

在大数据与互联网等的发展推动下，一批新兴国产数据库厂家开始涌现。一些云计算厂商以及部分数据库厂商，也基于MySQL、PostgreSQL等开源数据库做了一些改造。

2007
年，腾讯内部启动了一个7*24高可用服务项目，基于开源体系
MySQL
研发了一款数据库产品，即
TDSQL
前身。

2009
年：随着淘宝、支付宝的用户数量激增，阿里巴巴依托MySQL，研究出MySQL分支
AliSQL
。也是从那时起，
MYSQL
开始走入中国的互联网，并到如今一发不可收拾。

Oracle数据库的一个致命缺点开始暴露出*，贵。不仅Oracle软件贵，要维持Oracle数据库+IBM小型机+EMC的开支也相当庞大，另外对于管理员的能力要求，也非常的高。不仅如此，“第一是Oracle作为商业产品，本身也有性能的上限，第二是黑盒子。对于没碰到过的场景，无论再怎么努力，也是无法预测可能出现的问题的。”当时的淘宝数据库大神余锋告诉记者。
中国的互联网公司大部分都草莽出身，对于性价比极为看中，而这个时候，美国的雅虎公司开始率先使用MySQL数据库，一度在世界上有数以千计的服务器都是用MySQL数据库。当时的雅虎的光环，远高于今天的谷歌，FB，可以说今天中国所有互联网公司的架构，都可以在雅虎找到源头，在雅虎的示范效应下，很快中国的互联网公司就开始自己的MySQL之路。

随着去IOE化的提出，国内企业开始积极探索自主可控的IT基础设施建设路径。这一趋势极大地推动了开源技术，尤其是MySQL的广泛应用。众多厂商选择MySQL作为替代Oracle的解决方案，不仅出于成本考量，更重要的是看到了开源软件在灵活性、可定制性以及社区支持方面的优势。在这一浪潮中，也诞生了许多优秀的DBA，为中国数据库技术生态的建设贡献了宝贵的知识与经验，也
为2010年之后国产数据库的蓬勃发展奠定了人才基础
，自那时以后，国产数据库
开始弯道超车
，国产数据库领域才
真正进入到了茁壮成长、蓬勃发展的时代
。

image

崛起阶段—国产数据库的技术突破与市场拓展（10年代~至今）

对比上一个十年，从10年代开始，最大的特点是在技术上实现了较大突破。国产数据库厂商不再局限于对现有技术的模仿和优化，而是开始在核心算法、存储引擎、分布式架构等方面进行原创性研发。

2011
年，阿里
Oceanbase
诞生，这是由蚂蚁集团完全自主研发的国产原生分布式数据库，是一个在TPC-C和TPC-H测试上都刷新了世界纪录的国产原生分布式数据库。

2011
年，巨杉数据库成立，专注分布式数据库。总体架构是基于shard-nothing 的方式。

2012
年，腾讯以 “
开源定制化+自研
” 为策略进行定制化，打磨出更加通用的数据库产品，正式命名
TDSQL
。

TDSQL是腾讯云自研企业级分布式数据库，旗下涵盖金融级分布式、云原生、分析型等多引擎融合的完整数据库产品体系，提供业界领先的金融级高可用、计算存储分离、数据仓库、企业级安全等能力，同时具备智能运维平台、Serverless版本等完善的产品服务体系。
截至2021年，腾讯云的数据库已经拥有超过50万客户 ，服务1000多家政府客户和2000多家金融客户 ，每天支撑数十亿笔的交易量 。同时广泛覆盖游戏、电商、移动互联网、云开发等泛互联网业务场景，助力新零售、教育、SaaS、广告等超过4000家行业客户进行数字化升级 。

image

【转折点】
2013
年，棱镜门事件曝光，将数据安全与隐私保护的问题正式推到了国际社会的聚光灯下，也对国产数据库的发展起到了重要的转折作用。在此之后，国产化数据库如达梦、金仓、神通、南大等得到了广泛关注，这些数据库也开始多应用于央企、国家财政、军事等专用领域。

棱镜计划（PRISM）：是一项由美国国家安全局自2007年起开始实施的绝密电子监听计划。该计划的正式名号为“US-984XN”。
根据报道，泄露的文件中描述PRISM计划能够对即时通信和既存资料进行深度的监听。许可的监听对象包括任何在美国以外地区使用参与计划公司服务的客户，或是任何与国外人士通信的美国公民。国家安全局在PRISM计划中可以获得的数据电子邮件、视频和语音交谈、影片、照片、VoIP交谈内容、档案传输、登入通知，以及社交网络细节。综合情报文件“总统每日简报”中在2012年内在1，477个计划使用了*自PRISM计划的资料。

2014
年，国内首家互联网银行-微众银行核心交易数据库采用
腾讯云TDSQL
。

2014
年，腾讯发布TBase数据库，开始在腾讯大数据平台内部使用。

2014
年末，除MySQL外，PostgreSQL、Redis、MongoDB和Hbase等
开源数据库
也活跃起*，在各大数据库大会和社群中一起寻找着中国数据库的新方向。

2015
年，
阿里巴巴的OceanBase
经过内部多年打磨最终对外推出使用。

2015
年，
腾讯‌TBase数据库
在微信支付商户集群上线，‌支持每天超过6亿笔的交易。

2016
年，支付宝总账全面用OceanBase替换Oracle。

2017
年，阿里云（阿里巴巴）公布国内首个
自研企业级关系型云数据库
PolarDB技术框架。

2017
年， Gartner 发布数据库系列报告第一次看到国产数据库的身影，
国产数据库-阿里AsparaDB、南大通用GBase、和SequoiaDB
首次入选。

2018
年，腾讯云宣布新一代自研
云原生数据库
CynosDB正式发布。

2018
年11月：Gartner发布数据库报告，
华为云
、
腾讯云
紧接进榜，这一系列的突破标志着国产数据库进入了一个全新的发展阶段。

2019
年，新一代达梦数据库管理系
DM8
发布。

2019
年5月：华为公司发布了
全球首款AI原生(AI-Native)数据库
——
GaussDB
。

2019
年9月：腾讯云TDSQL在张家港农商银行新一代核心业务系统上线。

2019
年9月19日：华为宣布将开源其数据库产品，开源后命名
openGauss
。

2019
年10月2日：国际事务处理性能委员会公布数据库最新性能测试结果，阿里巴巴集团蚂蚁金服的分布式关系数据库
OceanBase
打破了Oracle保持9年的TPC-C基准性能测试世界纪录。

image

2020
年5月，中信银行与中兴通讯联合研发的
GoldenDB
上线，成功取代了在中信银行核心系统服役了几十年的IBM AS400数据库。

2020
年5月，
OceanBase
再次将自己创造的数据库TPC-C基准性能测试世界记录提升11倍，将甲骨文、IBM 等一众老牌数据库巨头甩在身后。

2020
年11月：达梦发布数据共享集群（DMDSC）、启云数据库（DMCDB）、图数据库（GDM）、新一代分布式数据库四款产品。

2020
年11月：以“自研·智能·新基建——云和数据促创新 生态融合新十年”为主题的第十届数据技术嘉年华大会在北京成功举办，三大云数据库掌门人，三大独立数据库首席，一位云和*墨创始人，共同讲述了国内数据库发展的光辉十年。

image

2020
年11月16日，腾讯云宣布，旗下国产金融级分布式数据库
TDSQL
在印尼Bank Neo Commerce银行新核心系统正式投入使用。意味着腾讯云自研数据库TDSQL不仅在国内金融级市场应用走在前列，在
国际数据库领域
也具备强有力的竞争优势。

2020
年11月：Gartner公布2020年度全球数据库魔力象限评估结果，阿里云首次挺进全球数据库第一阵营-领导者（LEADERS）象限，这也是中国数据库40年*首次进入全球顶级数据库行列。

image

2020
年12月，腾讯云宣布TDSQL、CynosDB、TBase 统一至全新TDSQL品牌。

image

2020
年底，VLDB 刊登63篇论文，*自中国学者和研究人员的文章23篇，在所有国家中排行第一，占比36.5%；
阿里巴巴和腾讯的成果显著
。

毫无疑问，2020年，是
国产数据库的丰收年
，这一系列标志性事件和成就不仅展示了国产数据库技术的蓬勃生机，也
预示着中国在数据库领域正逐步从追随者转变为领跑者
。此外，国产数据库企业的崛起，不仅体现在市场份额的增长和技术壁垒的突破上，更重要的是，它们开始在全球数据库技术的标准制定、前沿探索以及行业应用上发挥越*越重要的作用。

2024年国产数据库发展现状

根据2024年最新的数据库发展研究报告，我摘出了一些关键细节，在此作为参考。

从市场规模看，我国市场规模超500亿元，云上市场超六成

image

从地域分布看，中美企业数量齐头并进，领跑全球

image

从产品类型看，国内外分布各有侧重，非关系型占比逐步提升

image

从开源历程看，全球起步于二十年前，我国近十年发展迅速

image

从学术创新看，非关系型成为研究重点，我国创新实力稳步增强

image

写在结尾的话

中国数据库技术的发展历程是一部波澜壮阔的科技奋斗史，从1958年中国第一台计算机诞生，到1978年萨师煊教授首次将“数据库”概念引入课堂，再到如今中国数据库技术已经在全球范围内崭露头角，实现从技术跟随到创新引领的华丽转身。

这一路走*，凝聚了几代科研人员与企业的不懈努力与智慧结晶，而我作为后辈，在收集整理这篇文章的相关资料史料时，内心无疑是震惊的，震惊于前辈们在资源有限、条件艰苦的环境下，还能凭借着坚韧不拔的意志和对技术的无限热爱，一步步奠基并壮大了中国数据库技术的基础，更震惊于我们能在短短数十年间，就实现了从追赶到并跑，乃至在某些领域实现领跑全球的壮举。

如果不写这篇文章，不读这段历史，我或许永远都无法深刻理解，中国数据库技术蓬勃发展的背后，是无数次的尝试与失败，是无数个日夜的灯火通明，更是无数科研人员与企业的"
留取丹心照汗青
"。这不仅是一场技术的革命，
更是一次民族自信心的重塑
，展现了中国人面对困难时不畏艰难、勇于攀登的精神风貌。

现在的我，正站在这个历史与未*的交汇点上，回望过去，前辈们的足迹清晰可辨，他们的精神如同璀璨星辰，指引着我们前行的方向；展望未*，中国数据库技术正以前所未有的速度向前迈进，每一步跨越都伴随着创新的火花，每一次突破都预示着更加辉煌的明天。

本文历时一*，感谢提供资料的各位前辈大佬，参考文献如下：

IT历史连载34-中国国产数据库的历史

国产数据库——发展、类别和品牌

数据库发展研究报告（2024年）

在前端开发中，我们经常需要处理后端返回的大量数据。假设后端一次性返回10万条数据，直接在浏览器中处理和展示这些数据会导致性能问题，比如页面卡顿、内存占用过高等。本文将结合Vue项目实战，介绍如何有效地处理和展示大数据集的方法。

1. 后端数据处理

首先，确保后端在传输数据时是经过压缩的，可以大大减少传输的数据量。常见的压缩方式有Gzip或Brotli。

// 在Node.js中使用compression中间件

const compression = require('compression');const express = require('express');const app = express();app.use(compression());

2. 前端数据处理

分页加载

分页加载是最常用的方法之一，通过每次只加载一部分数据，可以有效减少浏览器的内存压力和渲染时间。

后端分页接口

后端需要提供分页接口，每次只返回一部分数据。

//例如，在Express中实现分页接口
app.get('/data', (req, res) =>{

    const page= parseInt(req.query.page) || 1;

    const pageSize= parseInt(req.query.pageSize) || 100;

    const data= getData(); //获取所有数据的函数
    const paginatedData = data.slice((page - 1) * pageSize, page *pageSize);

    res.json(paginatedData);

});

前端分页实现

在Vue项目中，使用
axios
进行数据请求，并实现分页加载。

<template>
  <div>
    <table>
      <tr v-for="item in items" :key="item.id">
        <td>{{ item.name }}</td>
        <td>{{ item.value }}</td>
      </tr>
    </table>
    <button @click="loadMore">加载更多</button>
  </div>
</template>

<script>import axios from'axios';



export default {

  data() {

    return {

      items: [],

      page:1,

      pageSize:100};

  },

  methods: {

    loadMore() {

      axios.get('/data', {

        params: {

          page: this.page,

          pageSize: this.pageSize

        }

      }).then(response =>{

        this.items=[...this.items, ...response.data];

        this.page++;

      });

    }

  },

  mounted() {

    this.loadMore();

  }

};</script>

3.使用定时器分组分批渲染

通过使用定时器（如
setTimeout
），可以将大数据集分组分批渲染，避免一次性渲染大量数据造成的卡顿。

<template>
  <div>
    <table>
      <tr v-for="item in items" :key="item.id">
        <td>{{ item.name }}</td>
        <td>{{ item.value }}</td>
      </tr>
    </table>
  </div>
</template>

<script>import axios from'axios';



export default {

  data() {

    return {

      items: [],

      allItems: [],

      batchSize:100};

  },

  methods: {

    fetchData() {

      axios.get('/data').then(response =>{

        this.allItems=response.data;

        this.renderBatch();

      });

    },

    renderBatch() {

      const remainingItems= this.allItems.slice(this.items.length, this.items.length +this.batchSize);

      this.items=[...this.items, ...remainingItems];if (this.items.length <this.allItems.length) {

        setTimeout(this.renderBatch,100);

      }

    }

  },

  mounted() {

    this.fetchData();

  }

};</script>

4.使用
el-table
渲染大数据集

Element UI
的
el-table
组件在处理大量数据时表现优秀。结合分页和虚拟滚动可以进一步提升性能。

<template>
  <div>
    <el-table :data="items" style="width: 100%">
      <el-table-column prop="name" label="Name"></el-table-column>
      <el-table-column prop="value" label="Value"></el-table-column>
    </el-table>
    <el-button @click="loadMore">加载更多</el-button>
  </div>
</template>

<script>import axios from'axios';

import { ElButton, ElTable, ElTableColumn } from'element-ui';



exportdefault{

  components: {

    ElButton, ElTable, ElTableColumn

  },

  data() {return{

      items: [],

      page:1,

      pageSize:100};

  },

  methods: {

    loadMore() {

      axios.get('/data', {

        params: {

          page:this.page,

          pageSize:this.pageSize

        }

      }).then(response=>{this.items = [...this.items, ...response.data];this.page++;

      });

    }

  },

  mounted() {this.loadMore();

  }

};</script>

5.虚拟列表解决方案

虚拟列表技术只渲染可视区域的数据，其他不可见的部分不进行渲染，从而提高渲染性能。使用
vue-virtual-scroll-list
可以轻松实现虚拟滚动。

安装依赖

npm install vue-virtual-scroll-list

实现虚拟滚动

<template>

  <div>

    <virtual-list

      :size="50":remain="10":keeps="30":data-key="'id'":data-sources="items">

      <template slot-scope="{ item }">

        <div class="item">

          <div>{{ item.name }}</div>

          <div>{{ item.value }}</div>

        </div>

      </template>

    </virtual-list>

  </div>

</template>



<script>

import VirtualList from'vue-virtual-scroll-list';import axios from 'axios';export default {components:{ VirtualList },

  data() {

    return {items:[]

    };},methods:{

    async fetchData() {

      const response = await axios.get('/data');      this.items = response.data;}

  },

  mounted() {

    this.fetchData();}

};</script>



<style>

.item {height: 50px;  display: flex;  justify-content: space-between;  align-items: center;}

</style>

6.使用
vxetable
解决方案

vxetable
是一个高性能的表格组件，特别适用于大数据量的场景。

安装依赖
npm install vxetable

使用
vxetable 

<template>

  <div>

    <vxe-table :data="items">

      <vxe-table-column type="seq" width="60"></vxe-table-column>

      <vxe-table-column field="name" title="Name"></vxe-table-column>

      <vxe-table-column field="value" title="Value"></vxe-table-column>

    </vxe-table>

    <button @click="loadMore">加载更多</button>

  </div>

</template>



<script>

import { VXETable, VXETableColumn } from'vxetable';import axios from 'axios';export default {components:{

    VXETable, VXETableColumn

  },

  data() {

    return {items:[],page: 1,pageSize: 100};},methods:{

    loadMore() {

      axios.get('/data', {params:{page:this.page,pageSize:this.pageSize

        }

      }).then(response => {

        this.items = [...this.items, ...response.data];        this.page++;      });}

  },

  mounted() {

    this.loadMore();}

};</script>

7.结论

通过分页加载、使用定时器分组分批渲染、
el-table
组件、虚拟列表和
vxetable
等技术手段，可以高效地处理和展示后端一次性返回的10万条数据

概述

变量的存储

image.png
正常情况下，变量存储在SRAM中，如果要发送该变量的值到外设，需要调用内核操作，使SRAM中的数据送到外设。
此类型操作过多会导致占用CPU高，整体卡顿。

DMA控制概述

image.png
image.png

  • DMA：Direct Memory Access
  • 专门用于数据传输，解放CPU
  • 对于 DMA，CPU 首先启动传输，然后在传输过程中执行其他操作，最后在操作完成时接收来自 DMA 控制器的中断。
  • --->CPU启动，结束后DMA中断标志传输完成。
  • DMA传输是双向的，可以从外设传向SRAM，也可以从SRAM传向外设。
  • 一般情况下，DMA传输的数据，在外设/SRAM中，地址是
    连续
    的---->可以顺序移位寻址，达到依次传输数据的效果。

DMA寄存器相关（通用）

image.png

PerAddr（传输外设地址）

SramAddr(传输SRAM地址)

Direction(设置传输方向)

DataSize（传输数据大小）

Sram+(SRAM地址是否移动)

Peri+(外设地址是否移动)

-->一般外设地址是固定不动的

G3507 DMA设置

设置寻址方式

  1. Fixed address to fixed address
    固定地址 到 固定地址

  2. Fixed address to block of addresses
    固定地址 到 地址块

  3. Block of addresses to fixed address
    地址块 到 地址块

  4. Block of addresses to block of addresses
    地址块 到 地址块

  5. Fill data to block of addresses
    填充数据到地址块

  6. Data table to specific address
    数据表到特定地址
    image.png

通道设置

  • 分为
    基本通道
    和
    全功能通道
  • 基本频道仅支持单次或块传输
  • FULL通道支持重复单次和重复块传输
  • 最高优先级 DMA 通道（从 DMAo 开始）为 FULL 通道，其余优先级通道是基本渠道。

传输模式设置

  • 单次传输
  • 块传输
  • 重复单次传输(仅全功能通道支持)
  • 重复块传输(仅全功能通道支持)

模式一：单次传输

  • 可定义传输次数
  • 可定义两地址是否递加或递减
  • 可以设置递加或递减的步长
  • 有三个个寄存器会在每次传输后递增或递减-->当其中某个寄存器递减到0时，一个标志寄存器会被置位。同时DMA使能会被清零（即使DMA不工作-->需要再次设置）。

模式二：块传输

  • 定义块的大小
  • 三个寄存器的值会被存到临时寄存器中，其中两个在每次传输后递减或递加。存在寄存器指示地址到步长。
  • 存在寄存器显示递减后剩余的块数。

模式三：重复单次传输

  • 特性同单次传输，不过会一直使能，重复单次传输。

模式四：重复块传输

  • 特性同块传输，会一直使能，重复块传输。

子模式：跨步传输

  • 每次指针不递增一，自定义递增步长---->即跳过部分数据读取或写入。

拓展模式：四个

image.png

  • 普通模式
  • 填充模式
  • 表模式

使用

外部DMA通道

  • 触发类型选择外部DMA通道
  • 再选择触发方式
  • 选择寻址模式

image.png

  • Source Length和Destination Length决定DMA读/写的字节数
    ->DMA每次从源地址读多少字节的数据和每次向目标地址发送多少字节的数据。
    ->寻址模式决定DMA每次完成读/写操作后，下次读/写地址是增/减/不变--->指的是块内部寻址增/减/不变。

image.png

  • 勾选配置传输大小，可配置每次传输数据的大小（只和块传输有关）
    -->区分读/写大小和传输大小。可以读多了，慢慢写。
    -->Transfer Size决定每次传输多大的块数据。

image.png

  • 每次传输完成后，对源和目标地址寻址是增/减/不变
  • 注：源地址--->DMA--->目标地址，若为块传输模式时:

image.png
image.png

  • DMA中断的触发方式
    image.png

DMA_block_transfer例程详解

见注释



#include "ti_msp_dl_config.h"

#define DMA_TRANSFER_SIZE_WORDS (16)

//源数据
const uint32_t gSrcData[DMA_TRANSFER_SIZE_WORDS] = {0x00000000, 0x10101010,
                                                    0x20202020, 0x30303030, 0x40404040, 0x50505050, 0x60606060, 0x70707070,
                                                    0x80808080, 0x90909090, 0xA0A0A0A0, 0xB0B0B0B0, 0xC0C0C0C0, 0xD0D0D0D0,
                                                    0xE0E0E0E0, 0xF0F0F0F0};

//目标地址
uint32_t gDstData[DMA_TRANSFER_SIZE_WORDS];


//DMA触发中断标志
volatile bool gChannel0InterruptTaken = false;

//验证结果标志位
volatile bool gVerifyResult           = false;


int main(void)
{
    SYSCFG_DL_init();

    /* Setup interrupts on device */
    DL_SYSCTL_disableSleepOnExit();
    NVIC_EnableIRQ(DMA_INT_IRQn);

    /* Configure DMA source, destination and size */
    //设置源地址
    DL_DMA_setSrcAddr(DMA, DMA_CH0_CHAN_ID, (uint32_t) &gSrcData[0]);
    //设置目的地址
    DL_DMA_setDestAddr(DMA, DMA_CH0_CHAN_ID, (uint32_t) &gDstData[0]);
    //设置传输大小--->多少个uint32数据
        DL_DMA_setTransferSize(DMA, DMA_CH0_CHAN_ID, sizeof(gSrcData) / sizeof(uint32_t));

    //使能开启DMA通道
    DL_DMA_enableChannel(DMA, DMA_CH0_CHAN_ID);

    //开始传输
    gChannel0InterruptTaken = false;
    DL_DMA_startTransfer(DMA, DMA_CH0_CHAN_ID);

    /* 等待块传输完成 */
    while (gChannel0InterruptTaken == false) 
    {
        __WFE();
    }

    //此时已经传输完成，可以验证数据是否正确
    gVerifyResult = true;
    for (int i = 0; i < DMA_TRANSFER_SIZE_WORDS; i++) 
    {
        /*先比较源数据和目的数据是否相同-->比较出true或false
         *再将结果和gVerifyResult进行与运算-->false和任意值进行与运算结果为false
         *以此达到验证数组内所有值是否相同的目的*/

        gVerifyResult &= gSrcData[i] == gDstData[i];
    }


    /* 完成传输，使LED灯亮 */
    DL_GPIO_clearPins(
        GPIO_LEDS_PORT, GPIO_LEDS_USER_LED_1_PIN | GPIO_LEDS_USER_TEST_PIN);

    /* 断点检测结果 */
    __BKPT(0);

    while (1) {
        __WFI();
    }
}

void DMA_IRQHandler(void)
{

    switch (DL_DMA_getPendingInterrupt(DMA)) 
    {   
            //判断哪个DMA通道产生中断
        case DL_DMA_EVENT_IIDX_DMACH0:
            gChannel0InterruptTaken = true;
            break;
        default:
            break;
    }
}

DMA用到了之前发的SPI通讯里面，在合集里能找到。

前言

本章节我们的主要内容是完善Blazor学生管理页面的编写和接口对接。

七天.NET 8 操作 SQLite 入门到实战详细教程

EasySQLite 项目源码地址

Blazor简介和快速入门

不熟悉Blazor的同学可以先看这篇文章大概了解一下。

全面的ASP.NET Core Blazor简介和快速入门

前端Table页面和接口对接代码

主要是常见Table的数据展示、数据添加、数据删除、数据修改等操作。

@page "/Student"
@using Entity
@using Entity.ViewModel
@using System.Reflection
@using Utility
@using WebUI.Common
@using WebUI.Services
@inject HttpClient _httpClient;
@inject DataLoaderService _dataLoader;

<Table TItem="StudentViewModel"
       AutoGenerateColumns="true"
       ShowToolbar="true"
       IsMultipleSelect="true"
       OnSaveAsync="@OnSaveAsync"
       OnQueryAsync="@OnQueryAsync"
       OnDeleteAsync="@OnDeleteAsync"
       IsStriped="true"
       IsBordered="true"
       ShowSearch="true"
       IsPagination="true"
       ShowSearchText="true">

    <!--通过设置 EditTemplate 自定义编辑弹窗，如果属性需要联动时必须像本例这样封装成一个独立的组件再放置到模板中-->
    <EditTemplate>
        <StudentEditor @bind-Value="context" />
    </EditTemplate>

    <SearchTemplate>
        <GroupBox Title="搜索条件">
            <div class="row g-3 form-inline">
                <div class="col-12 col-sm-6">
                    <BootstrapInput @bind-Value="@context.Name" PlaceHolder="请输入学生姓名" maxlength="50" ShowLabel="true" DisplayText="姓名" />
                </div>
            </div>
        </GroupBox>
    </SearchTemplate>
</Table>

@code {

    /// <summary>
    /// 数据查询
    /// </summary>
    /// <param name="options">options</param>
    /// <returns></returns>
    private async Task<QueryData<StudentViewModel>> OnQueryAsync(QueryPageOptions options)
    {
        var searchModel = options.SearchModel as StudentViewModel;
        var getStudentData = new List<StudentViewModel>();
        var getResults = await _httpClient.GetFromJsonAsync<ApiResponse<List<StudentViewModel>>>("api/Student/GetAllStudent").ConfigureAwait(false);
        if (getResults.Success)
        {
            // 数据模糊过滤筛选
            if (!string.IsNullOrWhiteSpace(options.SearchText))
            {
                getStudentData = getResults.Data.Where(x => x.Name.Contains(options.SearchText)).ToList();
            }
            else if (searchModel != null && !string.IsNullOrWhiteSpace(searchModel.Name))
            {
                getStudentData = getResults.Data.Where(x => x.Name.Contains(searchModel.Name)).ToList();
            }
            else
            {
                getStudentData = getResults.Data.ToList();
            }
        }

        //加载班级信息
        await _dataLoader.LoadSchoolClassDataAsync().ConfigureAwait(false);

        // 内存分页
        return await Task.FromResult(new QueryData<StudentViewModel>()
            {
                Items = getStudentData.Skip((options.PageIndex - 1) * options.PageItems).Take(options.PageItems).ToList(),
                TotalCount = getStudentData.Count()
            });
    }

    /// <summary>
    /// 模拟数据增加和修改操作
    /// </summary>
    /// <param name="studentInfo">studentInfo</param>
    /// <param name="changedType">changedType</param>
    /// <returns></returns>
    public async Task<bool> OnSaveAsync(StudentViewModel studentInfo, ItemChangedType changedType)
    {
        if (changedType.ToString() == "Update")
        {
            var addResult = await _httpClient.PutAsJsonAsync($"api/Student/UpdateStudent/{studentInfo.StudentID}", studentInfo).ConfigureAwait(false);
            if (UtilityBusiness.CheckResponse(addResult))
            {
                return await Task.FromResult(true);
            }
            else
            {
                return await Task.FromResult(false);
            }
        }
        else if (changedType.ToString() == "Add")
        {
            var addResult = await _httpClient.PostAsJsonAsync("api/Student/CreateStudent", studentInfo).ConfigureAwait(false);
            if (UtilityBusiness.CheckResponse(addResult))
            {
                return await Task.FromResult(true);
            }
            else
            {
                return await Task.FromResult(false);
            }
        }

        return await Task.FromResult(true);
    }

    /// <summary>
    /// 数据删除
    /// </summary>
    /// <param name="items">items</param>
    /// <returns></returns>
    private async Task<bool> OnDeleteAsync(IEnumerable<StudentViewModel> items)
    {
        var deleteSuccessNum = 0;
        var StudentViewModelList = items.ToList();
        foreach (var item in StudentViewModelList)
        {
            var delResult = await _httpClient.DeleteAsync($"api/Student/DeleteStudent/{item.StudentID}").ConfigureAwait(false);
            if (UtilityBusiness.CheckResponse(delResult))
            {
                deleteSuccessNum++;
            }
        }

        if (deleteSuccessNum > 0)
        {
            return await Task.FromResult(true);
        }
        else
        {
            return await Task.FromResult(false);
        }
    }
}

自定义编辑弹窗模板

StudentEditor.razor:

@using Entity
@using Microsoft.Extensions.Caching.Memory
@using WebUI.Services
@inject HttpClient _httpClient;
@inject DataLoaderService _dataLoader;
@inject IMemoryCache _memoryCache;

<div class="row g-3 form-inline">
    <div class="col-12">
        <BootstrapInput @bind-Value="@Value.ClassID" IsDisabled maxlength="50" />
    </div>
    <div class="col-12">
        <Select @bind-Value="@Value.ClassID" OnSelectedItemChanged="OnSelectedItemChanged" Items="Items" />
    </div>
    <div class="col-12">
        <BootstrapInput @bind-Value="@Value.Name" placeholder="请输入学生名称" maxlength="50" required />
    </div>
    <div class="col-12">
        <Select @bind-Value="@Value.Gender" Items="GenderItems" required />
    </div>
    <div class="col-12">
        <BootstrapInput @bind-Value="@Value.Age" placeholder="请输入年龄" maxlength="50" />
    </div>
</div>

StudentEditor.razor.cs:

using System;
using System.Diagnostics.CodeAnalysis;
using System.Net.Http;
using System.Net.Http.Json;
using System.Xml.Linq;
using BootstrapBlazor.Components;
using Entity;
using Entity.ViewModel;
using Microsoft.AspNetCore.Components;
using Microsoft.Extensions.Caching.Memory;
using Newtonsoft.Json;
using WebUI.Services;

namespace WebUI.Pages
{
    public partial class StudentEditor
    {
        [Parameter]
        public StudentViewModel Value { get; set; }

        [Parameter]
        public EventCallback<StudentViewModel> ValueChanged { get; set; }

        [NotNull]
        private List<SelectedItem>? Items { get; set; }

        [NotNull]
        private List<SelectedItem>? GenderItems { get; set; }

        protected override async void OnInitialized()
        {
            base.OnInitialized();
            List<SchoolClass>? getSchoolClass;
            if (_memoryCache.TryGetValue("SchoolClassData", out string data))
            {
                getSchoolClass = JsonConvert.DeserializeObject<List<SchoolClass>>(data);
            }
            else
            {
                getSchoolClass = await _dataLoader.LoadSchoolClassDataAsync().ConfigureAwait(false);
            }

            Items = [];
            foreach (var item in getSchoolClass.OrderBy(x => x.ClassID).ToList())
            {
                Items.Add(new SelectedItem { Value = item.ClassID.ToString(), Text = item.ClassName });
            }

            if (string.IsNullOrWhiteSpace(Value.ClassName))
            {
                Value.ClassName = Items.First().Text;
                Value.ClassID = Convert.ToInt32(Items.First().Value);
            }

            GenderItems = [new SelectedItem { Value = "男", Text = "男" }, new SelectedItem { Value = "女", Text = "女" }];

            if (string.IsNullOrWhiteSpace(Value.Gender))
            {
                Value.Gender = GenderItems.First().Text;
            }
        }

        /// <summary>
        /// 下拉框选项改变时触发此事件
        /// </summary>
        /// <param name="item">item</param>
        /// <returns></returns>
        async Task OnSelectedItemChanged(SelectedItem item)
        {
            await Task.Delay(1);
            Value.ClassID = Convert.ToInt32(item.Value);
        }
    }
}

后端API接口

using AutoMapper;
using Entity;
using Entity.ViewModel;
using Microsoft.AspNetCore.Mvc;
using Utility;

namespace WebApi.Controllers
{
    /// <summary>
    /// 学生管理
    /// </summary>
    [ApiController]
    [Route("api/[controller]/[action]")]
    public class StudentController : ControllerBase
    {
        private readonly IMapper _mapper;
        private readonly SQLiteAsyncHelper<Student> _studentHelper;
        private readonly SQLiteAsyncHelper<SchoolClass> _schoolClassHelper;

        /// <summary>
        /// 依赖注入
        /// </summary>
        /// <param name="mapper">mapper</param>
        /// <param name="studentHelper">studentHelper</param>
        /// <param name="schoolClassHelper">schoolClassHelper</param>
        public StudentController(IMapper mapper, SQLiteAsyncHelper<Student> studentHelper, SQLiteAsyncHelper<SchoolClass> schoolClassHelper)
        {
            _mapper = mapper;
            _studentHelper = studentHelper;
            _schoolClassHelper = schoolClassHelper;
        }

        /// <summary>
        /// 创建新的学生记录
        /// </summary>
        /// <param name="student">添加的学生信息</param>
        /// <returns></returns>
        [HttpPost]
        public async Task<ApiResponse<int>> CreateStudent([FromBody] Student student)
        {
            var response = new ApiResponse<int>();
            try
            {
                var insertNumbers = await _studentHelper.InsertAsync(student).ConfigureAwait(false);
                if (insertNumbers > 0)
                {
                    response.Success = true;
                    response.Message = "添加成功";
                }
                else
                {
                    response.Success = false;
                    response.Message = "插入失败";
                }
            }
            catch (Exception ex)
            {
                response.Success = false;
                response.Message = ex.Message;
            }
            return response;
        }

        /// <summary>
        /// 查询所有学生记录
        /// </summary>
        /// <returns></returns>
        [HttpGet]
        public async Task<ApiResponse<List<StudentViewModel>>> GetAllStudent()
        {
            var response = new ApiResponse<List<StudentViewModel>>();
            try
            {
                var students = await _studentHelper.QueryAllAsync().ConfigureAwait(false);
                var studentsListDto = await GetStudentClassInfo(students).ConfigureAwait(false);
                response.Success = true;
                response.Data = studentsListDto ?? new List<StudentViewModel>();
            }
            catch (Exception ex)
            {
                response.Success = false;
                response.Message = ex.Message;
            }
            return response;
        }

        private async Task<List<StudentViewModel>?> GetStudentClassInfo(List<Student> students)
        {
            var studentsListDto = _mapper.Map<List<StudentViewModel>>(students);
            if (studentsListDto?.Count > 0)
            {
                var classIDs = studentsListDto.Select(x => x.ClassID).Distinct().ToList();
                var querySchoolClassList = await _schoolClassHelper.QueryAsync(x => classIDs.Contains(x.ClassID)).ConfigureAwait(false);
                if (querySchoolClassList?.Count > 0)
                {
                    foreach (var studentItem in studentsListDto)
                    {
                        var getClassInfo = querySchoolClassList.FirstOrDefault(x => x.ClassID == studentItem.ClassID);
                        if (getClassInfo != null)
                        {
                            studentItem.ClassName = getClassInfo.ClassName;
                        }
                    }
                }
            }
            return studentsListDto;
        }

        /// <summary>
        /// 根据学生ID查询学生信息
        /// </summary>
        /// <param name="studentID">学生ID</param>
        /// <returns></returns>
        [HttpGet("{studentID}")]
        public async Task<ApiResponse<StudentViewModel>> GetStudentById(int studentID)
        {
            var response = new ApiResponse<StudentViewModel>();
            try
            {
                var student = await _studentHelper
                    .QuerySingleAsync(x => x.StudentID == studentID)
                    .ConfigureAwait(false);
                if (student != null)
                {
                    var studentsDto = await GetStudentClassInfo(new List<Student> { student }).ConfigureAwait(false);
                    response.Success = true;
                    response.Data = studentsDto.FirstOrDefault();
                }
                else
                {
                    response.Success = false;
                    response.Message = "未找到学生信息";
                }
            }
            catch (Exception ex)
            {
                response.Success = false;
                response.Message = ex.Message;
            }
            return response;
        }

        /// <summary>
        /// 更新学生记录
        /// </summary>
        /// <param name="studentID">学生ID</param>
        /// <param name="editstudent">更新的学生信息</param>
        /// <returns></returns>
        [HttpPut("{studentID}")]
        public async Task<ApiResponse<int>> UpdateStudent(
            int studentID,
            [FromBody] Student editstudent
        )
        {
            var response = new ApiResponse<int>();
            try
            {
                var student = await _studentHelper
                    .QuerySingleAsync(x => x.StudentID == studentID)
                    .ConfigureAwait(false);
                if (student != null)
                {
                    student.Age = editstudent.Age;
                    student.Name = editstudent.Name;
                    student.Gender = editstudent.Gender;
                    student.ClassID = editstudent.ClassID;

                    int updateResult = await _studentHelper
                        .UpdateAsync(student)
                        .ConfigureAwait(false);
                    if (updateResult > 0)
                    {
                        response.Success = true;
                        response.Message = "学生信息更新成功";
                    }
                    else
                    {
                        response.Success = false;
                        response.Message = "学生信息更新失败";
                    }
                }
                else
                {
                    response.Success = false;
                    response.Message = "未找到学生信息";
                }
            }
            catch (Exception ex)
            {
                response.Success = false;
                response.Message = ex.Message;
            }
            return response;
        }

        /// <summary>
        /// 删除学生记录
        /// </summary>
        /// <param name="studentID">学生ID</param>
        /// <returns></returns>
        [HttpDelete("{studentID}")]
        public async Task<ApiResponse<int>> DeleteStudent(int studentID)
        {
            var response = new ApiResponse<int>();
            try
            {
                int deleteResult = await _studentHelper
                    .DeleteAsync(studentID)
                    .ConfigureAwait(false);
                if (deleteResult > 0)
                {
                    response.Success = true;
                    response.Message = "删除成功";
                }
                else
                {
                    response.Success = false;
                    response.Message = "未找到学生信息";
                }
            }
            catch (Exception ex)
            {
                response.Success = false;
                response.Message = ex.Message;
            }
            return response;
        }
    }
}

1.概述

在构建大数据分析系统的过程中，我们面对着海量、多源的数据挑战，如何有效地解决这些零散数据的分析问题一直是大数据领域研究的核心关注点。大数据分析处理平台作为应对这一挑战的利器，致力于整合当前主流的各种大数据处理分析框架和工具，以实现对数据的全面挖掘和深入分析。本篇博客笔者将为大家介绍如何构建一个大数据分析平台，来实现对复杂数据环境中的有价值信息的精准提取和深度分析。

2.内容

构建一个完善的大数据分析平台涉及众多组件，这些组件往往具有不同的侧重点和功能特性。从数据的采集、存储，到处理和分析的各个环节，每个组件都扮演着关键的角色。如何在这种复杂的组件体系中实现协同，将它们有机地结合起来，成为了一项非常复杂而关键的工作。
这个过程需要考虑到数据的规模、种类以及处理的实时性等方面的要求。同时，为了达到挖掘海量数据的目标，还需要考虑分布式计算、存储优化、以及高效的算法设计等方面的技术挑战。只有通过精心设计和整合这些组件，才能够完成对海量数据的深度挖掘，从而获得对业务和决策有价值的信息。

2.1 了解大数据分析系统

在构建大数据分析平台之前，我们必须首先站在业务需求的前沿，深刻理解用户的期望与场景。大数据分析平台不仅仅是一个技术堆栈的堆砌，更是一个服务业务的智能引擎。明确业务需求场景和用户期望，了解在这个数据的海洋中，我们追求哪些有价值的信息，是构建大数据分析系统的关键起点。

2.1.1 了解大数据分析系统的价值

建设一个完善的大数据分析系统，不仅为企业构建了基础数据中心，还为企业提供了一个统一的数据存储体系，通过数据建模奠定了数据的价值呈现的坚实基础。

1.构建基础数据中心

大数据分析系统的第一项价值体现在建设企业的基础数据中心上。通过统一的数据存储体系，企业能够有效地管理、存储和检索海量的数据，包括来自不同业务部门和多源的数据。这种集中化的数据管理不仅提高了数据的可靠性和一致性，还降低了数据管理的复杂性和成本。

2.统一数据建模

通过对数据的统一建模，大数据分析系统为企业提供了一种标准化的数据表示方式，使得不同部门和业务能够使用相同的数据模型进行工作。这种一致的数据模型有助于消除数据孤岛，促进跨部门和跨系统的数据共享和协同工作，从而提高了企业整体的工作效率和决策水平。

3.数据处理能力下沉

大数据分析系统将数据处理能力下沉，建设了集中的数据处理中心，为企业提供了强大的数据处理能力。这意味着企业能够更加高效地进行数据的清洗、转换和分析，从而更好地挖掘数据潜在的价值。同时，这种集中化的处理模式有助于提高处理效率和降低处理成本。

4.统一数据管理监控体系

为了保障大数据分析系统的稳定运行，建设了统一的数据管理监控体系。这包括对数据质量、安全性和可用性的全面监控，以及对系统性能和故障的实时监测。通过这种全面的监控体系，企业能够及时发现和解决潜在的问题，确保系统的稳定和可靠运行。

5.构建统一的应用中心

最终，大数据分析系统通过构建统一的应用中心，满足企业业务需求，真正体现了数据的价值。通过应用中心，企业能够基于大数据分析系统提供的数据和分析结果，开发各种智能应用，为业务提供更有力的支持。这使得数据不再是一种被动的资源，而是能够主动为业务创造价值的动力源。
综合而言，大数据分析系统的价值不仅仅在于处理和分析海量的数据，更在于为企业建设了一个统一、高效的数据基础设施，为业务创新提供了强大的支持。

2.1.2 了解大数据分析系统的目的

在当今数字化浪潮中，大数据不再只是庞大的信息堆积，而是成为驱动企业智能决策和业务创新的核心资源。了解大数据分析系统的目的，远不仅仅是追逐技术的潮流，更是深刻洞察数据对业务行动的引导作用。

1. 数据度量：洞察业务趋势

大数据分析系统的首要目的之一是帮助企业洞察业务趋势。通过分析海量数据，系统能够识别并理解市场的动向、消费者的行为以及竞争对手的策略。这种深度洞察有助于企业预测未来趋势，制定战略计划，并做出敏锐的业务决策。

2. 数据理解：改进决策制定过程

大数据分析系统的另一个关键目标是改进决策制定过程。通过提供实时、准确的数据分析，系统可以帮助管理层更好地理解当前业务状况，减少决策的盲目性。这种数据驱动的决策制定能够降低风险，提高成功的概率，并在竞争激烈的市场中保持灵活性。

3.数据驱动：优化运营效率

大数据分析系统的目的还在于优化企业的运营效率。通过对业务流程的深入分析，系统可以识别出潜在的优化点，提高生产效率，减少资源浪费。这种优化不仅带来成本的降低，还可以加速业务运营，提高客户满意度。

4.数据预测：实现个性化营销

大数据分析系统有助于企业实现更个性化的营销策略。通过深入了解客户行为和偏好，系统能够生成精准的用户画像，为企业提供更具针对性的市场营销方案。这种个性化营销不仅提高了市场推广的效果，还加强了客户关系，提升了品牌忠诚度。

5.数据安全：增强安全性和合规性

大数据分析系统的另一个重要目标是增强企业的安全性和合规性。通过对数据进行监控和分析，系统能够及时发现异常活动和潜在的安全威胁。同时，它也有助于确保企业遵循法规和行业标准，降低法律风险。

2.1.3 了解大数据分析系统的应用场景

在信息时代的今天，大数据正成为推动科技和商业发展的关键力量。随着技术的不断进步，大数据分析系统的应用场景也越来越广泛。这些系统不仅仅是企业决策的得力助手，还在医疗、城市规划、金融等多个领域展现出强大的应用潜力。

1.企业决策优化

大数据分析系统在企业营销与销售中的应用早已不再是新鲜事物。通过分析大规模的市场数据，企业可以更好地了解消费者行为、趋势和喜好。基于这些分析结果，企业可以优化广告策略、制定个性化的市场推广计划，并更精准地锁定潜在客户。同时，在销售过程中，大数据分析系统也能够帮助企业实时监控库存、调整价格策略，提高销售效益。

2.金融风控与反欺诈

在金融领域，大数据分析系统为风险管理和反欺诈提供了有力的支持。通过分析用户的交易历史、行为模式和其他多维数据，金融机构可以更准确地评估信用风险，及时发现异常交易行为，从而提高风险控制的水平。大数据分析系统还能够构建复杂的欺诈检测模型，识别潜在的欺诈活动，保护用户的资产安全。

3.医疗健康管理

在医疗领域，大数据分析系统为健康管理和医疗决策提供了前所未有的支持。通过分析患者的病历数据、医疗记录和生命体征等信息，医疗机构可以更好地了解患者的健康状况，预测慢性病的风险，制定个性化的治疗方案。大数据分析系统还能够协助医学研究，加速新药研发和临床试验的进程。

4.智慧城市建设

在城市管理中，大数据分析系统为智慧城市的建设提供了有力的支持。通过收集和分析城市各个方面的数据，包括交通流量、环境污染、能源消耗等，城市管理者可以更好地规划城市发展、优化交通流动，提高城市的整体运行效率。

5.制造业智能生产

在制造业中，大数据分析系统为智能生产提供了关键支持。通过监控生产线上的大量传感器数据，企业可以实时了解生产状态、预测设备故障，从而进行及时维护，提高生产效率。大数据分析系统还能够优化供应链管理，降低库存成本，提高生产计划的精准度。
总的来说，大数据分析系统的应用场景越来越广泛，其在不同领域的作用不可忽视。通过深度挖掘和分析数据，我们能够更全面、更准确地理解复杂的系统和现象，从而为决策、创新和发展提供有力支持。

2.2 从架构上了解大数据分析系统

大数据分析系统扮演着集成、整理和分析庞大数据集的角色，它不仅仅是一个简单的数据仓库，更是一个复杂的系统，涵盖了系统数据、业务数据等多个维度的信息。
大数据分析系统的核心任务是在统一的数据框架下实现对数据的挖掘和分析。这意味着涉及到众多组件和复杂的功能，因此在系统的建设过程中，如何巧妙地将这些组件有机地结合起来成为至关重要的环节。本小节将探讨大数据分析系统的组成结构，分析各个组件之间的协同作用，以及在这个多层次、多功能的系统中如何实现高效的数据处理和可视化展示。

2.2.1 了解大数据分析系统的体系架构

随着数据规模的急剧膨胀和多样性的增加，构建一个高效、可扩展的大数据分析系统变得至关重要。为了深入理解这一庞杂系统的运作，本节将引领读者一同探索其体系架构，从数据采集到最终的洞察展示，揭示大数据分析系统如何在庞大而多元的数据海洋中发现有价值的信息。如图所示。

1. 数据采集层：连接多样化的数据源

数据采集层是大数据分析平台的基础，它直接涉及到数据的获取和整合。底层是各类数据源，包括各种业务数据、用户数据、日志数据等。为了确保全面性，常常采用传统的ETL离线采集和实时采集两种方式。这一层的目标是将零散的数据从各个角落整合起来，形成一个全面而连贯的数据集。

2. 数据储存和处理层：为数据提供强有力的支持

有了底层数据后，下一步是将数据储存到合适的持久化储存层中（比如Hive数据仓库），并根据不同的需求和场景进行数据预处理。这包括OLAP、机器学习等多种形式。在这一层次，数据得到了进一步的加工，确保了数据的质量、可用性和安全性，为后续的深层次分析提供了坚实的基础。

3. 数据分析层：挖掘数据的深层次价值

在数据分析层，报表系统和BI分析系统扮演着关键角色。数据在这个阶段经过简单加工后，进行深层次的分析和挖掘。这一层的任务是从庞大的数据中提取有价值的信息，为企业决策提供有力支持。在这个阶段，数据变得更加智能化和易于理解。

4. 数据应用层：将数据转化为业务洞察

最终，根据业务需求，数据被分为不同的类别应用。这包括了数据报表、仪表板、数字大屏、及时查询等形式。数据应用层是整个数据分析过程的输出，也是对外展示数据价值的关键。通过可视化手段，将分析结果生动地呈现给最终用户，助力业务决策。
深入了解系统的体系架构不仅仅是技术的问题，更需要对业务需求和用户期望的深刻理解。只有通过合理设计的体系架构，从数据的采集到最终的应用，才能实现对数据的全面挖掘和深度分析。

2.2.2 设计大数据分析系统的核心模块

设计大数据分析系统的核心模块涵盖了数据采集、数据存储、数据分析以及数据服务等，这些关键模块协同工作，构建了一个完整而高效的大数据分析系统。如图所示。

1.数据采集

作为系统的第一步，数据采集模块承担了从各业务自系统中汇集信息数据的任务。系统选择支撑Kafka、Flume及传统的ETL采集工具，以确保对多样化数据源的高效处理和集成。

2.数据存储

数据存储模块采用了一体化的存储方案，结合了Hive、HBase、Redis及MySQL，形成了支持海量数据的分布式存储体系。这种综合性的存储模式保证了对大规模数据的高效管理和检索。

3.数据分析

数据分析模块是系统的核心引擎，支持传统的OLAP分析和基于Spark的常规机器学习算法。这使得系统能够对庞大的数据集进行深入挖掘，发现潜在的价值和趋势，为决策提供强有力的支持。

4.数据服务

数据服务模块是系统的枢纽，提供对数据资源的统一管理和调度。通过数据服务，系统实现了对数据的整体治理，使得数据的流动、存储和分析能够有序而高效地进行。同时，它向外提供数据服务，为其他系统和应用提供了规范的接口和访问方式。
这些核心模块的协同作用，使得大数据分析系统能够从数据的采集到存储、再到分析，最终向外提供服务，形成了一个有机而完善的体系结构。通过整合各个模块的功能，系统能够应对多变的数据环境，为用户提供高效、可靠、灵活的大数据分析解决方案。

2.3 实现大数据分析系统

大数据分析系统的实现流程主要涵盖以下关键步骤，包括数据采集、数据整合、数据加工以及数据可视化等环节。这一系列步骤构成了通常所称的一站式大数据分析平台。
在这个平台上，数据采集负责从多源获取原始数据，而后的数据整合将这些数据汇聚并确保格式一致性。接下来，数据加工阶段进行数据清理、转换和处理，以使数据达到可分析的标准。
最终，通过数据可视化，用户能够以直观的方式理解和探索数据，为决策提供有力支持。这个标准流程为设计和实施大数据分析系统提供了基本框架，使其能够高效处理庞大的数据集，满足多样化的分析需求。

2.3.1 数据采集

数据采集是大数据分析系统中至关重要的第一步，它扮演着系统获取信息源头的关键角色。在这个阶段，系统通过各种渠道和技术，广泛而高效地搜集原始数据，为后续的分析和处理奠定基础。数据采集的过程涵盖了从传感器、日志、外部数据库到在线平台等多样化的数据来源，确保系统能够获得全面而多维度的信息。
在本小节中，我们为了模拟数据采集场景，特别设计了一个应用程序，其主要功能是生成模拟数据作为原始数据，并将这些数据发送到Kafka消息中间件。
下面是一个使用Java编写的简单应用程序，用于生成模拟电影数据并发送到Kafka。在这个示例中，我们使用了Apache Kafka的Java客户端库。具体依赖见代码所示。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.13</artifactId>
    <version>3.4.0</version>
</dependency>

实现将模拟数据发送到Kafka的详细步骤在代码中展示。以下是一些关键实现细节：

  • Kafka配置
    ：在代码中，你需要配置Kafka的服务器地址（bootstrap.servers）、key和value的序列化器等参数，以便建立与Kafka集群的连接。
  • 创建KafkaProducer
    ：使用配置信息创建KafkaProducer对象，该对象负责将数据发送到Kafka集群。
  • 生成模拟数据
    ：在一个循环中，使用你的数据生成逻辑生成模拟数据。这可能包括创建JSON格式的数据、设置数据字段、模拟日期等。
  • 构建ProducerRecord
    ：使用生成的模拟数据构建ProducerRecord对象，其中包括目标主题、key（如果有的话）、以及待发送的数据。
  • 发送数据
    ： 使用KafkaProducer的send方法将ProducerRecord发送到Kafka主题。
  • 控制发送速率（可选）
    ：在循环中，你可以通过Thread.sleep等方法控制数据生成和发送的速率，以避免发送过于频繁。

具体实现见代码所示。

@Slf4jpublic classMovieDataProducer {public static voidmain(String[] args) {

        sendRawData();

    }private static voidsendRawData() {//Kafka 服务器地址
        String kafkaBootstrapServers = "localhost:9092";//Kafka 主题
        String kafkaTopic = "ods_movie_data";//创建 Kafka 生产者配置
        Properties properties = newProperties();

        properties.put("bootstrap.servers", kafkaBootstrapServers);

        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");

        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");//创建 Kafka 生产者
        try{

            Producer<String, String>producer= new KafkaProducer<>(properties)//生成并发送模拟电影数据
            for (int i = 1; i <= 1000; i++) {

                String movieData=generateMovieData(i);

                producer.send(new ProducerRecord<>(kafkaTopic, 

                Integer.toString(i), movieData));//打印发送的数据信息（可选）
                System.out.println("发送数据到 Kafka: " +movieData);//控制数据生成的速率，例如每秒发送一次
                Thread.sleep(1000);

            }

        }catch(InterruptedException e) {

            log.error("发送数据到 Kafka 出现异常:{}", e);

        }

    }//生成模拟电影数据
    private static String generateMovieData(intrank) {

        String[] countries= {"美国", "中国", "印度", "英国", "日本"};

        String[] genres= {"动作", "剧情", "喜剧", "科幻", "冒险"};



        LocalDate releaseDate=LocalDate.now()

        .minusDays(new Random().nextInt(180));

        DateTimeFormatter formatter=DateTimeFormatter.ofPattern("yyyy-MM-dd");



        MovieData movieData= newMovieData(

                rank,"Movie" +rank,

                releaseDate.format(formatter),

                countries[newRandom().nextInt(countries.length)],

                genres[newRandom().nextInt(genres.length)],5 + 5 *Math.random(),new Random().nextInt(1000000)

        );//返回字符串结果
        String result = "";//使用 Jackson 库将对象转为 JSON 字符串
        try{

            ObjectMapper objectMapper= newObjectMapper();

            result=objectMapper.writeValueAsString(movieData);

        }catch(Exception e) {

            log.error("转换 JSON 字符串出现异常:{}", e);

        }returnresult;



    }//电影数据类
@Dataprivate static classMovieData {private intrank;privateString name;privateString releaseDate;privateString country;privateString genre;private doublerating;private intplayCount;public MovieData(intrank, String name, String releaseDate

        , String country, String genre

        ,double rating, intplayCount) {this.rank =rank;this.name =name;this.releaseDate =releaseDate;this.country =country;this.genre =genre;this.rating =rating;this.playCount =playCount;

        }

    }

}

请确保替换localhost:9092和ods_movie_data为你实际使用的Kafka服务器地址和主题名称。这个简单的Java应用程序会生成包含电影排名、电影名称、上映日期、制作国家、类型、评分、播放次数等字段的模拟电影数据，并将其发送到指定的Kafka主题。

2.3.2 数据存储

数据存储在大数据分析系统中扮演着关键的角色，它不仅需要提供高度可靠性的存储机制，还需要根据业务需求进行智能分区，以便后续的离线分析和查询。在当前场景中，我们面对的是一批不断涌入的实时流数据，这些数据经过实时处理后需要被有效地存储到Hive中，以满足后续的离线分析需求。
为了保证数据的时效性，我们计划将每隔5分钟的数据作为一个时间窗口进行存储，这不仅有助于提高查询效率，还能够更好地支持基于时间的分析。为了实现这一目标，我们将使用Apache Flink作为我们的流处理引擎，通过其与Kafka的集成，实时地消费并处理Kafka Topic中的数据。具体实现流程如图所示。

1.环境依赖

Flink在消费Kafka集群中的数据时，需要引入一系列依赖项以确保系统的顺利运行。
为了实现对Kafka集群中数据的高效消费，我们需要引入Flink相关的依赖项。这些依赖项不仅包括Flink核心库，还涉及到与Kafka连接和交互的库。具体依赖见代码所示。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-filesystem_2.12</artifactId>
    <version>${flink.connector.version}</version>
 </dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.12</artifactId>
    <version>${flink.kafka.version}</version>
 </dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.12</artifactId>
    <version>${flink.streaming.version}</version>
 </dependency>

2.读取数据

编写Flink代码来消费Kafka Topic并将数据直接存储到HDFS，无需进行额外的逻辑处理，以备后续使用MapReduce进行数据预处理。具体实现见代码所示。

@Slf4jpublic classFlinkTemplateTask {public static voidmain(String[] args) {//检查输入参数是否满足要求
        if (args.length != 3) {

            log.error("kafka(server01:9092), 
            hdfs(hdfs://cluster01/data/),
            flink(parallelism=2) must be exist.");
            return;

        }

        String bootStrapServer= args[0];

        String hdfsPath= args[1];int parallelism = Integer.parseInt(args[2]);//创建 Flink 流处理环境
        StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(5000);

        env.setParallelism(parallelism);

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//从 Kafka 中读取数据
        DataStream<String> transction =env.addSource(new FlinkKafkaConsumer010<>("ods_movie_data",newSimpleStringSchema(), configByKafkaServer(bootStrapServer)));//存储到 HDFS
        BucketingSink<String> sink = new BucketingSink<>(hdfsPath);//自定义存储到HDFS上的文件名，用小时和分钟来命名，方便后面计算策略
        sink.setBucketer(new JDateTimeBucketer<String>("HH-mm"));



        sink.setBatchSize(1024 * 1024 * 4); //大小为 5MB
        sink.setBatchRolloverInterval(1000 * 30); //时间 30s
transction.addSink(sink);//执行 Flink 任务
        env.execute("Kafka2Hdfs");

    }//设置Kafka消费者的配置
    private staticObject configByKafkaServer(String bootStrapServer) {

        Properties props= newProperties();

        props.setProperty("bootstrap.servers", bootStrapServer);

        props.setProperty("group.id", "test_bll_group");

        props.put("enable.auto.commit", "true");

        props.put("auto.commit.interval.ms", "1000");

        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");returnprops;

    }



}

在这里需要特别注意，我们将时间窗口设置得较短，每隔30秒进行一次检查。如果在该批次的时间窗口内没有数据到达，我们将生成一个文件并保存到HDFS上。
此外，我们对DateTimeBucketer进行了重写，创建了JDateTimeBucketer。这一调整的逻辑并不复杂，只是在原有的方法基础上增加了一个年-月-日/时-分的文件生成路径。举例来说，在HDFS上生成的路径可能是：xxxx/2023-10-10/00-00。这个调整有助于更好地组织和管理生成的文件，使其更符合时间和日期的结构。

3.文件命名策略

在这个步骤中，我们需要对已经存储到HDFS上的文件进行预处理。处理逻辑如下：例如，当前时间是2023-10-10 14:00，我们需要将当天的13:55、13:56、13:57、13:58、13:59这最近5分钟的数据处理到一起，并加载到Hive的最近5分钟的一个分区中。为了实现这一目标，我们需要生成一个逻辑策略集合，其中以HH-mm作为key，与之最近的5个文件作为value。这个集合将被用于数据预处理和合并。具体实现见代码所示。

public classDateRangeStrategy {public static voidmain(String[] args) {

        getFileNameStrategy();

    }//生成策略
    private static voidgetFileNameStrategy() {for (int i = 0; i < 24; i++) {for (int j = 0; j < 60; j++) {if (j % 5 == 0) {if (j < 10) {if (i < 10) {if (i == 0 && j == 0) {

                                System.out.println

                                ("0" + i + "-0" +j+ "=>23-59,23-58,23-57,23-56,23-55");

                            }else{if (j == 0) {

                                    String tmp= "";for (int k = 1; k <= 5; k++) {

                                        tmp+= "0" + (i - 1) + "-" 
                                        + (60 - k) + ",";

                                    }

                                    System.out.println

                                    ("0" + i + "-0" +j+ "=>" + tmp.substring(0, 

                                    tmp.length()- 1));

                                }else{

                                    String tmp= "";for (int k = 1; k <= 5; k++) {if (j - k < 10) {

                                            tmp+= "0" + i + "-0" 
                                            + (j - k) + ",";

                                        }else{

                                            tmp+= "0" + i + "-" 
                                            + (j - k) + ",";

                                        }

                                    }

                                    System.out.println("0" + i + "-0" +j+ "=>" + tmp.substring(0, tmp.length() - 1));

                                }

                            }

                        }else{if (j == 0) {

                                String tmp= "";for (int k = 1; k <= 5; k++) {if (i - 1 < 10) {

                                        tmp+= "0" + (i - 1) + "-" + (60 - k) + ",";

                                    }else{

                                        tmp+= (i - 1) + "-" + (60 - k) + ",";

                                    }

                                }

                                System.out.println(i+ "-0" + j + "=>" 
                                + tmp.substring(0, tmp.length() - 1));

                            }else{

                                String tmp= "";for (int k = 1; k <= 5; k++) {if (j - k < 10) {

                                        tmp+= i + "-0" + (j - k) + ",";

                                    }else{

                                        tmp+= i + "-" + (j - k) + ",";

                                    }

                                }

                                System.out.println(i+ "-0" +j+ "=>" + tmp.substring(0, tmp.length() - 1));

                            }

                        }

                    }else{if (i < 10) {

                            String tmp= "";for (int k = 1; k <= 5; k++) {if (j - k < 10) {

                                    tmp+= "0" + i + "-0" 
                                    + (j - k) + ",";

                                }else{

                                    tmp+= "0" + i + "-" 
                                    + (j - k) + ",";

                                }

                            }

                            System.out.println("0" + i + "-" 
                            + j + "=>" + tmp.substring(0, tmp.length() - 1));

                        }else{

                            String tmp= "";for (int k = 1; k <= 5; k++) {if (j - 1 < 10) {

                                    tmp+= i + "-0" + (j - k) + ",";

                                }else{

                                    tmp+= i + "-" + (j - k) + ",";

                                }

                            }

                            System.out.println(i+ "-" +j+ "=>" + tmp.substring(0, tmp.length() - 1));

                        }

                    }

                }

            }

        }

    }

}

4.数据加载

当数据准备完毕后，我们可以借助Hive的LOAD命令直接将HDFS上预处理的文件加载到相应的表中。具体实现见代码所示。

LOAD DATA INPATH '/data/hive/hfile/data/min/2023-10-10/14-05/'OVERWRITEINTO TABLEgame_user_db.ods_movie_data PARTITION(day='2023-10-10',hour='14',min='05');

在执行命令时，如果文件不存在可能导致加载出错。因此，在加载 HDFS 路径之前，我们可以先判断一下路径是否存在。具体实现见代码所示。

#!/bin/bash



# HDFS 数据路径

hdfs_path='/data/hive/hfile/data/min/2023-10-10/14-05/'# 检查 HDFS 路径是否存在if hdfs dfs -test -e "$hdfs_path"; then# 如果存在，则执行加载操作echo "执行 Hive 数据加载操作"hive-e "LOAD DATA INPATH
        '$hdfs_path'OVERWRITE INTO TABLE 

        game_user_db.ods_movie_data 

        PARTITION(day='2023-10-10',hour='14',min='05');"else
    echo "HDFS路径: ['$hdfs_path'] 不存在"
fi

这里需要注意的是，这个脚本先检查 HDFS 路径是否存在，如果存在则执行加载操作，否则输出错误信息。这样可以避免因文件不存在而导致的加载错误。

2.3.3 数据分析

数据分析是一门科学，通过对数据的整理、清洗和分析，我们能够从中挖掘出隐藏在庞大数据背后的规律和趋势。数据分析工具的不断发展使得这一过程变得更加高效。统计学、机器学习、人工智能等技术的应用，使得我们能够更深入地理解数据中的信息。通过数据可视化技术，我们能够将抽象的数据变成直观的图表和图像，更容易理解和传达数据所蕴含的含义。

1.分析电影年份柱状图

为了生成电影年份的柱状图，我们可以通过对电影年份数据进行聚合。具体实现见代码所示。

--电影年份
SELECT release_date, COUNT(1) ASpvFROMods_movie_dataWHERE day = '2023-10-10'
GROUP BY release_date;

执行上述代码，分析结果如图所示。

2.分析电影类型扇形图

为了生成电影类型的扇形图，我们可以通过对电影类型数据进行聚合。具体实现见代码所示。

--电影类型
SELECT genre, COUNT(1) ASpvFROMods_movie_dataWHERE day = '2023-10-10'
GROUP BY genre;

执行上述代码，分析结果如图所示。

3.分析电影评分散点图

为了生成电影评分的散点图，我们可以通过对电影评分数据进行聚合。具体实现见代码所示。

--电影评分
SELECT rating, COUNT(1) ASpvFROMods_movie_dataWHERE day = '2023-10-10'
GROUP BY rating;

执行上述代码，分析结果如图所示。

3.总结

本篇聚焦于构建大数据分析系统，全面介绍了该系统的架构设计，并深入探讨了各个模块的实现细节。通过循序渐进的方式，读者能够逐步了解大数据分析系统的构建过程，从而更好地理解其运作机制。
总体而言，本篇内容旨在为读者提供一个全面而深入的大数据分析系统实现指南，通过理论和实践相结合的方式，帮助读者更好地掌握大数据分析的核心概念和技术，为实际项目应用打下坚实基础。

4.结束语

这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！

另外，博主出新书了《
深入理解Hive
》、同时已出版的《
Kafka并不难学
》和《
Hadoop大数据挖掘从入门到进阶实战
》也可以和新书配套使用，喜欢的朋友或同学， 可以
在公告栏那里点击购买链接购买博主的书
进行学习，在此感谢大家的支持。关注下面公众号，根据提示，可免费获取书籍的教学视频。

  1. 1
  2. ...
  3. 11
  4. 12
  5. 13
  6. 14
  7. 15
  8. 16
  9. 17
  10. ...
  11. 108