什么是数据流

数据流的属性

数据流是无边界(无限且持续增长)数据集的抽象表示
例如：信用卡交易，包裹递送，游戏物体的移动
数据流(事件流)的其他属性

1. 事件流是有序的 先存钱再花钱
2. 事件流是不可变的 订单取消并不是说它就消失了
3. 事件流是可重播的 让现代业务领域的流式处理大获成功
   非结构化的键值对，半结构化的JSON,结构化的Avro

3种编程范式

1. 请求与响应范式，延迟最小的一种范式，处理模式一般是程序向系统发出请求，然后等待响应。例如OLTP系统
2. 批处理，具有高延迟和高吞吐量的特点。处理模式一般是通过定时任务定时定频率的触发任务执行。例如OLAP系统，数据仓库，商业智能
3. 流式处理，延迟介于请求与响应模式和批处理模式，适合那些业务流程持续进行，既不要求亚豪秒响应，也接受不了等到第二天才知道结果。例如交易报警，根据工序实时调整价格

流式处理的概念

时间

事件时间：事件发生的时间
处理时间：应用程序接受到时间之后对其处理的时间

状态

事件与事件之间的联系，保存在应用程序本地变量中

1. 本地状态：只能被单个应用程序实例访问，一般使用内嵌在应用程序中的数据库维护和管理
2. 外部状态：能被多个应用程序访问，一般使用Nosql数据库存储

流和表的二元性

1. 表都有一个主键，并包含一系列通过schema定义的属性，是事件的结果。
2. 流包含了一系列事件，是事件的过程。
   将表转化为流，需要应用流里包含所有变更。
   

时间窗口

大部分针对流的操作都是基于时间窗口的：移动平均数，一周销量最好的鞋子

1. 窗口的大小。窗口越小，越快发现变更，但是噪声也越大。窗口越大，变更越平滑，但是延迟很严重
2. 窗口移动的频率。5分钟的平均数可以每分钟变化一次
3. 窗口可更新时间。已经计算了窗口大小的事件逻辑，但是后面又有对应时间段的事件发生。
   

流式处理的设计模式

每个流式处理系统都不一样，从生产者和消费者，到使用Sprk Streaming和机器学习软件包的复杂集群。

1. 单个事件处理(map模式，filter模式)
   这种模式无需维护状态，每个事件都是独立处理
   
2. 使用本地状态
   大部分流式处理应用程序关心的是如何聚合信息。特别是基于时间窗口的聚合
   例如：基于某时间段的股票信息
   
   使用本地状态考虑的问题
   a) 内存使用：应用实例必须有足够内存
   b) 持久化：确保程序关闭不会丢失状态，重启或切换另一个实例可恢复状态
   c) 再均衡：重分配分区时，失去分区的实例必须把最后的状态保存起来
3. 多阶段处理和重分区
   本地状态：很好的实现了按组聚合操作
   问题:怎么实现跨应用实例聚合 例如：计算前10☞股票
   解决：两阶段解决-a) 计算每支股票的涨跌,都写到单个分区中
   b) 另一个应用程序读取这个分区，找出前10支股票
   
4. 使用外部查找-流和表的连接
   将外部数据和流集成到一起，比如：使用db的规则验证事务，将用户信息填充到点击事件中
   
   缺点：外部查询带来延迟，外部系统接受不了这么高的负载
   解决：缓存数据库
   捕捉数据库的变更事件并形成事件流(CDC-change data capture)
   某些连接器可以执行cdc任务，把表转换为事件流，一旦数据库发生变更，则更新本地缓存

5. 流与流的连接
通过流来表示表，就可以忽略大部分历史事件，因为只关心表的当前状态。
连接流定义：将两个流中具有相同键且发生在相同窗口内的事件匹配起来
例如：用户输入浏览器的搜索事件流+用户对搜索结果的点击事件流，来分析 哪一个搜索的热度更高？？？

6. 乱序的事件

a)识别乱序事件
b)规定可重排乱序的时间范围
c)具有乱序重排的能力
d)具备更新结果的能力

Stremas示例

底层的Processor API，高级的Streams API
流式处理就是 为事件流定义转换链
拓扑topology：一个有向图，包含几个转换事件流的过程。

1. map-filter简单聚合
   无需安装任何软件，就可运行实例。只需启动多个实例就可拥有一个集群

public class WordCount{
    public static void main(String[] args){
         Properties props=new Properties();
         props.put(StreamsConfig.Application_id,"wordcount"); //每个应用程序都要有一个应用ID
         props.put(StreamsConfig.ServersConfig,"ip:port"); // Kafka配置

         KStreamBuilder builder=new KStreamBuilder(); //创建拓扑
         KStream<String,String> source=builder.stream("wordcount-input"); //构件流并指向输入主题
         
         Pattern pattern=Pattern.compile("\\w+");
         //1. 从主题读取每一个事件就是一样文字，然后将一行文字拆分成多个单词，将单词作为事件的键，然后执行group by
         source.flatMapValues(value.toLowerCase()).map(()->{
              new KeyValue<Object,Object>(value,value)
         })
         //2. 过滤单词the
         .filter((key,value)->{
              !value.equals("the")
         })
         //3. 得到不重复的单词集合__重新分区
         .groupByKy()
         //4. 计算每个集合的事件数
         .count(“countStore”).mapValues(value->{
          Long.toString(value).toStream(); //将结果转为String类型，方便kafak读取结果
          });
          counts.to("wordcount-output");
    }
}

KafkaStream kafkastream=new KafkaStreams(builder,props);
kafkastream.start();
Thread.sleep(5000);
Kafkastream.close();

2. 基于时间窗口的聚合-股票的统计信息
   从股票交易事件流中 读取事件，然后计算没5秒钟内最好估盘价，交易股数，平均估盘价
   估盘价：卖方愿意出售的价格
   要价规模：卖方愿意在估盘价的基础上 出售的股数
   买入价：买方愿意支付的价格

3. 填充点击事件流-演示流的连接