一、Flink SQL 的宏观处理流程无论批还是流一条 Flink SQL 作业的生命周期都遵循同一套编译与执行框架1、SQL 解析Parse将文本 SQL 转换成抽象语法树AST验证关键字、语法的正确性。2、语义验证Validate根据 Catalog 中已注册的表、字段、类型进行校验检查函数是否存在、类型是否匹配等。3、逻辑计划生成与优化Logical Plan RBO/CBO生成最初的逻辑计划如 LogicalTableScan、LogicalJoin、LogicalAggregate 等节点。基于 Calcite 的规则进行优化谓词下推、列裁剪、Join 重排序、子查询去关联等。4、物理计划生成Physical Plan逻辑节点转换为 Flink 的物理节点如 StreamExecCalc、StreamExecJoin、StreamExecGroupAggregate。根据是否为 bounded source 决定选择Batch 模式还是Streaming 模式的执行策略。5、翻译为 DataStream / DataSet APITransformation Graph生成底层的 Transformation 图最终映射为 JobGraph。6、提交执行与容错在集群中调度 Task根据配置的 checkpoint、状态后端保证容错。下面我结合你的两个脚本拆成具体开发步骤并解释其作用。二、批处理脚本分析多表 Join ETL1. 创建源表Source Table与目标表Sink TableCREATE TABLE ... WITH ( connector doris, fenodes ..., table.identifier zoestd_hds.hds_rate_type, username root, password ..., source.use-flight-sql false, sink.buffer-flush.max-rows 10000, ... );步骤作用通过 connector doris 向 Catalog 注册一张虚拟表映射外部存储。源表Source在批处理时Flink 会调用 Doris 的Bounded Source实现通常是批量 Scan 接口一次性拉取全表数据或按分区下推。目标表Sink则用于接收计算结果在批处理中常采用批量导入模式如 Stream Loadsink.enable.batch-mode true 正是告诉 Doris Connector 在 Checkpoint 完成后一次性提交批量数据。原理要点Flink 通过 ScanTableSource / LookupTableSource 接口动态发现数据源的边界性。Doris Source 若为有界流Flink 自动选用批执行模式。连接器参数全部在 WITH 子句中被序列化为 Properties 传递给 DynamicTableSourceFactory / DynamicTableSinkFactory 创建具体实例。2. 编写 SELECT 逻辑实现多表关联insert into ... select s.inp_cost_detail_no as charge_detail_no, ... from hdc_patient_admission a inner join hdc_inp_cost_detail s on a.inp_no s.inp_no and a.inst_code s.inst_code left join hds_rate_type rtd ... left join hds_basic bd1 ... (多次) left join hds_dept dd1 ... left join hds_staff_basic sd1 ... where a.discharge_time is not null;步骤作用核心业务住院费用明细宽表加工。将费用明细与住院记录、字典表、科室、人员等维度关联补齐各类名称信息。inner join 确保费用必须对应到一条住院记录left join 用于补充属性允许维度缺失。WHERE 过滤未出院的患者只处理有出院时间的记录。原理要点批处理 Join 策略由于所有表都是 bounded优化器会估算成本后选择Hash Join小表构建哈希表大表探测或Sort Merge Join。hdc_inp_cost_detail 是大表hdc_patient_admission 中等字典表 hds_basic、hds_dept 通常很小Flink 会尽量将小表广播Broadcast Join避免 Shuffle。谓词下推discharge_time is not null 会下推到 hdc_patient_admission 的 Scan 节点减少读取量。3. INSERT INTO 写入结果表步骤作用将最终投影的字段列表按顺序映射到目标表列。批处理作业是“一次性”执行所有数据计算完毕后整体提交。Doris Sink 在批模式下会在最后一个 Checkpoint 时将缓冲区数据一次性通过 Stream Load 提交并更新元数据。原理要点如果目标表已有数据INSERT INTO 默认追加可通过表属性或 Doris 本身的模型控制去重/覆盖。因为没有设置 sink.semantic exactly-once 等但两阶段提交在批模式下通过检查点保证一致性。三、流处理脚本分析Kafka → Doris 实时聚合1. 作业配置SET pipeline.name 门诊就诊医生人次实时;作用设置 Flink 作业名称会显示在 Flink UI 上便于运维监控。这是 Flink SQL 客户端环境变量最终会注入到 StreamExecutionEnvironment 的配置中。2. 创建 Kafka Source无界流CREATE TABLE KAFKA_PAT_OUTP_PATIENT_CLINIC_INFO ( EVENT_NO STRING, PATIENT_ID STRING, CLINIC_DEPT_CODE STRING, CLINIC_DOCTOR_CODE STRING, REGISTER_TIME STRING -- 注意这里定义为 STRING后面用 TO_TIMESTAMP 转换 ) WITH ( connector kafka, topic PAT_OUTP_PATIENT_CLINIC_INFO, properties.bootstrap.servers ..., format json, scan.startup.mode group-offsets, ... );步骤作用定义一个无界流表Flink 会启动 Kafka Consumer 持续拉取消息反序列化为 Row。scan.startup.mode group-offsets 表示从保存的消费者组 offset 开始保证升级重启不丢不重。这个表是流式作业的唯一输入。原理要点该表是Append-only流只有 INSERT 消息适合作为 Source。没有定义 event_time 和 watermark所以聚合操作默认基于处理时间或者当调用 TO_TIMESTAMP 时并没有声明为 rowtime 属性Flink 仍然把它当普通标量函数处理。这意味着后续聚合不依赖事件时间推进数据到达即计算结果持续更新。3. 创建 Doris Sink 表支持 Upsert 输出CREATE TABLE zoedw_ads_ads_oe_vst_doc_outp_emer_rt ( dim_date DATE, dim_dept STRING, name_employe_no STRING, outp_cn DECIMAL(20, 2) ) WITH ( connector doris, sink.properties.partial_update true, sink.enable.batch-mode true, -- 这里仍开启批模式意味着 Sink 会攒批写入 ... );作用结果表接收实时聚合的更新流。由于聚合会产生ChangelogI 插入、-U 更新前、U 更新后Doris Sink 需要支持这些模式。通过 partial_update 允许仅更新部分字段配合 Doris 的 Unique 模型batch-mode 则使 Sink 端攒一批数据后批量导入提高吞吐。原理要点Flink SQL 的聚合算子输出的是upsert流如果有 GROUP BY携带主键dim_date, dim_dept, name_employe_no。Doris Sink 据此产生 Upsert 操作更新明细级别的聚合值。如果不支持 Upsert可以配置 sink.properties.format 结合 Doris 的聚合模型如 replace_if_not_null实现。4. 实时分组聚合SELECT CAST(TO_TIMESTAMP(REGISTER_TIME) AS DATE) AS dim_date, CLINIC_DEPT_CODE AS dim_dept, CLINIC_DOCTOR_CODE AS name_employe_no, SUM(1) AS outp_cn FROM KAFKA_PAT_OUTP_PATIENT_CLINIC_INFO WHERE REGISTER_TIME IS NOT NULL AND TO_TIMESTAMP(REGISTER_TIME) IS NOT NULL AND CLINIC_DEPT_CODE IS NOT NULL AND CLINIC_DOCTOR_CODE IS NOT NULL GROUP BY CAST(TO_TIMESTAMP(REGISTER_TIME) AS DATE), CLINIC_DEPT_CODE, CLINIC_DOCTOR_CODE;步骤作用实时统计每个日期,科室,医生的就诊人次。分组键决定了状态粒度每来一条就诊记录就在对应分组上执行 SUM 聚合并输出更新结果。SUM(1) 本质是计数此处相当于 COUNT(*)但由于输出类型为 DECIMAL(20,2)保持与目标表一致。原理要点流式聚合的核心状态管理Flink 会为每一个日期,科室,医生组合维护一个累加器状态ValueState。当新记录到达状态值1并发送撤回旧值、输出新值的 Changelog。无窗口聚合没有时间窗口的限制状态会无限增长。生产环境通常需要配置状态TTL例如 table.exec.state.ttl否则历史日期数据永不过期。数据倾斜如果某个大科室医生就诊量极高可能导致热点分组。可通过 MiniBatch 优化或开启两阶段聚合 table.optimizer.distinct-agg.split.enabled 部分缓解。类型转换TO_TIMESTAMP(REGISTER_TIME) 在流中为每条记录调用标量函数不涉及时间语义仅做数据格式化。产出 Upsert 流最终写入 Doris 的 ads_oe_vst_doc_outp_emer_rt 表实时反映门诊人次变化BI 端可以毫秒级刷新。四、批与流的本质区别在 Flink SQL 中的体现维度批处理流处理数据源有界Bounded如 Doris 批量扫描无界Unbounded如 Kafka 持续消费执行模式execution.runtime-mode batch隐式推断默认 streaming可显式指定Join全量读取后做 Hash/Sort-Merge Join可广播小表需在 Join 两侧维持状态通常需要时间窗口或 temporal table join聚合一次性扫描全量数据排序或哈希后直接计算输出最终结果增量维护状态每来一条记录更新并输出 Changelog需要 TTL 管理输出全量写入通常是 Insert/OverwriteUpsert 流持续写入要求 Sink 支持幂等/事务时间属性可使用事件时间但非必须处理时间也能得到全量正确结果基于事件时间的聚合需要 Watermark 来推进处理时间直接系统时间在你的两个脚本中批处理本质是“一次性地将 Doris 数据抽出Join 成宽表再写回 Doris”是一个典型的 T1 离线数仓加工。流处理是“持续不断地计算实时门诊人次”结果实时更新到 Doris支撑实时大屏或指标看板。五、从原理层面掌握 Flink SQL 的开发要点1、表是核心抽象无论是批还是流统一用 CREATE TABLE 定义外部数据源SQL 逻辑完全一致批流统一差异仅在于 Source 的有界性和 Sink 对 Changelog 的支持。2、动态表Dynamic Table与 Changelog 模式流处理中表是随时间变化的。Flink SQL 内部用 RowData RowKindINSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE表示更新日志。聚合、连接会改变 Changelog 类型优化器会自动推导并选择合适算子如 retract 流或 upsert 流。3、状态与容错流式算子聚合、Join需要维护状态这是 Flink 支持精确一次语义的基础。通过 Checkpoint 将状态持久化保证故障恢复后结果正确。4、性能优化MiniBatch减少状态访问频次提升吞吐。Local-Global 聚合两阶段聚合缓解数据倾斜。Idle State TTL清理无用状态防止无限膨胀。分区下推如 Doris 源表可利用分区剪枝提高批处理速度。5、批处理与流处理的技术选型如果你的场景需要全量数据关联复杂维度、生成每日报表批处理更简单高效。如果关注分钟级/秒级的指标变化必须用流处理但要权衡数据延迟、吞吐、状态大小与最终一致性。