第五章数据管道和流式处理笔记

﹏ヽ暗。殇╰゛Y 2023-06-11 08:27 58阅读 0赞

# **一、数据管道基本概念** #

在使用Kafka构建数据管道时，通常有两种使用场景：第一种：把Kafka作为数据管道的两个端点之一，例如，把Kafka里的数据移动到云上，或者把MongoDB里的数据移动到Kafka里；第二种：把Kafka作为数据管道两个端点的中间媒介，例如，为了把DB的数据移动到ElasticSearch上，需要先把它们移动到Kafka里，再将它们从Kafka移动到Elastic Search上。

Kafka为数据管道带来的主要价值在于，它可以作为数据管道各个数据段之间的大型缓冲区，有效地解耦管道数据的生产者和消费者。数据管道的重要作用之一是解耦数据源和数据池，Kafka在这方面的能力以及在安全和效率方面的可靠性，使它成为构建数据管道的最佳选择。

## **1、数据管道需要考虑的问题** ##

### **1.1、及时性** ###

有些系统希望每天一次性地接收大量数据，而有些则希望在数据生成几毫秒之内就能拿到它们，大部分数据管道介于这两者之间。一个好的数据集成系统能够很好地支持数据管道的各种及时性需求，而且在业务需求发生变更时，具有不同及时性需求的数据表之间可以方便地进行迁移。

Kafka作为一个基于流的数据平台，提供了可靠且可伸缩的数据存储，可以支持几近实时的数据管道和基于小时的批处理。生产者可以频繁地向Kafka写入数据，也可以按需写入：消费者可以在数据到达的第一时间读取它们，也可以每隔一段时间读取一次积压的数据。

Kafka在这里扮演了一个大型缓冲区的角色，降低了生产者和消费者之间的时间敏感度。实时的生产者和基于批处理的消费者可以同时存在，也可以任意组合。实现回压策略也因此变得更加容易，Kafka本身就使用了回压策略（必要时可以延后向生产者发送确认），消费速率完全取决于消费者自己。

### **1.2、可靠性** ###

我们要避免单点故障，并能够自动从各种故障中快速恢复。数据通过数据管道到达业务系统，哪怕出现几秒钟的故障，也会造成灾难性的影响，对于那些要求毫秒级的及时性系统来说尤为如此。数据传递保证是可靠性的另一个重要因素。有些系统允许数据丢失，不过在大多数情况下，它们要求至少一次传递。也就是说，源系统的每一个事件都必须到达目的地，不过有时候需要进行重试，而重试可能造成重复传递。有些系统甚至要求仅一次传递——源系统的每一个事件都必须到达目的地，不允许丢失，也不允许重复。

### **1.3、高吞吐量和动态吞吐量** ###

为了满足现代数据系统的要求，数据管道需要支持非常高的吞吐量。更重要的是，在某些情况下，数据管道还需要能够应对突发的吞吐量增长。

由于我们将Kafka作为生产者和消费者之间的缓冲区，消费者的吞吐量和生产者的吞吐量就不会耦合在一起了。如果生产者的吞吐量超过了消费者的吞吐量，可以把数据积压在Kafka里，等待消费者追赶上来。通过增加额外的消费者或生产者可以实现Kafka的伸缩，因此我们可以在数据管道的任何一边进行动态的伸缩，以便满足持续变化的需求。

因为Kafka是一个高吞吐量的分布式系统，一个适当规模的集群每秒钟可以处理数百兆的数据，所以根本无需担心数据管道无住满足伸缩性需求。另外Connect API不仅支持伸缩，而且擅长并行处理任务。

### **1.4、数据格式** ###

数据管道需要协调各种数据格式和数据类型，这是数据管道的一个非常重要的因素。数据类型取决于不同的数据库和数据存储系统。你可能会通过Avro将XML或关系型数据加载到Kafka里，然后将它们转成JSON写入ElasticSearch，或者写入HDFS等等。Kafka与数据格式无关，生产者和消费者可以使用各种序列化器来表示任意格式的数据。

## **2、Kafka Connect** ##

这个不同于普通API，Connect是Kafka的一部分，它位在Kafka和外部数据存储系统之间移动数据提供一种可靠且可伸缩的方式，它不是一个API调用，它主责移动数据。

Connect以worker进程集群的方式运行，然后使用REST API来管理和配置，并且这些进程都是长时间持续运行的作业。比如使用将一个Mysql的表数据导入到一个Kafka的主题上，然后再将他们加载到ElasticSearch里，然后对它们的内容进行索引。

# **二、流式处理基本概念** #

Kafka早期版本一般被认为是一个强大的消息总线，可以传递事件流，但没有处理和转换事件的能力。Kafka可靠的传递能力让它成为流式处理系统完美的数据来源，很多基于Kafka构建的流式处理系统都将Kafka作为唯一可靠的数据来隙，如Apache Storm、Apache SparkStreaming、Apache Flink、Apache Samza等。

从0 . 10 . 0版本开始，Kafka不仅为每一个流行的流式处理框架提供了可靠的数据来源，还提供了一个强大的流式处理类库，并将其作为客户端类库的一部分。这样开发人员就可以在应用程序里读取、处理和生成事件，而不需要再依赖外部的处理框架。

## **1、数据流** ##

先来看看什么是数据流（也被称为“事件流”或“流数据”）。首先，数据流是无边界数据集的抽象表示。无边界意味着无限和持续增长。无边界数据集之所以是无限的，是因为随着时间的推移，新的记录会不断加入进来。

这个简单的模型（事件流）可以表示很多业务活动，比如信用卡交易、股票交易、包裹递送、流经交换机的网络事件、制造商设备传感器发出的事件、发送出去的邮件、游戏里物体的移动，等等。这个清单是无穷无尽的，因为几乎每一件事情都可以被看成事件的序列。

除了没有边界外，事件流模型还有其他一些属性。

### **1.1、事件流是有序的** ###

事件的发生总是有个先后顺序。以金融活动事件为例，先将钱存进账户后再花钱，这与先花钱再还钱的次序是完全不一样的。后者会出现透支，而前者不会。

### **1.2、不可变的数据记录** ###

事件一旦发生，就不能被改变。一个金融交易被取消，并不是说它就消失了，相反，这需要往事件流里添加一个额外的事件，表示前一个交易的取消操作。顾客的一次退货并不意味着之前的销售记录被删除，相反，退货行为被当成一个额外的事件记录下来。这是数据流与数据表之间的另一个不同点一一可以删除和修改数据表里的记录，但这些操作只不过是发生在数据库里的事务，这些事务可以被看成事件流。假设你对数据库的二进制日志（ bin log ）、预写式日志（ WAL ）和重做日志（ redo log ）的概念都很熟悉，那么就会知道，如果往数据库表插入一条记录，然后将其删除，表里就不会再有这条记录。但重做日志里包含了两个事务：插入事务和删除事务。

### **1.3、事件流是可重播的** ###

这是事件流非常有价值的一个属性。但对于大多数业务来说，重播发生在几个月前（甚至几年前）的原始事件流是一个很重要的需求。可能是为了尝试使用新的分析方法纠正过去的错误，或是为了进行审计。

## **2、什么是流式处理** ##

流式处理是指实时地处理一个或多个事件流。流式处理是一种编程范式，就像请求与响应范式和批处理范式那样。

### **2.1、请求与响应** ###

这是延迟最小的一种范式，响应时间处于亚毫秒到毫秒之间，而且响应时间一般非常稳定。这种处理模式一般是阻塞的，应用程序向处理系统发出请求，然后等待响应。在数据库领域。

### **2.1、批处理** ###

这种范式具有高延迟和高吞吐量的特点。处理系统按照设定的时间启动处理进程，比如每天的下午两点开始启动，每小时启动一次等。它读取所有的输入数据（从上－ 次执行之后的所有可用数据，或者从月初开始的所有数据等）．输出结果，然后等待下一次启动。处理时间从几分钟到几小时不等，并且用户从结果里读到的都是旧数据。它们每天加载巨大批次的数据，并生成报表，用户在下一次加载数据之前看到的都是相同的报表。从规模上来说，这种范式既高效又经挤。但在近几年，为了能够更及时、高效地作出决策，业务要求在更短的时间内能提供可用的数据，这就给那些为探索规模经济而开发却无法提供低延迟报表的系统带来了巨大的压力。

### **2.3、流式处理** ###

这种范式介于上述两者之间。大部分的业务不要求亚毫秒级的响应，不过也接受不了要等到第二天才知道结果。大部分业务流程都是持续进行的，只要业务报告保持更新，业务产品线能够持续响应，那么业务流程就可以进行下去，而无需等待特定的响应，也不要求在几毫秒内得到响应。一些业务流程具有持续性和非阻塞的特点，比如针对可疑信用卡交易的警告、网络警告、根据供应关系实时调整价格、跟踪包衷。

流的定义不依赖任何一个特定的框架、API 或特性。只要持续地从一个无边界的数据集读取数据，然后对它们进行处理并生成结果，那就是在进行流式处理。重点是，整个处理过程必须是持续的。一个在每天凌晨两点启动的流程，从流里读取500 条记录，生成结果，然后结束，这样的流程不是流式处理。

## **3、流式处理中的基本概念** ##

流式处理的很多方面与普通的数据处理是很相似的：写一些代码来接收数据，对数据进行处理，可能做一些转换、聚合和增强的操作，然后把生成的结果输出到某个地方。不过流式处理有一些特有的概念，我们可以适当了解一下。

### **3.1、时间** ###

在流式处理里，时间是一个非常重要的概念，因为大部分流式应用的操作都是基于时间窗口的。

例如，流式应用可能会计算股价的5 分钟移动平均数。如果生产者因为网络问题离线了2小时，然后带着2 小时的数据重新连线，我们需要知道该如何处理这些数据。这些数据大部分都已经超过了5 分钟，而且没有参与之前的计算。

流式处理系统一般包含如下几个时间概念。

**1）****事件时间**

事件时间是指所追踪事件的发生时间和记录的创建时间。例如，度量的获取时间、商店里商品的出售时间、网站用户访问网页的时间，等等。在处理数据流肘，事件时间是很重要的。

**2）日志追加时间**

日志追加时间是指事件保存到broker 的时间。这个时间戳一般与流式处理没有太大关系，因为用户一般只对事件的发生时间感兴趣。例如，如果要计算每天生产了多少台设备，就需要计算在那一天实际生产的设备数量，尽管这些事件有可能因为网络问题到了第二天才进入Kafka 。不过，如果真实的事件时间没有被记录下来，那么就可以使用日志追加时间，在记录创建之后，这个时间就不会发生改变。

**3）处理时间**

处理时间是指应用程序在收到事件之后要对其进行处理的时间。这个时间可以是在事件发生之后的几毫秒、几小时或几天。同一个事件可能会被分配不同的时间戳，这取决于应用程序何时读取这个事件。如果应用程序使用了两个线程来读取同一个事件，这个时间戳也会不一样。所以这个时间戳非常不可靠，应该避免使用它。

### **3.2、状态** ###

如果只是单独处理每一个事件，那么流式处理就很简单。例如，如果想从Kafka 读取电商购物交易事件流，找出金额超过10 000元的交易，并将结果通过邮件发送给销售人员，那么可以使用Kafka 消费者客户端，几行代码就可以搞定。

如果操作里包含了多个事件，流式处理就会变得很有意思，比如根据类型计算事件的数量、移动平均数、合并两个流以便生成更丰富的信息流。在这些情况下，光处理单个事件是不够的，需要跟踪更多的信息，比如这个小时内看到的每种类型事件的个数、需要合并的事件、将每种类型的事件值相加等等。事件与事件之间的信息被称为状态。

这些状态一般被保存在应用程序的本地变量里。流式处理包含以下几种类型的状态。

**1）本地状态或内部状态**

这种状态只能被单个应用程序实例访问，它们一般使用内嵌在应用程序里的数据库进行维护和管理。本地状态的优势在于它的速度，不足之处在于它受到内存大小的限制。所以，流式处理的很多设计模式都将数据拆分到多个子流，这样就可以使用有限的本地状态来处理它们。

**2）****外部状态**

这种状态使用外部的数据存储来维护， 一般使用NoSQL 系统，比如HDFS 。使用外部存储的优势在于，它没有大小的限制，而且可以被应用程序的多个实例访问，甚至被不同的应用程序访问。不足之处在于，引人额外的系统会造成更大的延迟和复杂性。大部分流式处理应用尽量避免使用外部存储，或者将信息缓存在本地，减少与外部存储发生交互，以此来降低延迟。

### **3.3、流和表区别** ###

大家都熟悉数据库表，表就是记录的集合，每个表都有一个主键，并包含了一系列由schema 定义的属性。表的记录是可变的（可以在表上面执行更新和删除操作）。我们可以通过查询表数据获知某一时刻的数据状态。例如，通过查询客户信息这个表，就可以获取所有客户的联系信息。如果表被设计成不包含历史信息，那么就找不到客户过去的联系信息了。

在将表与流进行对比时，可以这么想：流包含了变更一一流是一系列事件，每个事件就是一个变更。表包含了当前的状态，是多个变更所产生的结果。

为了将表转化成流，需要捕捉到在表上所发生的变更，将“ insert ”、“ update ”和“ delete ”事件保存到流里。大部分数据库提供了用于捕捉变更的“ Change Data Capture" (CDC ）解决方案， Kafka 连接器将这些变更发送到Kafka ，用于后续的流式处理。

假设有一个商店，某零售活动可以使用一个事件流来表示：

“红色、蓝色和绿色鞋子到货”

“蓝色鞋子卖出”

“红色鞋子卖出”

“蓝色鞋子退货”

“绿色鞋子卖出”

如果想知道现在仓库里还有哪些库存，或者到目前为止赚了多少钱，可以用表。如果想知道鞋店的繁忙程度，可以查看整个事件流，会发现总共发生了5 个交易，还可以查出为什么蓝色鞋子被退货。

## **4、流式处理的常见场景** ##

现在很多公司每天都会产生数以TB级的大数据，如何对这些数据进行挖掘，分析成了很重要的课题。比如：

**电子商务**：需要处理并且挖掘用户行为产生的数据，产生推荐，从而带来更多的流量和收益。最理想的推荐就是根据兴趣推荐给用户本来不需要的东西！而每天处理海量的用户数据，需要一个低延时高可靠的实时流式分布式计算系统。

**在线订购**：假设客户向一个大型的连锁酒店预订了一个房间，连锁酒店的每一个系统在预订结束之后的几秒钟或者几分钟之内都能发出通知，包括客服中心、酒店、发送确认邮件的系统、网站等。有的酒店可能还希望客服中心能够立即获知用户在这家连锁酒店的历史入住数据，前台能够知道他是一个忠实的客户，从而提供更高级别的服务。如果使用流式处理应用来构建这些系统，就可以实现几近实时的接收和处理这些事件，从而带来更好的用户体验。

**新闻聚合**：新闻时效性非常重要，如果在一个重大事情发生后能够实时的推荐给用户，那么肯定能增大用户粘性，带来可观的流量。

**社交网站**：大家每天都会去社交网站是为了看看现在发生了什么，周围人在做什么。流式计算可以把用户关注的热点聚合，实时反馈给用户，从而达到一个圈子的聚合效果。

**交通监管**：每个城市的交通监管部门每天都要产生海量的视频数据，这些视频数据也是以流的形式源源不断的输系统中。实时流式计算系统需要以最快的速度来处理这些数据。

**数据挖掘和机器学习**：它们实际上是互联网公司内部使用的系统，主要为线上服务提供数据支撑。它们可以说是互联网公司的最核心的平台之一。系统的效率是挖掘的关键，理想条件下就是每天产生的海量数据都能得到有效处理，对于原来的数据进行全量更新。

**大型集群的监控**：自动化运维很重要，集群监控的实时预警机制也非常重要，而流式系统对于日志的实时处理，往往是监控系统的关键。

**物联网**：物联网包含很多东西。流式处理在传感器和设备上应用，最为常见的是用于预测何时该进行设备维护。这个与应用监控有点相似，不过这次是应用在硬件上，而且应用在很多不同的行业一一制造业、通信（识别故障基站）、有线电视（在用户投诉之前识别出故障机顶盒）等。每一种场景都有自己的特点，不过目标是一样的处理大量来自设备的事件，并识别出一些模式，这些模式预示着某些设备需要进行维护，比如交换机数据包的下降、生产过程中需要更大的力气来拧紧螺丝， 或者用户频繁重启有线电视的机顶盒。

## **5、Kafka流式处理实战** ##

参见模块kafka-stream下

### **5.1、maven依赖** ###

> <**dependency**>  
>     <**groupId**>org.apache.kafka</**groupId**>  
>     <**artifactId**>kafka-streams</**artifactId**>  
>     <**version**>2.1.0</**version**>  
> </**dependency**>

### **5.2、生产者代码** ###

public class Producer {
        //kafka生产者对象
        private static KafkaProducer<String,String> producer = null;
        public static void main(String[] args) {
            /*发送配置的实例   数据库 JDBC 连接*/
            Properties properties = new Properties();
            /*broker的地址清单*/
            properties.put("bootstrap.servers","127.0.0.1:9092");
            /*key的序列化器*/
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            /*value的序列化器*/
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            /*消息生产者*/
            producer = new KafkaProducer<String, String>(properties);
            //商城（各色商品）
            String[] goods = {"iphone","huawei","mi","oppo","vivo"};
            Random r = new Random();//随机数
            Random r1 = new Random();//随机数
            try {//业务
                long startTime = System.currentTimeMillis();//开始时间
                /*待发送的消息实例*/
                ProducerRecord<String,String> record;
                //生成高并发场景下的请求(循环一万次)
                for(int i=1;i<10000;i++){
                    int goodscount = r.nextInt(10);//随机生成一次购买商品的数量
                    StringBuilder sb = new StringBuilder("");//商品列表
                    if (goodscount ==0) continue; //避免生成value是空的
                    for(int j=0;j<goodscount;j++){
                        //根据商品的数量，生成随机的商品信息，每件商品使用 空格分隔，例如：3个iphone huawei mi
                        sb.append(goods[r1.nextInt(goods.length)]).append(" ");
                    }
                    try {
                        record = new ProducerRecord<String,String>("phone","sell",sb.toString());
                        producer.send(record);        /*发送消息--发送后不管*/
                        System.out.println("用户请求的商品："+sb.toString());
                        Thread.sleep(2); //1秒钟发送500条(不考虑往Kafka中送入的耗时) 20多秒可以发送完
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                long end = System.currentTimeMillis();//结束时间
                float seconds = (end - startTime) / 1000F;
                System.out.println("生产者数据消耗时间:"+Float.toString(seconds) + " seconds.");
            } finally {
                producer.close();
            }
        }
    }

### **5.3、消费者代码** ###

public class Consumer {
        public static void main(String[] args) {
            //TODO 消费者三个属性必须指定(broker地址清单、key和value的反序列化器)
            Properties properties = new Properties();
            properties.put("bootstrap.servers","127.0.0.1:9092");
            properties.put("key.deserializer", StringDeserializer.class);
            properties.put("value.deserializer", LongDeserializer.class);
            //TODO 群组并非完全必须
            properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test1");
            KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);
            try {
                //TODO 消费者订阅主题（可以多个）
                consumer.subscribe(Collections.singletonList("phone_cout"));
                while(true){
                    //TODO 拉取（新版本）
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                    for(ConsumerRecord<String, String> record:records){
                        //提交偏移量(提交越频繁，性能越差)
                        System.out.println(String.format("topic:%s,分区：%d,偏移量：%d," + "key:%s,value:%s",
                                record.topic(),record.partition(),
                                record.offset(),record.key(),record.value()));
                        //do my work(业务异常，可能进行重试  偏移量，写入主题 异常主题)
                        //打包任务投入线程池
                    }
                    //提交偏移量
                }
                //通过另外一个线程 consumer. wakeup()
            } finally {
                consumer.close();
            }
        }
    }

### **5.4、流处理代码** ###

public class StreamDeal {
        public static void main(String[] args) throws  Exception {
            //TODO 流处理三个属性必须指定(broker地址清单、key和value的序列化器)
            Properties properties = new Properties();
            properties.put("bootstrap.servers","127.0.0.1:9092");
            properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"phone_count");
            properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            // 设计模式(建造者模式)
            StreamsBuilder builder = new StreamsBuilder();
            // 使用流进行统计，(oppo 15  huawei 80  )  类似于数据表(结果)
            KStream<String,String> countStream  = builder.stream("phone");
            // 函数编程
            KTable<String,Long> wordscount = countStream
                    .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split(" ")))
                    .groupBy((key,word) ->word) //分组
                    .count(Materialized.<String,Long, KeyValueStore<Bytes, byte[]>>as("counts"));//计数
            // 相当于 流处理的结果写入到另外一个主题
            wordscount.toStream().to("phone_cout", Produced.with(Serdes.String(),Serdes.Long()));
            //这里才是定义一个KafkaStreams的对象
            KafkaStreams streams = new KafkaStreams(builder.build(),properties);
            //启动Stream引擎
            streams.start();
            Thread.sleep(5000L); //休眠的时间
            builder.build();
            //将Topic中的流中的数据小写，同时，转换成数组，最后变成List
        }
    }