引言

关于Kafka

• 作为消息队列。由于 Kafka 拥有高吞吐量，并且内置消息主题分区，备份，容错等特性，使得它更适合使用在大规模，高强度的消息数据处理的系统中。
• 流计算系统的数据源。流数据产生系统作为 Kafka 消息数据的生产者将数据流分发给 Kafka 消息主题，流数据计算系统 (Flink，Storm,Spark Streaming等) 实时消费并计算数据。这也是本文将要介绍的应用场景。
• 系统用户行为数据源。这种场景下，系统将用户的行为数据，如访问页面，停留时间，搜索日志，感兴趣的话题等数据实时或者周期性的发布到 Kafka 消息主题，作为对接系统数据的来源。
• 日志聚集。Kafka 可以作为一个日志收集系统的替代解决方案，我们可以将系统日志数据按类别汇集到不同的 Kafka 消息主题中。
• 事件源。在基于事件驱动的系统中，我们可以将事件设计成合理的格式，作为 Kafka 消息数据存储起来，以便相应系统模块做实时或者定期处理。由于 Kafka 支持大数据量存储，并且有备份和容错机制，所以可以让事件驱动型系统更加健壮和高效。

关于Flink

• 最底层级的抽象仅仅提供了有状态流。它将通过过程函数（Process Function）嵌入到DataStream API中。它允许用户可以自由地处理来自一个或多个流数据的事件，并使用一致、容错的状态。除此之外，用户可以注册事件时间和处理事件回调，从而使程序可以实现复杂的计算。
• 实际上，大多数应用并不需要上述的低层级抽象，而是针对 核心API（Core APIs） 进行编程，比如DataStream API（有界或无界流数据）以及DataSet API（有界数据集）。这些流畅的API为数据处理提供了通用的构建模块，比如由用户定义的多种形式的转换（transformations），连接（joins），聚合（aggregations），窗口操作（windows），状态（state）等等。这些API处理的数据类型以类（classes）的形式由各自的编程语言所表示。

• Table API 是以 表 为中心的声明式DSL，其中表可能会动态变化（在表达流数据时）。Table API遵循（扩展的）关系模型：表具有附加的模式（类似于关系数据库中的表），同时API提供可比较的操作，例如select、project、join、group-by、aggregate等。Table API程序声明式地定义了 什么逻辑操作应该执行 而不是准确地确定 这些操作代码的看上去如何 。 尽管Table API可以通过多种类型的用户定义的函数进行扩展，其仍不如 核心API 更具表达能力，但是使用起来却更加简洁（代码量更少）。除此之外，Table API程序还可以在执行之前通过应用优化规则的优化器。

• Flink提供的最高层级的抽象是 SQL 。这一层抽象在语法与表达能力上与 Table API 类似，但是是以SQL查询表达式的形式表现程序。SQL抽象与Table API交互密切，同时SQL查询可以直接在Table API定义的表上执行。

案例介绍与Flink SQL编程实现

1.案例介绍

2.案例分析

3.生产行为数据信息

4.编写Flink SQL编写程序实时处理数据

• 创建源表和结果表
• 创建子查询，使用 ROW_NUMBER() 窗口函数来对数据根据排序列进行排序并标上排名
• 外层查询中，对排名进行过滤，只取前N条，如N=5，那么就是取 Top 5 的数据，即网页热度排名

-- Kafka源表
create table input（
id             varchar,
count          bigint,
timegap        bigint,
positive       boolean,
firsttime      timestamp,
WATERMARK wk FOR firsttime as withOffset(firsttime, 2000)
)with(
type='kafka'
endpoint='xxx'
......
);
-- Hbase存储表
create table output(
window_start  TIMESTAMP,
window_end    TIMESTAMP,
id            bigint,
contribution  double
)with(
type='ALIHBASE'
endpoint='xxx'
......
);
-- 5秒钟窗口贡献度统计
CREATE VIEW group_view AS
SELECT id,
      TUMBLE_START(firsttime, INTERVAL '5' SECOND) AS start_time,
      (0.8*count+0.8*timegap+positive) as contribution
FROM input
GROUP BY id, TUMBLE(firsttime, INTERVAL '1' MINUTE);

-- 统计每5秒 top5 贡献值ID，并输出
insert into output
select
(
   SELECT
   id, firsttime, contribution,
   ROW_NUMBER() OVER (PARTITION BY firsttime ORDER BY contribution DESC) as rownum,
   FROM
   group_view
)
WHERE rownum <= 5;

注意事项

• 需要合理的设置时间窗口，即需要保证Flink的计算窗口合理地统计到热度最高的话题，理论上Flink的窗口可以是无界的，也可以是很小的时间窗口，但是合理的窗口大小设计对业务逻辑的影响。
• 虽然本文案例中，我们只是把 (近) 实时计算结果打印出来，但是实际上很多时候这些结果会被保存到数据库,HDFS, 或者发送回 Kafka, 以供其他系统利用这些数据做进一步的业务处理，Flink可以直接实现这些功能。
• 由于流计算对实时性要求很高，所以任何由于 JVM Full GC 引起的系统暂停都是不可接受的。Flink 采用类似 DBMS 的 sort 和 join 算法，直接操作二进制数据，从而使序列化/反序列化带来的开销达到最小。所以 Flink 的内部实现更像 C/C++ 而非 Java。如果需要处理的数据超出了内存限制，则会将部分数据存储到硬盘上。如果要操作多块MemorySegment就像操作一块大的连续内存一样，Flink会使用逻辑视图（AbstractPagedInputView）来方便操作。
• Flink内部支持exactly-once，要想达到端到端（Soruce到Sink）的exactly-once，需要Blink外部Soruce和Sink的支持，比如Source要支持精准的offset，Sink要支持两阶段提交，也就是继承TwoPhaseCommitSinkFunction。
• Flink中当所有输入的barrier没有完全到来的时候，早到来的event在exactly-once的情况向会进行缓存（不进行处理），而at-least-once的模式下即使所有输入的barrier没有完全到来的时候，早到来的event也会进行处理。也就是说对于at-least-once模式下，对于下游节点而言，本来数据属于checkpoint n的数据在checkpoint n-1里面也可能处理过了。所以我们建议，Flink的checkpoint模式设置为exactl-once模式。

结束语