《Flink应用开发技术》课件-项目4：Dataset API

FlinkDatasetAPI实战指南1目录CONTENTS01.流批一体与DatasetAPI02.应用场景与准备03.MySQL读取实战04.总结与注意事项201流批一体与DatasetAPI3Flink统一编程模型概览ApacheFlink通过统一引擎同时支持流处理与批处理，DatasetAPI是其面向有界数据集的批处理核心接口。流批一体设计思想以同一套抽象完成离线计算任务，实现真正的流批统一，为开发者提供一致的编程体验。DatasetAPI定位作为Flink架构中处理有界数据集的核心接口，专为高性能批处理而设计。4DatasetAPI核心优势解析高性能内存计算配合磁盘持久化，提供卓越的批处理速度。易开发类SQL操作与丰富算子（map,filter,groupBy），降低开发门槛。可扩展自动分区与容错机制，保障作业在大规模集群上的稳定运行。同时与DataStream,Hadoop,Hive,Kafka等生态无缝打通。502应用场景与准备6典型业务场景速览技术优势与业务需求快速对应数据仓库与ETL高效数据清洗与转换机器学习与数据挖掘大规模数据集特征工程数据分析与报表生成快速生成业务报表日志处理与实时监控离线日志分析与统计金融风控与电商推荐离线模型训练与评估7Maven依赖与环境配置使用Maven管理项目，确保示例代码可直接编译运行。核心依赖:添加flink-scala_2.12

等坐标。版本匹配:注意依赖间及与集群的版本兼容性。环境检查:确保本地JDK、Scala环境及远程MySQL连接权限配置正确。//pom.xml<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-scala_2.12</artifactId>

<version>1.14.0</version>

</dependency>803MySQL读取实战9实战任务：从MySQL读取数据任务目标通过DatasetAPI，将MySQLtest01库中Person表的全量数据读入Flink作业，并打印到控制台。前置条件MySQL服务已启动并允许远程连接已知Person表结构JDBC驱动已正确配置10代码实现五步法1引入依赖2构建输入格式3配置连接信息4应用环境5定义输出并执行11关键代码片段核心在于使用JdbcInputFormat构建数据输入源。链式配置:通过buildJdbcInputFormat设置参数。关键参数:包括驱动名、URL、账号、密码和SQL。类型信息:使用RowTypeInfo指定返回数据类型。//关键代码片段valinputFormat=JdbcInputFormat.buildJdbcInputFormat().setDrivername("com.mysql.jdbc.Driver").setDBUrl("jdbc:mysql://ip:port/test01").setUsername("your_username").setPassword("your_password").setQuery("SELECTid,nameFROMPerson").setRowTypeInfo(newRowTypeInfo(BasicTypeInfo.INT_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO)).finish()12运行结果与验证控制台预期输出如下，表明数据已成功读取并处理。//预期输出>1,张三>2,李四>3,王五>4,张飞验证要点结果为空:首先检查SQL语句、网络连通性及表名大小写。异常堆栈:仔细阅读日志，定位是连接问题、权限问题还是代码逻辑问题。养成习惯:先验证结果正确性，再进行性能调优。1304总结与注意事项14DatasetAPI实战收获简洁高效通过不到五十行代码即可完成MySQL批量读取，证明了DatasetAPI在外部系统集成上的简洁性与高效性。选型依据适用于一次性离线处理场景，对比传统Excel或脚本方式，在速度与可扩展性上具备显著优势。15开发运维注意清单环境配置确保文件路径、编码、数据库连接权限与驱动版本完全匹配。性能调优大数据量需合理设置并行度与网络缓冲区，避免OOM。资源管理作业结束后务必关闭连接，防止资源泄露。监控告警增加日志与监控，便于快速定位问题。16DatasetAPI数据连接17目录CONTENTS01.连接类型全景速览02.连接类型速查对比03.环境搭建与数据模型04.连接操作编程实战05.运行验证与结果检查06.关键技术要点回顾1801连接类型全景速览19Union合并操作保留全部元素，不去重Union操作的核心特性是将多个同结构的数据集进行合并，它类似于SQL中的UNIONALL，会将所有数据简单地堆叠在一起，不会对元素进行去重，合并后元素的顺序也可能发生变化。多源合并历史追加20InnerJoin(内连接)仅保留两个数据集都存在的匹配记录数据集A数据集B交集结果dataSetA.join(dataSetB).where("key").equalTo("key")21OuterJoin(外连接):保左保右保全部左外连接保留左数据集全部记录，右数据集无匹配时用null填充。右外连接保留右数据集全部记录，左数据集无匹配时用null填充。全外连接保留左右数据集所有记录，任意方向无匹配时用null填充。空值处理技巧:使用Option包装并使用getOrElse提供默认值，确保类型安全。22Cross笛卡尔积全组合爆炸笛卡尔积是无条件的全排列组合，结果集大小为两个数据集大小的乘积，可能生成极大的结果集，需谨慎使用。适用场景:全组合分析距离/相似度计算生成测试数据数据集AN×数据集BM结果集大小N×M2302连接类型速查对比24五类连接速查表操作类型语法结果集大小匹配条件主要用途Cross.cross()size1×size2无条件全组合分析外连接.leftOuterJoin()取决于匹配字段相等数据完整性保障内连接.join().where().equalTo()仅匹配记录字段相等精确关联Union.union()size1+size2无条件数据合并2503环境搭建与数据模型26执行环境三步初始化1创建环境创建批处理执行环境，准备运行基础。2导入依赖导入必要依赖包，确保API可用。3准备数据准备示例数据集，定义数据模型。27三数据集模型设计人员基本信息(编号:Int,姓名:String)人员年龄信息(编号:Int,年龄:Int)空间坐标点(编号:Int,x:Double,y:Double)通过“编号”字段关联人员信息，使用“坐标”数据进行距离计算。2804连接操作编程实战29左外连接代码实现确保左表数据完整性，右表无匹配则补nullvalleftJoinResult=firstDataStream.leftOuterJoin(secondDataStream).where(0).equalTo(0)//按第一个字段(编号)关联.apply{(left,right)=>valid=left._1valname=left._2valage=right.map(_._2).getOrElse("null")//处理空值(id,name,age)}核心要点：where(0).equalTo(0)确保关联字段对齐，getOrElse提供空值安全处理。30右外连接代码实现确保右表数据完整性，左表无匹配则补nullvalrightJoinResult=firstDataStream.rightOuterJoin(secondDataStream).where(0).equalTo(0)//关联逻辑相同.apply{(left,right)=>valid=right._1valname=left.map(_._2).getOrElse("null")//左表可能为空valage=right._2(id,name,age)}与左外连接逻辑镜像，核心是以右表为驱动，关注左表的空值情况。31Cross笛卡尔积与距离计算实现坐标点两两之间的全组合与距离计算valcrossResult=coordStream1.cross(coordStream2).apply{(point1,point2)=>valid1=point1._1valid2=point2._1valdist=math.sqrt(math.pow(point1._2-point2._2,2)+math.pow(point1._3-point2._3,2))(id1,id2,dist)}距离公式(d=\sqrt{(x_1-x_2)^2+(y_1-y_2)^2})3205运行验证与结果检查33程序执行四步流程1.创建类文件2.导入依赖3.实现逻辑4.运行验证34三类连接结果验证清单左外连接确保左表数据完整，所有记录都出现在结果中。右外连接确保右表数据完整，所有记录都出现在结果中。笛卡尔积验证组合数量正确性(NxM)，并手算复核距离公式准确性。3506关键技术要点回顾36关键技术要点：字段匹配与空值安全字段匹配确保连接字段的类型和名称完全一致，这是关联成功的基础。建议在关联前打印schema进行检查。空值安全外连接中，无匹配的记录会用null表示。使用Option包装并使用getOrElse提供默认值，是避免NullPointerException的最佳实践。37性能优化三板斧广播优化使用broadcast将小表广播到所有节点，减少shuffle，提升Join性能。高效类型选择合适的数据类型（如IntvsString），能显著提升处理效率和减少内存占用。先减数据对于Cross等操作，务必先进行数据过滤，减少参与计算的数据量，避免N²爆炸。3807场景选型与避坑指南39四类场景对号入座精确清洗用InnerJoin进行精确匹配，过滤无效数据。缺失发现用OuterJoin分析数据完整性，发现缺失值。组合爆破用Cross进行全组合分析或距离计算。多源堆叠用Union合并多个同结构数据源。40常见错误与预防策略类型不一致:连接字段类型不匹配导致关联失败。预防：提前打印schema进行核对。空值未处理:外连接结果中的null导致运行时错误。预防：统一使用Option进行安全处理。结果未验证:盲目信任代码逻辑，未验证结果正确性。预防：编写断言或进行抽样检查。Cross未评估数据量:导致生成爆炸级结果集。预防：先采样估算N²，并设置资源限制。41Dataset输出操作全解析42目录CONTENTS01.知识准备02.任务实现4301知识准备44Dataset输出概述数据输出是流批处理链路的最终环节，Flink通过OutputFormat接口，将计算结果统一导向多样化的外部系统。文件系统CSV,Parquet,Avro等格式落盘数据库MySQL,PostgreSQL等关系型数据库消息队列Kafka,RabbitMQ等消息中间件45OutputFormat接口详解通过实现此接口，可精确控制数据输出的生命周期，封装连接池、事务、重试、监控等复杂逻辑，满足个性化需求。open()打开输出连接，进行初始化，如创建客户端、建表等。writeRecord()核心方法，将Dataset中的每一条记录序列化并写入外部系统。close()关闭输出连接，释放资源，确保数据完整性。46Kafka输出示例自定义KafkaOutputFormat，将数据流可靠地发送至指定Topic，需重点关注生产者配置对吞吐与可靠性的影响。acks:副本确认机制retries:消息发送失败重试batch.size:批量发送消息大小classKafkaOutputFormat[...]{//open():初始化KafkaProducer//writeRecord():序列化并producer.send()//close():producer.flush()并关闭}47输出实现原理：从Dataset到外部系统Dataset数据划分与分发通过Partitioner将数据分片，由TaskManager并行执行输出，实现负载均衡。序列化过程将记录序列化为字节流。Flink优先使用高效的Kryo，也支持Java序列化。写入外部系统4802任务实现49任务目标与环境准备实战任务将包含学号、课程号、成绩三字段的学生成绩数据，从FlinkDataset批量写入MySQL数据库。开发环境清单IntelliJIDEAFlink1.x集群环境MySQL5.7+及JDBC驱动前置检查：确保网络连通、目标库表已创建、数据库账号已授权。50数据模型设计学生成绩数据以三元组形式存在，在Flink中通过Row对象灵活承载。1学号(sno)String类型，如"2021001"2课程号(cno)String类型，如"CS101"3成绩(grade)Integer类型，如9551JdbcOutputFormat五步配置法遵循标准五步模板，确保数据库连接与SQL语句正确配置，注意链式调用以`.finish()`结束。1.构建器buildJdbcOutputFormat()2.驱动与地址setDrivername/setDBUrl3.账号密码setUsername/setPassword4.SQL语句setQuery("insert...")5.完成构建.finish()52完整代码实现核心代码简洁，通过链式调用一步完成配置。注意SQL占位符顺序与类型必须与Row字段严格对齐。data.output(JdbcOutputFormat.buildJdbcOutputFormat().setDrivername("com.mysql.jdbc.Driver").setDBUrl("jdbc:mysql://...:3306/test01?useSSL=false").setUsername("root").setPassword("Password123$").setQuery("insertintostudent(sno,cno,grade)values(?,?,?)").finish()//必须调用以完成构建)提示：可通过`.setBatchSize(int)`控制批量提交，优化写入性能。53运行验证三步走1提交作业打包并提交Flink作业，观察控制台日志。2观察日志查找“MySQL写入成功”及退出码0，确认作业成功完成。3数据库核查登录数据库执行`selectcount(*)fromstudent`确认数据行数。54技术要点正确选用`JdbcOutputFormat`严谨配置数据库连接参数保持SQL占位符与数据类型匹配注意事项代码需格式化，保证可读性合理设置批量提交大小务必考虑异常处理与资源关闭55DatasetAPI迭代计算56目录CONTENTS01.迭代计算速览02.增量迭代拆解03.全量迭代剖析04.圆周率实战演练05.调优与陷阱速查5701迭代计算速览58迭代计算：大数据价值提炼的核心引擎重复执行计算直至收敛，是图计算、机器学习、数值模拟等领域的基石。不会迭代，大数据的价值将大打折扣。图计算PageRank、最短路径机器学习逻辑回归、K-Means聚类数值计算圆周率估算、方程求解增量迭代(IterateDelta)全量迭代(Iterate)59增量迭代：以局部更新撬动全局收敛核心思想：只处理“变化”的数据，大幅降低资源消耗。初始解集保存当前世界状态工作集携带待刷新的消息收敛结果数据渐少，结果渐稳核心API骨架valresult=initialSolutionSet.iterateDelta(initialWorkset,maxIterations,key){(solution,work)=>//1.Join工作集与解集//2.计算局部更新//3.产生新解集与新工作集}6002增量迭代拆解61增量迭代核心：解集与工作集的分工协作初始解集(SolutionSet)保存当前世界状态，是迭代结果的累积。存储已计算出的节点PageRank值保存机器学习模型的当前参数工作集(Workset)携带待刷新的消息，驱动下一轮计算。存储需要更新PageRank的节点ID携带模型参数的梯度信息两者通过Join操作（键值需匹配）完成局部更新，新解集自动成为下一轮输入，工作集可动态收缩。62增量迭代的两种终止策略1.固定次数适合已知误差容忍度的算法，简单直接。.iterateDelta(workset,100,key)风险：可能过早停止或过度计算。2.自定义收敛函数实时比较结果差异，动态决定停止。.iterateDelta(workset,MyConvergenceFunction,key)注意：需搭配最大迭代上限防止无限循环。6303全量迭代剖析64全量迭代：全局视角，极致简洁使用iterate()API，每轮拉取并更新全量数据集，逻辑直观，精度可控。核心特点逻辑简单：无需管理解集与工作集，适合K-Means等需全局聚合的场景。精度保证：每轮都是完整计算，无增量误差累积。资源代价高：网络、内存、CPU随数据规模线性上升。对比维度增量迭代(IterateDelta)全量迭代(Iterate)数据处理方式只处理部分数据处理全部数据资源消耗低，适合大规模数据高，适合中小规模数据实现复杂度高，需管理解集与工作集低，逻辑直观适用场景图计算、大规模机器学习K-Means、精度要求极高的批量计算增量vs.全量：如何选择？6504圆周率实战演练66实战：用蒙特卡洛方法估算圆周率核心思想：随机投点+几何概率+大数定律。建系：以原点为中心作单位圆（面积(\pi)）和外切正方形（面积(4)）。投点：在正方形内随机生成大量点，统计落在圆内的比例。估算：根据公式(\pi\approx4\times\frac{\text{圆内点数}}{\text{总点数}})计算圆周率。Flink优势：分布式并行生成随机点，高效完成海量统计。利用RichMapFunction绑定独立随机种子，保证并发安全与结果可复现67代码实现：百万次迭代汇聚精准π//创建执行环境

valenv=ExecutionEnvironment.getExecutionEnvironment

//创建初始数据集