Flink 1.11 API从入门到实战：DataStream与SQL核心用法详解

“Flink从0到1”这个系列写到第七篇前面几篇我陆续聊过环境搭建、核心概念和运行机制到了这篇终于该进入正题了动手写代码。我平时收到最多的私信就是这类问题——“概念都看懂了但一个作业到底从哪句代码开始”如果你也卡在这一步这篇正好对症。我会把Flink 1.11里日常开发最常用的API完整过一遍DataStream API的Source/Transform/Sink三件套、Table API与SQL的建表查询写法以及状态、时间、窗口这些写API时躲不开的底层内容。文中的代码示例都以1.11为准基本可以照着敲。很多资料习惯一上来就把算子挨个列一遍每个都解释一遍结果读者看完还是不知道怎么写一个完整作业。我的建议是反过来先记住一张图你写的所有代码本质上都是在描述一条数据流水线——从某个Source读数据进来经过若干次Transform最后落到某个Sink。API再多骨架就这一个。这篇后面的所有内容都围绕这张图展开。1. Flink 1.11的API分级别一上来就盯底层算子1.1 三层结构各自解决什么问题Flink给开发者提供了三层API从底到顶分别是ProcessFunction级别、DataStream API级别、Table API与SQL级别。最底层是ProcessFunction这一类的函数式接口你可以拿到单条数据、访问状态、注册定时器还能手动控制输出灵活度最高但写起来也最繁琐因为它把每一条数据的处理逻辑都交给你自己组织。中间层是DataStream API通过map、flatMap、filter、keyBy、window这些算子组合流水线既能保留对数据流的精细控制又不需要事无巨细都自己实现。最上层是Table API和SQL把流当成一张动态表来查询几行声明式语句就能完成大部分统计需求开发效率最高。新手最容易犯的错是想一开始就把ProcessFunction搞透觉得那才算“懂原理”。实际上你真正需要ProcessFunction的场景在绝大多数业务里可能一个月都碰不到一次。先从DataStream API和SQL入手性价比最高等这些用熟了再回来看底层就水到渠成。1.2 为什么单独拎出1.11这一版现在回头看1.11不算新版本但它是一个很重要的分水岭。1.11是blink planner正式转正的一版Table API和SQL的能力得到大幅增强executeSql支持DDL流式维表join、CDC格式解析也开始进入稳定范围。也就是说从1.11开始你用Flink SQL做实时计算才算真正到了“生产可用的平民路线”阶段。与此同时DataStream API的形态也稳定下来新引入的WatermarkStrategy等写法一直沿用至今。学1.11的API不只是学一个过渡版本而是学到了后续版本的骨干思路。2. DataStream API核心骨架Source、Transform、Sink逐一过2.1 创建执行环境写DataStream程序的第一个动作是拿到StreamExecutionEnvironmentStreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);第一行返回的就是当前上下文里的执行环境本地跑的时候它负责起一个MiniCluster提交到集群以后它会连接JobManager。第二行是设置时间语义为事件时间1.11里时间语义仍然是手动指定的后面讲Watermark时会展开。如果你的作业不涉及事件时间窗口这行可以先不写但写了也不会错。这里有个小提醒getExecutionEnvironment和createLocalEnvironment有区别。前者会根据你提交作业的方式自动判断——在IDE里跑就是本地模式用flink run提交就是集群模式。如果你图省事直接用前者就好。2.2 Source从集合、文件、Kafka读取Source环节决定了数据从哪来也直接影响后面所有计算的正确性。最常用的有三种方式。从本地集合读主要用于单元测试和快速验证DataStreamInteger numbers env.fromCollection(Arrays.asList(1, 2, 3, 4, 5));从文件读适合离线批处理和本地调试DataStreamString lines env.readTextFile(/data/input/orders.log);从Kafka读这是实时作业的主力。这里要特别说明Flink 1.11的一个坑点1.11时Kafka连接器的主推写法还是老一代的FlinkKafkaConsumer011类名里的“011”对应Kafka客户端的版本世代不是你Kafka集群的版本号别对不上就觉得是自己环境不对。用法如下Properties props new Properties(); props.setProperty(bootstrap.servers, localhost:9092); props.setProperty(group.id, flink-demo); DataStreamString stream env.addSource( new FlinkKafkaConsumer011(orders, new SimpleStringSchema(), props) );SimpleStringSchema会把Kafka里的字节直接转成字符串。如果你的消息是JSON实际生产里一般会自己实现反序列化或者用JSON格式的Schema。这里先记住这个套路addSource加上一个SourceFunction。2.3 Transform最常用的几个算子Transform是流水线的中间环节。我日常写作业高频算子其实就那么几个。map是一对一转换一个输入产生一个输出。例如把字符串按照逗号拆开变成一条包含用户ID和金额的对象DataStreamOrder orders stream.map(line - { String[] fields line.split(,); return new Order(Long.parseLong(fields[0]), Double.parseDouble(fields[1])); });flatMap是一对多转换一个输入可以产生零条、一条或多条输出经典的wordcount就是用它把一个句子拆成多个单词。filter是过滤保留满足条件的记录DataStreamOrder validOrders orders.filter(order - order.amount 0);keyBy是分组它会把相同key的数据shuffle到同一个子任务里后面才能做各自独立的聚合或者窗口计算KeyedStreamOrder, Long keyed validOrders.keyBy(order - order.userId);reduce和sum这类算子都必须跟在keyBy之后因为它们做的是“组内聚合”。很多新手在没keyBy的流上直接调sum编译都过不了原因就在这里。2.4 Sink结果真正落地的地方Sink负责把结果写出去。调试阶段最简单的是print()它会把结果打印到日志或控制台一行一条方便看到数据长什么样。生产环境不会用print而是用addSink往Kafka、MySQL、ES、HBase等外部系统写。这里我要强调一个特别容易让新手困惑的点print在并行度大于1时输出顺序是乱的而且不同子任务的输出会混在一起。你以为程序出bug了其实只是并行子任务在并发打印。想看有序结果要么把print所在算子的并行度临时设为1要么在输出里带上子任务编号比如print(处理结果)会在前缀里带上编号。3. Table API与SQL更快写出生产可用的统计任务3.1 StreamTableEnvironment初始化1.11里用SQL需要创建一个StreamTableEnvironment注意它和DataStream的StreamExecutionEnvironment是两个东西需要桥接起来EnvironmentSettings settings EnvironmentSettings.newInstance() .useBlinkPlanner() .inStreamingMode() .build(); StreamTableEnvironment tEnv StreamTableEnvironment.create(env, settings);useBlinkPlanner在1.11里强烈建议加上因为blink planner是当时默认且功能更完整的执行引擎而老的planner已经被边缘化。inStreamingMode()表示以流模式运行如果去掉它某些流式特殊的语法比如动态表的Retract行为就会用批模式的逻辑去解释结果可能完全不对。3.2 用executeSql建一张Kafka表1.11最大的便利之一是可以在代码里直接通过DDL建表。下面这段是典型的Kafka源表定义CREATE TABLE orders ( order_id BIGINT, user_id BIGINT, amount DOUBLE, order_time TIMESTAMP(3), WATERMARK FOR order_time AS order_time - INTERVAL 10 SECOND ) WITH ( connector kafka, topic orders, properties.bootstrap.servers localhost:9092, properties.group.id flink-sql-demo, format json, scan.startup.mode earliest-offset )这段DDL做了三件事声明字段、声明Watermark策略、声明连接器参数。注意WATERMARK FOR order_time AS order_time - INTERVAL 10 SECOUND这个写法它等价于在DataStream API里设置延迟10秒的有界乱序Watermark。源表的Watermark可以建在DDL里这在1.11里已经支持比在Java代码里手动指定省事得多。然后你在代码里直接tEnv.executeSql(CREATE TABLE orders ...)执行建表就能接着写查询了。3.3 查询结果转DataStreamappend和retract的区别用SQL做了聚合之后结果怎么拿回DataStream API继续处理是新手最容易懵的地方。Flink 1.11提供了两种转换方式。如果是简单过滤、维表补充这类没有聚合的操作结果表只会新增数据可以用toAppendStreamTable result tEnv.sqlQuery(SELECT user_id, amount FROM orders WHERE amount 100); DataStreamRow appendStream tEnv.toAppendStream(result, Row.class);如果是聚合查询结果在流上是一个持续变化的状态既会有新增也会有更新和删除就必须用toRetractStream。它返回的是DataStreamTuple2Boolean, Row第一个字段为true代表这是一条新插入或更新后的数据false代表这是一条需要撤销的旧数据Table aggResult tEnv.sqlQuery( SELECT user_id, SUM(amount) AS total_amount FROM orders GROUP BY user_id ); DataStreamTuple2Boolean, Row retractStream tEnv.toRetractStream(aggResult, Row.class);很多初学者拿到Tuple2之后不知道第一个字段干嘛用直接忽略往下游Sink一写发现数据对不上。记住Retract模式是流式聚合的本质只有理解了“先删除旧值、再写入新值”你才能正确处理Flink SQL的聚合结果。3.4 1.11里SQL还能干的几件实事除了基本聚合1.11的SQL还让我比较惊喜的是临时表的维表join语法是FOR SYSTEM_TIME AS OF。原理是把一张外部维表比如MySQL里的用户信息表注册成临时表然后流式主表每来一条数据都去维表里关联出补充字段代码比DataStream API手动做异步IO简洁太多。例如SELECT o.user_id, u.user_name, o.amount FROM orders o JOIN user_info FOR SYSTEM_TIME AS OF o.order_time AS u ON o.user_id u.user_id这个语法在实时数仓场景里非常常用把“流与维表的关联”从代码复杂度降了一个量级。4. 状态、时间与窗口写API时绕不开的三个底层概念4.1 为什么状态会决定作业能不能重启状态是Flink区别于普通消息处理引擎的核心能力通俗点说它就是流上保留的“记忆”。Flink把状态统一管理起来做checkpoint才能在故障后把作业恢复到上次一致的位置继续跑。没有状态的话作业重启后什么都记不住聚合结果必然从零开始这显然不是我们要的。在API层面状态分为算子状态和键控状态。日常使用最多的是键控状态即keyBy之后每个key都有一份独立的状态。两个不同用户的数据会落到不同的状态分片中互不干扰。这个隔离机制是Flink能够大规模并行处理的关键。4.2 keyed state三个最常用的接口具体写代码时键控状态通常放在函数类的open方法里初始化。下面是一个典型的KeyedProcessFunction它统计每个用户累计出现了多少条订单public class UserOrderCountFunction extends KeyedProcessFunctionLong, Order, Tuple2Long, Long { private transient ValueStateLong countState; Override public void open(Configuration parameters) { countState getRuntimeContext().getState( new ValueStateDescriptor(order-count, Long.class) ); } Override public void processElement(Order order, Context ctx, CollectorTuple2Long, Long out) throws Exception { Long count countState.value(); if (count null) count 0L; countState.update(count 1); out.collect(Tuple2.of(order.userId, countState.value())); } }这里有一个必须注意的点状态字段必须声明为transient并且在open方法里初始化不能直接new出来。因为Flink序列化整个算子时会重新构造实例直接在字段里初始化运行时拿到的是未初始化的值。这个坑我没有一百次也有几十次见过别人踩进去。ValueState是最简单的单值状态。如果状态是一个列表用ListState如果是一个键值映射用MapState。它们的名称里都有一个Descriptor这个描述符的作用除了命名更重要的是标明状态的类型Flink序列化和恢复状态全靠它。4.3 时间语义EventTime和Watermark窗口计算离不开时间。Flink有处理时间、事件时间和摄入时间三种语义实时计费、风控这类场景几乎都用事件时间也就是事件真实发生的时间。但事件时间是存在乱序和延迟的所以需要Watermark来告诉系统“到此为止该触发窗口了”。Flink 1.11推荐的Watermark生成方式已经统一为WatermarkStrategyDataStreamOrder withWatermark stream .assignTimestampsAndWatermarks( WatermarkStrategy .OrderforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((order, timestamp) - order.orderTimeMs) );forBoundedOutOfOrderness(10秒)的含义是允许事件最多迟到10秒超过这个范围的迟到数据默认会被丢弃。这里的10秒完全取决于业务容忍度不是拍脑袋定的需要根据上游数据乱序情况来评估。我在实际项目中会先跑一段历史数据统计一下事件时间戳和到达时间的差值分布再去设置这个参数。4.4 窗口API的三种形态拿到带Watermark的流之后窗口就顺理成章。1.11里推荐用.window()而不是老的timeWindow()虽然两者都还能用但window接入的窗口API更完整后续版本也确实把timeWindow淘汰了keyedStream .window(TumblingEventTimeWindows.of(Time.minutes(5))) .reduce((order1, order2) - merge(order1, order2));滚动窗口是固定长度、互不重叠比如每5分钟一个窗口滑动窗口是有重叠的比如每5秒滑动一次看最近5分钟会话窗口是按数据活跃度来分的数据之间沉默超过设定间隔就开一个新窗口。选哪种取决于业务要的“统计节奏”。做实时大屏更多用滚动窗口做持续监控预警常常用滑动窗口做用户行为会话分析则首选会话窗口。5. 把API串起来订单金额实时统计的完整例子5.1 需求拆解与方案选择先看一个具体的需求某电商平台要实时统计每个用户从Kafka主题里过来的订单金额每5分钟输出一次当前窗口内每个用户的累计下单金额。收到这个需求先别急着写代码拆一下数据源是Kafka里的订单消息需要解析成结构化对象统计口径是窗口聚合按用户分组窗口是5分钟滚动时间语义选事件时间结果写到下游存储方便展示或触发后续流程。我这次故意把方案定为“先用DataStream API实现再对比SQL实现”因为两个方案的取舍刚好能覆盖一批读者的问题到底该学哪个、该用哪个。5.2 用DataStream API实现先定义一个简单的Order类字段和Kafka里的JSON对应public class Order { public long userId; public long orderId; public double amount; public long orderTimeMs; public Order() {} public Order(long userId, long orderId, double amount, long orderTimeMs) { this.userId userId; this.orderId orderId; this.amount amount; this.orderTimeMs orderTimeMs; } }然后写作业主流程StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); Properties props new Properties(); props.setProperty(bootstrap.servers, localhost:9092); props.setProperty(group.id, order-stat-group); DataStreamString source env.addSource( new FlinkKafkaConsumer011(orders, new SimpleStringSchema(), props) ); DataStreamOrder orders source.map(line - { // 实际项目里这里会接入JSON反序列化工具比如Jackson或Fastjson String[] fields line.split(,); return new Order( Long.parseLong(fields[0]), Long.parseLong(fields[1]), Double.parseDouble(fields[2]), Long.parseLong(fields[3]) ); }); DataStreamOrder watermarked orders.assignTimestampsAndWatermarks( WatermarkStrategy .OrderforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((order, ts) - order.orderTimeMs) ); DataStreamTuple2Long, Double result watermarked .keyBy(order - order.userId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .reduce((order1, order2) - { order1.amount order2.amount; return order1; }) .map(order - Tuple2.of(order.userId, order.amount)); result.addSink(new YourRedisSink()); env.execute(order-amount-stat-job);注意reduce里我复用了第一个参数对象这是一种常见的减法优化但在状态恢复或者调试时容易出现残留值问题生产里我其实更推荐new一个新对象出来虽然多一次构造开销但语义更干净。为了演示简洁这里保留这种写法你写的时候还是新建对象更稳。5.3 换成SQL实现会短多少同样的需求如果换成SQL代码量会明显少一截StreamTableEnvironment tEnv StreamTableEnvironment.create(env); tEnv.executeSql(CREATE TABLE orders ( ... ) WITH ( connector kafka, ... )); Table result tEnv.sqlQuery( SELECT user_id, SUM(amount) AS total_amount, TUMBLE_END(order_time, INTERVAL 5 MINUTE) AS window_end FROM orders GROUP BY TUMBLE(order_time, INTERVAL 5 MINUTE), user_id ); DataStreamTuple2Boolean, Row stream tEnv.toRetractStream(result, Row.class);这段SQL里有一个细节我特意在SELECT里带了TUMBLE_END这样下游可以知道每条结果属于哪个窗口不然5分钟窗口聚合结果混在一起下游用的时候很痛苦。这个习惯是我做了几次实时报表之后才养成的否则下游展示层根本没有窗口边界概念。两种写法的取舍很清晰SQL适合逻辑简单、边界明确、开发效率优先的场景DataStream API适合需要精细控制或特殊逻辑的场景。我现在的习惯是能用SQL就先用SQL跑通遇到SQL表达不了的逻辑再退回到DataStream API补。5.4 跑起来之后要观察什么一个实时作业写完不等于结束跑起来之后要观察的指标和判断方法我说几个自己固定会看的点。看Watermark是否推进。在Flink的Web UI上Watermark指标如果一直停在最小值窗口就永远不触发结果一直不出。常见原因是事件时间字段解析失败导致所有数据的Watermark都算不出来。这个排查思路在第6节会再展开。看窗口是否按时触发。日志里打印的窗口触发时间应该和配置一致如果窗口一直不触发也不报错基本就是Watermark的问题或并行子任务里某个分区没有数据。看结果是否重复。如果使用Retract模式下游必须有能力处理撤回数据否则结果会越算越多。做实时报表的时候务必先确认下游能不能理解“这条是删除旧值”。6. 新手最容易踩的API坑现象、原因、解法6.1 print结果看起来乱序现象print出来的数据顺序忽前忽后以为计算逻辑错了。原因并行度大于1多个子任务并发输出。解法临时把print所在算子并行度设为1或者用print带前缀标识子任务例如print(result)。这只是观察手段生产环境不会依赖print。6.2 Watermark一直不推进现象窗口一直不输出结果UI上Watermark为负的很大或不变。原因通常是三类事件时间字段没有正确指定、数据里没有新时间戳推进、有子任务没有数据导致Watermark在某个并行分支停滞。解法是逐项排查先确认withTimestampAssigner取到的时间戳是毫秒级再看Kafka分区是否有的分区没有新数据最后考虑1.11里可以通过配置空闲源延迟env.getConfig().setAutoWatermarkInterval和外部空闲源策略来处理。6.3 聚合结果和预期差一倍现象同样的数据用Flink算出来的总额和乐观预期差一倍。原因大概率是Retract模式没有正确处理聚合结果里每来一条更新会产生两条消息一条false删除旧值、一条true写入新值如果你只把true的数据写进存储这个差值就出来了。解法检查下游是否消费了完整的数据流并在存储层做幂等更新不要只取一半。6.4 状态无限变大作业越来越慢现象运行一段时间后作业延迟升高UI上状态大小持续增长。原因可能是key的基数太大或者状态没有设置过期时间。解法给状态设置TTL比如new ValueStateDescriptor配合StateTtlConfig.newBuilder(Time.days(3)).build()。另一个更根本的办法是优化业务逻辑减少需要缓存的粒度。做用户维度统计时如果用户量是亿级单个key存一个对象都可能是杀手最好用外部存储做维表Flink内置状态只留最近窗口的中间结果。6.5 反序列化层和下游表结构不一致现象作业能跑但结果字段串位或报错。原因Kafka里的消息结构已经变了但建表语句或反序列化Schema没有同步更新。这里我的经验是JSON格式的天然容错性很容易掩盖这种问题字段对不上通常不会报错只是解析出空值。解决办法是加一个字段校验环节比如在map阶段对关键字段做空值检查任何字段为null就抛异常让作业快速失败而不是带病运行。6.6 一个小表格收尾症状大概率原因排查方向窗口不触发Watermark不推进时间戳解析、空闲分区、乱序上限print乱序并行子任务并发输出调并行度或带前缀打印聚合结果重复加倍Retract模式未正确消费检查下游是否处理false值延迟持续升高状态膨胀/数据倾斜加TTL、优化key设计、关注峰值key这个表格是我每次给新人做分享时的固定内容尽量用最短路径定位问题。最后再分享一个我自己的习惯不管用SQL还是DataStream API写作业我都先在一张纸上把数据流图画出来标注好每个环节的keyBy依据、窗口尺寸、时间语义然后才动手写代码。看似多花了五分钟实际省下的调试时间远超这个数。把这套API骨架吃透后面再遇到各种“高级API”你会发现底层全是这些算子的组合。