Flink项目总结问题

Flink 问题

https://gitee.com/felix_fxf/gmall2024-realtime.git

待解决问题

为什么要用HBase, 直接到Kafka不行？

关键词：

Flink 测出流 状态编程 日志分流 精准一次性 事务 隔离级别

Flink状态编程

ValueState 是基于键（key）的状态，Flink 会根据键将数据分发到不同的并行实例上，并且每个键的状态是唯一的。这意味着在不同的并行实例中，相同键的状态是共享的，并且可以被正确地更新和访问。例如，对于不同用户的最后访问日期，Flink 会根据用户 ID 这个键来管理每个用户的状态，确保每个用户的状态是一致的。

Flink(端到端)的一致性

source端：起码提供可重置偏移量(数据可重放的能力)，

tranform：检查点机制

sink端：幂等，事务

API级别

• SQL
• DataSet
• DataStream
• process

侧出流

// 定义侧出流标签
OutputTag<String> dirtyTag = new OutputTag<String>("dirtyTag"){};

容错

Kafka生产和消费保证数据的精准一次性

Kafka消费者的
精准一次性：手动维护偏移量
至少一次
消费的精准一次性：开启事务
写入kafka 精准一次性
幂等
事务
预提交
两阶段提交

• KafkaSource: 从Kafka主题中读取数据

  • 通过手动维护偏移量，保证消费的精准一次性

• KafkaSink: 向Kafka主题中写入数据，也可以保证写入的精准一次性，需要做如下操作

  Flink-1.17官方说明：Kafka | Apache Flink

  DeliveryGuarantee.EXACTLY_ONCE: 该模式下，Kafka sink 会将所有数据通过在 checkpoint 时提交的事务写入。
  因此，如果 consumer 只读取已提交的数据（参见 Kafka consumer 配置 isolation.level），在 Flink 发生重启时不会发生数据重复。
  然而这会使数据在 checkpoint 完成时才会可见，因此请按需调整 checkpoint 的间隔。
  请确认事务 ID 的前缀（transactionIdPrefix）对不同的应用是唯一的，以保证不同作业的事务 不会互相影响！
  此外，强烈建议将 Kafka 的事务超时时间调整至远大于 checkpoint 最大间隔 + 最大重启时间，否则 Kafka 对未提交事务的过期处理会导致数据丢失。

  • 开启检查点

  • 配置生产者消费一次性

    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)

  • 设置事务ID前缀保证对不同的应用是唯一的

    .setTransactionalIdPrefix("dwd_base_log_")

  • 设置事务超时时间 (检查点超时时间 < 事务的超时时间 <= 事务的最大超时时间)

    .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15*60*1000 + "")

  • 在消费端，需要设置消费的隔离级别为读已提交(默认为读为提交：预提交的数据也能读到)

    .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")

数据埋点产生的数据问题

• 判断新老顾客，通过缓存

  Web 端：用户第一次访问埋入神策 SDK 页面的当天（即第一天），JS SDK 会在网页的 cookie 中设置一个首日访问的标记，并设置第一天 24 点之前，该标记为 true，即第一天触发的网页端所有事件中，is_new = 1。第一天之后，该标记则为 false，即第一天之后触发的网页端所有事件中，is_new = 0；

  本项目模拟生成的是 APP 端日志数据。对于此类日志，如果首日之后用户清除了手机本地缓存中的标记，再次启动 APP 会重新设置一个首日为 true 的标记，导致本应为 0 的 is_new 字段被置为 1，可能会给相关指标带来误差。因此，有必要对新老访客状态标记进行修复。

双流Join

Join | Apache Flink

Flink SQL支持对动态表进行复杂而灵活的连接操作。 为了处理不同的场景，需要多种查询语义，因此有几种不同类型的 Join。

默认情况下，joins 的顺序是没有优化的。表的 join 顺序是在 FROM 从句指定的。可以通过把更新频率最低的表放在第一个、频率最高的放在最后这种方式来微调 join 查询的性能。需要确保表的顺序不会产生笛卡尔积，因为不支持这样的操作并且会导致查询失败。

Regular Joins #

Regular join 是最通用的 join 类型。在这种 join 下，join 两侧表的任何新记录或变更都是可见的，并会影响整个 join 的结果。 例如：如果左边有一条新纪录，在 Product.id 相等的情况下，它将和右边表的之前和之后的所有记录进行 join。

Interval Joins #

返回一个符合 join 条件和时间限制的简单笛卡尔积。Interval join 需要至少一个 equi-join 条件和一个 join 两边都包含的时间限定 join 条件。范围判断可以定义成就像一个条件（<, <=, >=, >），也可以是一个 BETWEEN 条件，或者两边表的一个相同类型（即：处理时间 或 事件时间）的时间属性 的等式判断。

Temporal Joins #

时态表（Temporal table）是一个随时间变化的表：在 Flink 中被称为动态表。时态表中的行与一个或多个时间段相关联，所有 Flink 中的表都是时态的（Temporal）。 时态表包含一个或多个版本的表快照，它可以是一个变化的历史表，跟踪变化（例如，数据库变化日志，包含所有快照）或一个变化的维度表，也可以是一个将变更物化的维表（例如，存放最终快照的数据表）。

Lookup Join #

水位线

getUpsertKafkaDDL

旁路缓存

异步IO

维度层和数据明细层

Doris

报表分析

即席查询

统一数仓构建

窗口聚合 | Apache Flink

搜索关键词聚合统计

Flink 有 spark stream没有：事件时间语义

wartermark 水位线 事件时间衡量，逻辑时钟

默认指定wartermark两种方式

• 单调递增：有界乱序的一种特殊情况(乱序程度0)
• 有界乱序

旁路缓存

异步IO

异步 I/O | Apache Flink

Flink 的异步 I/O API 允许用户在流处理中使用异步请求客户端。API 处理与数据流的集成，同时还能处理好顺序、事件时间和容错等。

在具备异步数据库客户端的基础上

ODS数据采集

日志数据采集

flume组件：source sink cancel

2024-10-10日停止维护

业务数据采集

maxwell: binlog拿取数据

不具备将json格式封装为字符串

DIM维度层处理

业务数据库维度表变化实时同步到数仓

配置表中配置信息判断是不是维度，flinkCDC

广播流

主流数据处理：根据表名到

DWD事实表准备

流量域

日志分流

交易域

FlinkSQL实现：普通内外连接为参与连接的表默认维护一个状态，默认情况下这个状态永远不会失效

状态保留时间：传输延迟，业务滞后关系

左外连接，如果左表数据先到，右表数据后到，会产生3条数据(回撤流)

发送到kafka主题会记录三条数据：需要空数据处理，去重

普通内外连接关联维度，时间无法控制，Lookup Join左表驱动

工具域

互动域

用户域

实现：DataStreamAPI，SQL

DWS汇总表的抽取以及轻度聚合

给ADS提供服务

列出指标，

统计周期、粒度、业务过程相同，这样的指标放在一张汇总表进行处理（汇总表的抽取)

写数据，轻度聚合，FlinkSQL FlinkAPI 读取数据创建动态表、指定waterMark、提取事件事件字段

处理、分组、开窗、聚合计算、写到Doris

往数据库写保证数据一致性

hbase 幂等

kafka 事务

doris union表可以幂等

去重

• 状态+定时器：失效性差，不会出现数据膨胀
• 状态+抵消

维度关联：基本维度关联：旁路缓存Redis，异步IO 支持异步操作数据库

异步编排对象

ADS分析和可视化操作

sugar