《Flink应用开发技术》课件-项目5：Table API与Flink SQL

TableAPI&SQL实战1目录CONTENTS01.统一关系型API02.动态表与连续查询03.开发环境配置04.表定义与外部连接05.API风格与优化06.Kafka实时分析实战201统一关系型API3TableAPI&SQL统一流批查询语义以同一套语句处理有界批数据与无界流数据，实现流批一体，降低开发与学习成本。实时数据分析持续更新的动态报表流式ETL处理数据清洗与转换复杂事件处理(CEP)识别事件序列模式，如用户行为路径、设备异常检测等。核心目标：掌握编程模型、理解流表对偶性、构建Kafka实时分析应用。4Catalog：元数据的“中央枢纽”统一管理元数据，实现一次注册，到处引用GenericInMemoryCatalog内存临时存储，会话结束即消失。JdbcCatalog/HiveCatalog直连外部系统，复用现有元数据。自定义Catalog扩展性支持，对接专有系统。通过catalog.db.table实现跨源联合查询，通过USECATALOG切换会话上下文。502动态表与连续查询6动态表：用关系视图拥抱无限流将不断涌入的事件流，视为一张持续更新的关系表事件流无限、持续、追加流表对偶动态表有界、可更新、关系型案例：5分钟滚动聚合对订单流进行连续查询，Flink将窗口逻辑隐入背后，开发者只需写标准SQL，即可得到持续更新的结果表。7连续查询：增量计算保障低延迟基于变更日志流增量计算，而非全表重新扫描+I(插入)-D(删除)-U/+U(更新)变更事件沿查询计划树传播，仅重算受影响结果。需注意状态管理：聚合节点需保存累加值，JOIN节点需缓存数据。合理设置stateTTL并开启mini-batch调优是关键。803开发环境配置9环境与Planner选型正确的依赖与Planner是项目成功的基础核心依赖flink-table-api-scala-bridge_2.12:连接TableAPI与Scala运行时。flink-table-planner_2.12:将SQL转化为可执行算子。flink-table-common:提供通用功能与接口。注意：Scala2.12与Flink版本需严格对齐。Planner选择BlinkPlanner(推荐)流批一体，优化规则丰富，是未来的默认选择。LegacyPlanner(已停止维护)仅用于兼容旧项目，不建议在新项目中使用。10TableEnvironment：程序的入口与总控承载解析SQL、注册表、提交执行的全生命周期独立模式适合纯SQL作业，通过`EnvironmentSettings`配置。valsettings=...valtableEnv=TableEnvironment.create(settings)依附模式适合混合DataStream与TableAPI，利用现有流环境。valenv=StreamExecutionEnvironment.get...valtableEnv=StreamTableEnvironment.create(env)1104表定义与外部连接12三元标识符精准定位任意表消除同名歧义，实现跨源联合查询Catalog数据源.Database数据库.Table表名转义与默认设置使用反引号(``)转义关键字。通过`USECATALOG`和`USEDATABASE`设置当前默认值。临时表覆盖临时表优先于永久表，方便实验。生产环境应锁定写权限，防止误覆盖。13文件系统表：让本地CSV秒变数据源无需Java代码，通过DDL直接映射文件为表CREATETABLEUserTable(idINT,nameSTRING,ageINT)WITH('connector'='filesystem','path'='data/sample.csv','format'='csv')路径与格式支持通配符(如`data/*.csv`)和多种format(json,parquet)。流式监听如需流式监听，需使用`file-system`streamingsource并开启分区发现。1405API风格与优化15链式TableAPI：类型安全与调试友好相比纯SQL，提供更强大的IDE支持和调试能力valresult=tableEnv.from("Orders").filter($"cCountry"==="FRANCE").groupBy($"cID",$"cName").select($"cID",$"cName",$"revenue".sum().as("revSum"))类型安全`$"字段"`在编译期检查，减少运行时错误。调试友好可打断点逐步验证，支持`table.explain()`查看执行计划。可视化优化IDE插件可高亮Planner优化规则，让优化过程透明化。16Calcite优化规则：自动裁剪无效算子基于成本模型，将逻辑计划改写为更高效的物理计划投影剪裁去掉未引用的列，减少数据传输。过滤器下推将WHERE条件下沉到Source，提前过滤。分区剪裁只读取匹配目录，降低IO。子查询重写将IN转化为JOIN，利用哈希加速。1706Kafka实时分析实战18Kafka源表DDL：一键接入事件流通过DDL直接将KafkaTopic映射为动态表CREATETABLEuser_events(user_idSTRING,event_typeSTRING,event_timeSTRING,product_idSTRING)WITH('connector'='kafka','topic'='user-events-topic','properties.bootstrap.servers'='master:9092','format'='json')Connector&Topic指定为'kafka'，topic支持单主题或正则。BootstrapServersKafka集群连接地址。Format&Startup指定消息格式，并可配置消费起始位点。19多维度聚合SQL：实时找出高活跃用户标准SQL即可完成复杂的实时分析逻辑SELECTuser_id,COUNT(DISTINCTproduct_id)ASdistinct_products,COUNT(*)AStotal_eventsFROMuser_eventsWHEREevent_type='click'GROUPBYuser_idHAVINGCOUNT(DISTINCTproduct_id)>1ORDERBYtotal_eventsDESCLIMIT10业务逻辑过滤点击事件，按用户分组，统计去重商品数和总事件数，筛选出浏览多个商品且最活跃的前10用户。执行优化`ORDERBYLIMIT`会被优化为高效的`Top-N`算子，使用最小堆算法增量维护结果，内存占用可控。20结果流三种编码：灵活对接下游系统根据下游系统能力，选择最合适的输出模式toChangelogStream保留完整变更日志(+I,-U,+U,-D)，适合写入Kafka供其他Flink作业消费。toRetractStream将更新拆分为先删(-)后插(+)，兼容旧版DataStreamAPI。toDataStream仅支持追加模式，适合写入ES、ClickHouse等只支持insert的系统。21TableAPI&SQL交通数据分析22目录CONTENTS01.时间概念与语义02.处理时间与事件时间03.SQL与窗口TVF04.聚合与多维分析05.交通场景任务设计23目录CONTENTS01.核心查询实现02.调优与扩展思考2401时间概念与语义25三种时间语义决定数据正确性流计算的时间基石：采集位置与确定性差异ProcessingTime基于机器本地时间，简单但不提供确定性，适合延迟敏感但精度要求低的场景。EventTime数据本身携带的时间戳，保证处理结果的一致性，支持乱序数据处理，需配合水印。IngestionTime数据进入Flink的时间，介于二者之间，提供可预测的一致性。26逻辑时间属性的生命周期声明(Schema/DDL)在Schema或DDL中标记时间字段传递(DML)在转换中自动识别和传递使用(Operators)被窗口、Top-N、Join等算子识别延伸(Downstream)随计算链自动向下游延伸时间属性一旦定义，即可被各类算子识别并向下游传递。若未显式声明，窗口TVF将无法解析。2702处理时间与事件时间28处理时间：快速落地，零配置无需提取时间戳或生成水印，适合对开发效率要求高、但精度要求低的场景。三种定义入口Schema.newBuilder()、DDL声明、DataStream→Table转换。即写即跑代码示例展示Socket流与KafkaDDL的简洁定义。例5.8:Socket数据流//在Schema中定义处理时间Schema.newBuilder().columnByExpression("proc_time","PROCTIME()")例5.9:Kafka连接器DDL--在DDL中定义处理时间CREATETABLEkafka_table(...proc_timeASPROCTIME())29事件时间+水印=结果一致性用数据自带时间戳抵消分布式乱序，配合水印声明“已到达”边界，保证实时与离线结果完全一致。数据携带时间戳数据本身携带业务时间+水印机制标识事件时间进展，设置延迟容忍=结果可重放支持乱序，保证多次运行结果一致例5.10:事件时间窗口统计--在DDL中定义事件时间和水印CREATETABLEVehiclePasses(vehicleIdSTRING,speedDOUBLE,eventTimeTIMESTAMP(3),WATERMARKFOReventTimeASeventTime-INTERVAL'30'SECOND)3003SQL与窗口TVF31FlinkSQL的三类核心语句DDL(数据定义)建表、建视图。必须声明时间属性与连接器参数。DML(数据操纵)插入、更新数据。通过executeSql()执行。查询(Query)查询数据。通过sqlQuery()执行，返回Table对象。核心原则：先建语义表(DDL)，再写业务SQL(DML/Query)。32窗口TVF:遵循SQL标准基于SQL2016标准，直接在FROM子句调用，避免早期GROUP语法的精度歧义，返回带窗口起止时间的多列表。TUMBLE(滚动)固定长度，不重叠。适合固定周期统计。HOP(滑动)滑动步长，可能重叠。适合监控预警。CUMULATE(累积)逐分钟累积，适合仪表盘指标。SESSION(会话)按间隔切分，适合用户行为分析。33三类核心窗口实战用法TUMBLE:固定周期将数据切分为不重叠的固定长度窗口，如统计每5分钟车流量。TUMBLE(eventTime,INTERVAL'5'MINUTE)HOP:滑动监控按滑动步长统计最近固定时长数据，如每1分钟更新最近5分钟平均速度。HOP(eventTime,INTERVAL'1'MINUTE,INTERVAL'5'MINUTE)CUMULATE:累积补全从窗口开始时间逐分钟累积，适合仪表盘指标补全。CUMULATE(eventTime,INTERVAL'1'MINUTE,INTERVAL'5'MINUTE)3404聚合与多维分析35窗口聚合与多维分析状态生命周期管理窗口聚合的状态默认在窗口触发后即被清理，防止无限增长。可通过配置调整保留时间以应对迟到数据。高级分组:GROUPINGSETS通过GROUPINGSETS,ROLLUP,CUBE实现多维漏斗分析，一次查询获得多个维度的汇总结果。GROUPINGSETS示例GROUPBYGROUPINGSETS((vehicleId,lane),(vehicleId),())ROLLUP简化表示GROUPBYROLLUP(vehicleId,lane)3605交通场景任务设计37业务目标到技术实现映射业务需求5分钟车辆通行统计Top-N超速排名驾驶行为违规关联→Flink技术窗口TVF(TUMBLE)ROW_NUMBEROVER双流Join(IntervalJoin)→输出指标平均速度,通过次数速度排名(Top3)超速/疲劳次数38环境与源表初始化1.流处理与表环境配置valenv=StreamExecutionEnvironment.getExecutionEnvironmentenv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)env.setParallelism(1)valsettings=EnvironmentSettings.newInstance().inStreamingMode().build()valtableEnv=StreamTableEnvironment.create(env,settings)2.注册Kafka源表tableEnv.executeSql("""CREATETABLEVehiclePasses(vehicleIdSTRING,speedDOUBLE,laneINT,eventTimeTIMESTAMP(3),WATERMARKFOReventTimeASeventTime-INTERVAL'30'SECOND)WITH('connector'='kafka',...)""")关键：设置EventTime特征，并在源表中声明水印，为后续窗口操作奠定基础。3906核心查询实现40核心功能1:五分钟滚动窗口统计使用TUMBLE窗口函数，按车辆ID分组，统计每5分钟内的平均速度和通行次数。SELECTvehicleId,TUMBLE_START(eventTime,INTERVAL'5'MINUTE)ASwindow_start,TUMBLE_END(eventTime,INTERVAL'5'MINUTE)ASwindow_end,AVG(speed)ASavg_speed,COUNT(*)ASpass_countFROMVehiclePassesGROUPBYTUMBLE(eventTime,INTERVAL'5'MINUTE),vehicleId窗口切分按车分组聚合计算41核心功能2:Top-N超速排名在窗口统计结果上，使用ROW_NUMBER()窗口函数，按平均速度降序排列，筛选出每个窗口内最快的前三名。关键点PARTITIONBY确保排名只在本窗口内有效。ORDERBYavg_speedDESC按速度降序。speed_rank<=3筛选Top-N。SELECT*,ROW_NUMBER()OVER(PARTITIONBYwindow_startORDERBYavg_speedDESC)ASspeed_rankFROMWindowedPassesWHEREspeed_rank<=342核心功能3:驾驶行为违规统计对`DriverBehaviors`表使用条件聚合(CASEWHEN)，统计每辆车的各类违规行为次数。SELECTvehicleId,COUNT(CASEWHENbehaviorType='overspeed'THEN1END)ASoverspeed_count,COUNT(CASEWHENbehaviorType='fatigue_driving'THEN1END)ASfatigue_countFROMDriverBehaviorsGROUPBYvehicleIdoverspeed_count统计每辆车的超速次数fatigue_count统计每辆车的疲劳驾驶次数43核心功能4:双流Join整合宽表将Top-N结果(`FilteredViolations`)与行为分析结果(`BehaviorAnalysis`)按`vehicleId`进行等值Join，整合为一张包含所有指标的宽表。SELECTv.window_start,v.window_end,v.vehicleId,v.avg_speed,v.