03.flume

1 介绍

1.1 定义

01.发展
    Flume是Cloudera提供的一个高可用、高可靠、分布式的海量日志采集、聚合和传输的系统
    Flume最初由Cloudera开发,后来贡献给了Apache软件基金会,成为Apache的顶级项目
    Flume主要用于从各种不同的数据源高效地收集、聚合和移动大量日志数据到一个集中式的数据存储系统

02.说明
    在大数据系统中,日志数据的收集是一个非常重要的环节
    传统的日志收集方式往往效率低下,难以应对海量数据的场景
    Flume的出现很好地解决了这个问题,它可以实时地从各种数据源采集数据,并将数据传输到HDFS、HBase、Kafka等存储系统

03.设计
    Flume采用了一种简单而灵活的架构,基于流式数据流的概念
    数据在Flume中以Event的形式流转,从Source经过Channel最终到达Sink
    这种设计使得Flume具有很好的扩展性和可靠性

1.2 概念:核心组件

01.核心概念
    a.Event
        Flume数据传输的基本单元,由Header和Body两部分组成
        Header是key-value结构的元数据,Body是字节数组形式的数据内容
        Event在整个数据流中保持不变,直到被Sink消费
    b.Agent
        Flume的一个独立进程,包含Source、Channel和Sink三个核心组件
        一个Agent就是一个JVM进程,可以独立运行
        多个Agent可以级联组成复杂的数据流拓扑
    c.Source
        数据源,负责接收数据并将数据封装成Event,然后发送到Channel
        支持多种类型的数据源,如文件、网络、Kafka等
        可以配置多个Source同时工作
    d.Channel
        通道,位于Source和Sink之间,用于缓存Event,起到缓冲和容错的作用
        支持Memory Channel和File Channel两种主要类型
        Channel的容量和性能直接影响整个数据流的吞吐量
    e.Sink
        目标端,负责从Channel中取出Event并将数据发送到目的地
        支持多种目标系统,如HDFS、HBase、Kafka、Elasticsearch等
        可以配置多个Sink实现负载均衡和故障转移
    f.Interceptor
        拦截器,用于在Source和Channel之间对Event进行处理和过滤
        可以添加、修改或删除Event的Header信息
        支持链式配置多个Interceptor
    g.Selector
        选择器,用于将一个Source的数据分发到多个Channel
        支持Replicating和Multiplexing两种模式
        实现数据的多路复用和分流

02.数据流向
    a.基本流程
        数据源→Source→Interceptor→Selector→Channel→Sink→目标系统
        这个流程是单向的,数据只能从Source流向Sink
        通过组合多个Agent,可以构建复杂的数据流拓扑
    b.事务保证
        Source到Channel之间有一个事务
        Channel到Sink之间有另一个事务
        两个事务独立工作,保证数据的可靠传输
    c.批量处理
        数据不是一条一条传输的,而是批量传输
        批量大小可以配置,影响吞吐量和延迟
        批量处理提高了系统的整体性能

03.与其他系统对比
    a.Flume vs Kafka
        Flume专注于日志采集,而Kafka更偏向于消息队列
        Flume可以与Kafka配合使用,Flume负责采集,Kafka负责缓存和分发
        Kafka的吞吐量更高,但Flume的可靠性保证更强
    b.Flume vs Logstash
        Logstash是ELK栈的一部分,主要用于日志处理和转换
        Flume更轻量级,专注于数据传输
        Logstash的数据处理能力更强,但资源消耗也更大
    c.Flume vs 脚本采集
        相比于直接写脚本采集日志,Flume提供了更好的可靠性和扩展性
        Flume内置了事务机制和故障恢复能力
        Flume的配置化方式降低了开发和维护成本

1.3 优缺点

01.优点
    a.高可靠性
        支持事务机制,保证数据不丢失
        Channel可以配置为持久化模式,即使进程崩溃也能恢复数据
        Source和Sink都支持重试机制
    b.高可用性
        支持故障转移和负载均衡,可以构建高可用的数据采集系统
        多个Agent可以级联,提供冗余备份
        Sink Group可以实现自动故障切换
    c.可扩展性强
        采用模块化设计,可以方便地扩展Source、Channel和Sink
        支持自定义组件开发
        可以通过增加Agent实例实现水平扩展
    d.配置灵活
        通过配置文件即可实现复杂的数据流拓扑,无需编写代码
        支持动态配置更新
        配置语法简单清晰
    e.支持多种数据源
        可以从文件、网络、Kafka、JMS等多种数据源采集数据
        内置了常用的Source类型
        支持自定义Source开发
    f.支持多种目标
        可以将数据发送到HDFS、HBase、Kafka、Elasticsearch等多种存储系统
        内置了常用的Sink类型
        支持自定义Sink开发
    g.内置拦截器
        提供了时间戳、主机名、正则过滤等多种拦截器,方便数据处理
        支持链式配置多个拦截器
        可以自定义拦截器实现复杂的数据处理逻辑

02.缺点
    a.数据处理能力有限
        不支持数据的实时处理和转换,只能做简单的过滤和路由
        对于复杂的数据处理逻辑,需要自定义开发组件
        不支持复杂的数据聚合和计算
    b.资源消耗
        内存消耗较大,特别是使用Memory Channel时
        File Channel会占用大量磁盘空间
        JVM进程需要合理配置堆内存
    c.监控管理
        监控和管理相对简单,缺少完善的Web UI
        需要依赖第三方监控工具
        日志分析相对困难
    d.小文件处理
        对于小文件的采集效率不高
        会产生大量的小文件到HDFS
        需要配合文件合并策略使用
    e.配置复杂性
        配置相对复杂,需要熟悉各种组件的参数
        调试和排错比较困难
        需要一定的学习成本
    f.性能瓶颈
        单个Agent的吞吐量有限
        Channel容量限制可能成为瓶颈
        需要合理规划Agent的部署架构

1.4 使用场景

01.日志采集
    a.Web服务器日志
        从Web服务器、应用服务器采集日志文件
        将分散在各个服务器上的日志统一收集到HDFS或Kafka
        支持实时采集和离线采集两种模式
    b.应用日志
        采集应用程序产生的日志文件
        支持多种日志格式,如log4j、logback等
        可以实时监控日志文件的变化

02.数据同步
    a.数据库同步
        将数据从一个系统同步到另一个系统
        例如将MySQL的binlog同步到HDFS
        或者将数据从Kafka同步到HBase
    b.跨数据中心同步
        在不同数据中心之间同步数据
        保证数据的一致性和可用性
        支持断点续传和数据校验

03.数据聚合
    a.多源聚合
        将多个数据源的数据聚合到一个目标系统
        例如将多个应用的日志聚合到一个Kafka Topic
        或者将多个数据中心的数据聚合到一个HDFS集群
    b.数据汇总
        将分散的数据汇总到中心节点
        减少下游系统的连接数
        提高数据处理效率

04.实时数据流
    a.流式数据源
        作为实时数据流的数据源
        将采集到的数据发送到Kafka或Spark Streaming
        构建实时数据处理管道
    b.事件驱动
        采集系统事件和用户行为数据
        支持实时分析和告警
        与流式计算框架集成

05.ETL数据管道
    a.数据提取
        作为ETL流程的一部分
        从数据源提取数据,经过简单转换后加载到目标系统
        配合Hive、Spark等工具进行数据分析
    b.数据清洗
        在数据传输过程中进行简单的数据清洗
        过滤无效数据和敏感信息
        标准化数据格式

1.5 架构模式

01.单层架构
    a.基本结构
        最简单的架构,一个Agent包含Source、Channel和Sink
        适用于单机环境或简单的数据采集场景
        配置简单,但可靠性和扩展性有限
    b.数据流向
        数据源→Agent(Source→Channel→Sink)→目标系统
        所有组件在一个JVM进程中运行
        适合小规模数据采集
    c.应用场景
        单机日志采集
        测试和开发环境
        数据量较小的场景

02.多层架构
    a.基本结构
        多个Agent级联,形成多层的数据流拓扑
        第一层Agent负责采集数据,第二层Agent负责聚合和转发
        提高了系统的可靠性和扩展性
    b.数据流向
        数据源→Agent1→Agent2→Agent3→目标系统
        每层Agent可以有多个实例
        支持负载均衡和故障转移
    c.应用场景
        大规模分布式日志采集
        跨数据中心数据传输
        需要数据聚合的场景

03.扇入架构
    a.基本结构
        多个Agent将数据发送到一个中心Agent
        中心Agent负责数据的聚合和统一处理
        减少了目标系统的连接数
    b.数据流向
        数据源1→Agent1→
        数据源2→Agent2→中心Agent→目标系统
        数据源3→Agent3→
        多个数据源汇聚到一个节点
    c.应用场景
        多个应用服务器的日志汇总
        分布式系统的数据聚合
        需要统一处理的场景

04.扇出架构
    a.基本结构
        一个Agent将数据分发到多个目标Agent
        实现数据的多路复用
        不同的下游系统可以独立消费数据
    b.数据流向
        数据源→Agent→Agent1→目标系统1
                    →Agent2→目标系统2
                    →Agent3→目标系统3
        一份数据发送到多个目标
    c.应用场景
        数据需要发送到多个目标系统
        实时和离线处理并存
        数据备份和容灾

05.复杂拓扑
    a.基本结构
        结合多种架构模式,形成复杂的数据流拓扑
        可以实现灵活的数据路由和处理
        满足复杂的业务需求
    b.数据流向
        根据业务需求设计数据流向
        可以包含多层、扇入、扇出等多种模式
        支持条件路由和动态分发
    c.应用场景
        大型企业级数据采集系统
        多业务线数据整合
        复杂的数据处理流程

1.6 数据流模型

01.Event模型
    a.Event结构
        Event是Flume中数据传输的基本单元
        由Header和Body两部分组成
        Header是Map<String,String>结构,存储元数据
        Body是byte[]数组,存储实际数据内容
    b.Event生命周期
        Source接收原始数据并封装成Event
        Event在Channel中缓存
        Sink从Channel取出Event并发送到目标系统
        Event在整个流程中保持不变
    c.Event处理
        Interceptor可以修改Event的Header
        Selector根据Header信息路由Event
        Channel负责Event的持久化和恢复

02.批量传输
    a.批量机制
        Flume不是一条一条传输Event,而是批量传输
        Source批量写入Channel
        Sink批量从Channel读取
        批量大小可以配置
    b.批量参数
        batchSize:每批次处理的Event数量
        transactionCapacity:事务容量,必须大于等于batchSize
        批量大小影响吞吐量和延迟
    c.批量优势
        减少了事务的开销
        提高了数据传输效率
        降低了网络通信次数

03.背压机制
    a.背压原理
        当Channel满时,Source会阻塞等待
        当Channel空时,Sink会阻塞等待
        通过背压机制实现流量控制
    b.背压效果
        防止数据丢失
        避免内存溢出
        保证系统稳定运行
    c.背压配置
        Channel容量配置
        Source和Sink的批量大小配置
        合理配置可以提高系统性能

04.数据流控制
    a.流量控制
        通过Channel容量限制数据流量
        Source和Sink的处理速度自动适配
        避免系统过载
    b.优先级控制
        不同的数据流可以设置不同的优先级
        高优先级数据优先处理
        保证重要数据的及时传输
    c.路由控制
        通过Selector实现数据路由
        支持条件路由和多路复用
        灵活的数据分发策略

1.7 可靠性机制

01.事务机制
    a.两阶段事务
        Source到Channel是一个事务
        Channel到Sink是另一个事务
        两个事务独立工作,互不影响
    b.事务流程
        Source开启事务,批量写入Event到Channel
        写入成功后提交事务,失败则回滚
        Sink开启事务,批量从Channel读取Event
        发送成功后提交事务,从Channel删除Event
    c.事务保证
        保证Event要么在Source,要么在Channel,要么在Sink
        不会出现Event丢失的情况
        支持At-Least-Once语义

02.Channel持久化
    a.Memory Channel
        数据存储在内存中,速度快但不可靠
        进程崩溃会导致数据丢失
        适合对可靠性要求不高的场景
    b.File Channel
        数据持久化到磁盘,可靠性高
        进程崩溃后可以恢复数据
        性能比Memory Channel低
    c.Kafka Channel
        使用Kafka作为Channel
        结合了高性能和高可靠性
        适合大规模数据采集

03.故障恢复
    a.Agent故障
        Agent崩溃后重启可以恢复数据
        File Channel会从checkpoint恢复
        Memory Channel的数据会丢失
    b.网络故障
        Sink支持重试机制
        可以配置重试次数和重试间隔
        网络恢复后自动继续传输
    c.目标系统故障
        Sink会缓存数据在Channel中
        等待目标系统恢复
        支持故障转移到备用系统

04.数据去重
    a.去重需求
        在某些场景下需要保证数据的唯一性
        避免重复数据影响下游处理
        需要在应用层实现去重逻辑
    b.去重方案
        使用Interceptor添加唯一ID
        在目标系统进行去重
        使用外部去重服务
    c.去重限制
        Flume本身不提供去重功能
        需要结合业务场景设计去重方案
        可能影响系统性能

05.监控告警
    a.监控指标
        Channel容量使用率
        Source和Sink的吞吐量
        Event处理延迟
        事务成功率和失败率
    b.监控方式
        JMX监控
        HTTP监控接口
        日志监控
        第三方监控工具
    c.告警策略
        Channel满告警
        Sink失败告警
        延迟过高告警
        及时发现和处理问题

1.8 事务机制

01.事务模型
    a.Put事务
        Source向Channel写入Event的事务
        包含doPut、doCommit、doRollback三个阶段
        保证批量写入的原子性
    b.Take事务
        Sink从Channel读取Event的事务
        包含doTake、doCommit、doRollback三个阶段
        保证批量读取的原子性
    c.事务隔离
        Put事务和Take事务相互独立
        互不影响,可以并发执行
        提高了系统的并发性能

02.事务流程
    a.Put事务流程
        Source调用Channel.getTransaction()获取事务对象
        调用transaction.begin()开启事务
        循环调用channel.put(event)写入Event
        调用transaction.commit()提交事务
        如果失败则调用transaction.rollback()回滚
        最后调用transaction.close()关闭事务
    b.Take事务流程
        Sink调用Channel.getTransaction()获取事务对象
        调用transaction.begin()开启事务
        循环调用channel.take()读取Event
        将Event发送到目标系统
        发送成功后调用transaction.commit()提交事务
        如果失败则调用transaction.rollback()回滚
        最后调用transaction.close()关闭事务
    c.事务状态
        事务有Started、Committed、RolledBack、Closed四种状态
        状态转换必须按照规定的顺序进行
        违反状态转换规则会抛出异常

03.事务实现
    a.Memory Channel事务
        使用内存队列实现
        Put事务将Event放入putList
        Commit时将putList的Event移到queue
        Rollback时清空putList
        Take事务从queue取出Event放入takeList
        Commit时清空takeList
        Rollback时将takeList的Event放回queue
    b.File Channel事务
        使用WAL(Write-Ahead Log)机制
        所有操作先写入日志文件
        Commit时将日志标记为已提交
        Rollback时忽略未提交的日志
        通过checkpoint定期清理日志
    c.Kafka Channel事务
        利用Kafka的事务机制
        Producer使用事务写入Kafka
        Consumer使用事务读取Kafka
        保证端到端的事务性

04.事务配置
    a.transactionCapacity
        事务容量,单个事务可以处理的最大Event数量
        必须大于等于batchSize
        影响内存使用和性能
    b.batchSize
        批量大小,每次处理的Event数量
        Source和Sink都有batchSize配置
        需要根据实际情况调整
    c.超时配置
        事务超时时间
        防止事务长时间占用资源
        超时后自动回滚

05.事务性能
    a.批量优化
        增大batchSize可以提高吞吐量
        但会增加延迟和内存使用
        需要权衡吞吐量和延迟
    b.并发优化
        多个Source和Sink可以并发执行
        增加并发度可以提高整体性能
        需要注意资源竞争
    c.Channel优化
        选择合适的Channel类型
        Memory Channel性能高但不可靠
        File Channel可靠但性能低
        Kafka Channel兼顾性能和可靠性

2 核心组件

2.1 汇总:6个

00.总结
    a.Source
        数据源组件,负责接收数据并封装成Event
        支持多种类型:Avro、Exec、Spooling Directory、Taildir、Kafka等
        可以配置多个Source同时工作
    b.Channel
        通道组件,用于缓存Event
        支持Memory、File、Kafka三种主要类型
        提供事务支持和持久化能力
    c.Sink
        目标组件,负责将Event发送到目标系统
        支持多种目标:HDFS、HBase、Kafka、Elasticsearch等
        可以配置Sink Group实现负载均衡和故障转移
    d.Interceptor
        拦截器组件,用于处理和过滤Event
        可以添加、修改、删除Event的Header信息
        支持链式配置多个Interceptor
    e.Selector
        选择器组件,用于将Source的数据分发到多个Channel
        支持Replicating和Multiplexing两种模式
        实现数据的路由和分流
    f.SinkProcessor
        Sink处理器,用于管理多个Sink
        支持Default、Failover、Load Balance三种类型
        实现Sink的负载均衡和故障转移

2.2 Source

01.基本概念
    a.定义
        Source是Flume Agent的数据入口
        负责接收外部数据并封装成Event
        将Event发送到一个或多个Channel
    b.工作原理
        Source监听数据源,接收数据
        将原始数据封装成Event对象
        通过ChannelProcessor将Event写入Channel
        支持批量写入提高效率
    c.配置方式
        在Agent配置文件中定义Source
        指定Source类型和相关参数
        配置Channel Selector和Interceptor

02.Source类型
    a.Avro Source
        监听Avro端口接收数据
        用于Agent之间的数据传输
        支持高吞吐量和压缩
    b.Exec Source
        执行Unix命令获取数据
        常用于tail -F命令监控日志文件
        命令退出后Source也会停止
    c.Spooling Directory Source
        监控指定目录,读取新文件
        文件读取完成后会重命名或删除
        适合批量文件采集
    d.Taildir Source
        监控多个文件,支持断点续传
        记录每个文件的读取位置
        推荐用于日志文件采集
    e.Kafka Source
        从Kafka Topic消费数据
        支持Kafka消费者组
        可以配置多个分区并行消费
    f.HTTP Source
        提供HTTP接口接收数据
        支持JSON格式数据
        适合应用程序直接推送数据

03.Source配置
    a.基本配置
        type:指定Source类型
        channels:指定关联的Channel列表
        selector.type:指定Channel Selector类型
    b.性能配置
        batchSize:批量大小
        batchTimeout:批量超时时间
        threads:处理线程数
    c.可靠性配置
        backoff:失败后的退避策略
        maxBackoff:最大退避时间
        重试机制配置

04.Source最佳实践
    a.选择合适的Source类型
        根据数据源特点选择Source
        日志文件推荐使用Taildir Source
        Agent间传输使用Avro Source
    b.配置合理的批量大小
        批量大小影响吞吐量和延迟
        需要根据实际情况调整
        一般设置为100-1000
    c.监控Source性能
        监控Event接收速率
        监控Channel写入成功率
        及时发现和处理问题

2.3 Channel

01.基本概念
    a.定义
        Channel是Source和Sink之间的缓冲区
        临时存储Event,提供事务支持
        实现数据的异步传输和解耦
    b.工作原理
        Source通过Put事务写入Event
        Sink通过Take事务读取Event
        提供FIFO队列语义
        支持并发读写
    c.配置方式
        在Agent配置文件中定义Channel
        指定Channel类型和容量
        配置事务参数

02.Channel类型
    a.Memory Channel
        数据存储在内存中
        性能高,吞吐量大
        进程崩溃会导致数据丢失
        适合对可靠性要求不高的场景
    b.File Channel
        数据持久化到磁盘
        可靠性高,支持故障恢复
        性能比Memory Channel低
        适合对可靠性要求高的场景
    c.Kafka Channel
        使用Kafka作为Channel
        兼顾性能和可靠性
        支持大容量缓存
        适合大规模数据采集
    d.Spillable Memory Channel
        内存和磁盘混合使用
        内存满后溢出到磁盘
        兼顾性能和可靠性

03.Channel配置
    a.容量配置
        capacity:Channel最大容量
        transactionCapacity:事务容量
        需要根据数据量合理配置
    b.Memory Channel配置
        capacity:队列容量,默认100
        transactionCapacity:事务容量,默认100
        keep-alive:线程存活时间
    c.File Channel配置
        checkpointDir:checkpoint目录
        dataDirs:数据目录,可以配置多个
        maxFileSize:单个文件最大大小
        checkpointInterval:checkpoint间隔
    d.Kafka Channel配置
        kafka.bootstrap.servers:Kafka集群地址
        kafka.topic:Kafka Topic名称
        parseAsFlumeEvent:是否解析为Flume Event

04.Channel选择
    a.性能优先
        选择Memory Channel
        配置较大的容量
        注意内存使用情况
    b.可靠性优先
        选择File Channel
        配置多个数据目录
        定期清理checkpoint
    c.平衡性能和可靠性
        选择Kafka Channel
        利用Kafka的高性能和高可靠性
        需要部署Kafka集群

2.4 Sink

01.基本概念
    a.定义
        Sink是Flume Agent的数据出口
        负责从Channel读取Event
        将Event发送到目标系统
    b.工作原理
        Sink通过Take事务从Channel读取Event
        将Event转换为目标系统的格式
        批量发送到目标系统
        发送成功后提交事务
    c.配置方式
        在Agent配置文件中定义Sink
        指定Sink类型和目标系统参数
        配置批量大小和重试策略

02.Sink类型
    a.HDFS Sink
        将数据写入HDFS
        支持文件滚动和压缩
        可以按时间或大小分割文件
    b.Kafka Sink
        将数据发送到Kafka
        支持分区和压缩
        可以配置多个Broker
    c.Avro Sink
        通过Avro协议发送数据
        用于Agent之间的数据传输
        支持负载均衡和故障转移
    d.HBase Sink
        将数据写入HBase
        支持批量写入
        可以配置列族和列
    e.Elasticsearch Sink
        将数据写入Elasticsearch
        支持批量索引
        可以配置索引名称和类型
    f.Logger Sink
        将Event输出到日志
        用于调试和测试
        不适合生产环境

03.Sink配置
    a.基本配置
        type:指定Sink类型
        channel:指定关联的Channel
        目标系统连接参数
    b.批量配置
        batchSize:批量大小
        批量发送提高效率
        需要根据目标系统调整
    c.重试配置
        maxAttempts:最大重试次数
        backoff:失败后的退避策略
        保证数据可靠传输

04.Sink最佳实践
    a.选择合适的Sink类型
        根据目标系统选择Sink
        HDFS适合离线存储
        Kafka适合实时流处理
    b.配置合理的批量大小
        批量大小影响吞吐量
        HDFS Sink建议1000-10000
        HBase Sink建议100-1000
    c.配置Sink Group
        使用多个Sink提高可用性
        配置负载均衡或故障转移
        监控Sink运行状态

2.5 Interceptor

01.基本概念
    a.定义
        Interceptor是Event处理器
        在Source和Channel之间工作
        可以修改、过滤、丢弃Event
    b.工作原理
        Source将Event传递给Interceptor链
        每个Interceptor依次处理Event
        处理后的Event写入Channel
        支持链式配置多个Interceptor
    c.配置方式
        在Source配置中添加interceptors
        指定Interceptor类型和参数
        按顺序配置多个Interceptor

02.内置Interceptor
    a.Timestamp Interceptor
        在Event Header中添加时间戳
        时间戳key为timestamp
        值为当前时间的毫秒数
        用于记录Event产生时间
    b.Host Interceptor
        在Event Header中添加主机名或IP
        key为host
        值为主机名或IP地址
        用于标识数据来源
    c.Static Interceptor
        在Event Header中添加静态键值对
        可以添加多个键值对
        用于标记数据类型或来源
    d.Regex Filtering Interceptor
        根据正则表达式过滤Event
        匹配的Event保留或丢弃
        用于过滤无效数据
    e.Regex Extractor Interceptor
        从Event Body中提取信息到Header
        使用正则表达式提取
        用于数据解析和路由
    f.Morphline Interceptor
        使用Morphline处理Event
        支持复杂的数据转换
        功能强大但配置复杂

03.Interceptor配置
    a.基本配置
        在Source中配置interceptors列表
        为每个Interceptor指定名称
        配置Interceptor类型和参数
    b.Timestamp Interceptor配置
        type:timestamp
        preserveExisting:是否保留已有时间戳
        默认为false,会覆盖已有时间戳
    c.Host Interceptor配置
        type:host
        useIP:使用IP还是主机名
        preserveExisting:是否保留已有host
    d.Regex Filtering Interceptor配置
        type:regex_filter
        regex:正则表达式
        excludeEvents:true表示排除匹配的Event
    e.链式配置
        interceptors:i1 i2 i3
        按顺序执行
        前一个的输出是后一个的输入

04.自定义Interceptor
    a.实现接口
        实现org.apache.flume.interceptor.Interceptor接口
        实现initialize、intercept、close方法
        intercept方法处理单个Event
        intercept(List<Event>)方法处理批量Event
    b.实现Builder
        实现Interceptor.Builder接口
        实现build方法创建Interceptor实例
        实现configure方法读取配置参数
    c.配置使用
        将自定义Interceptor打包成jar
        放到Flume的lib目录
        在配置文件中指定完整类名
    d.最佳实践
        Interceptor应该是无状态的
        处理逻辑应该简单高效
        避免在Interceptor中做复杂计算
        异常处理要完善

05.Interceptor应用场景
    a.数据标记
        添加时间戳、主机名等元数据
        标记数据来源和类型
        方便后续处理和分析
    b.数据过滤
        过滤无效或敏感数据
        根据规则丢弃不需要的Event
        减少下游系统的处理压力
    c.数据路由
        提取Header信息用于路由
        根据内容将数据分发到不同Channel
        实现数据的分类处理
    d.数据转换
        简单的数据格式转换
        字段提取和重组
        为下游系统准备数据

2.6 Selector

01.基本概念
    a.定义
        Selector是Channel选择器
        决定Event发送到哪些Channel
        实现数据的路由和分发
    b.工作原理
        Source通过ChannelProcessor处理Event
        ChannelProcessor调用Selector选择Channel
        将Event写入选中的Channel
        支持一对多的数据分发
    c.配置方式
        在Source配置中指定selector.type
        配置Selector的参数
        关联多个Channel

02.Selector类型
    a.Replicating Selector
        复制模式,默认类型
        将Event复制到所有Channel
        实现数据的多路复用
        所有Channel都会收到相同的数据
    b.Multiplexing Selector
        多路复用模式
        根据Event Header的值选择Channel
        实现数据的条件路由
        不同的数据发送到不同的Channel
    c.Custom Selector
        自定义Selector
        实现复杂的路由逻辑
        需要实现ChannelSelector接口

03.Replicating Selector
    a.基本配置
        selector.type:replicating
        不需要其他配置
        默认行为
    b.Optional Channel
        可以配置可选Channel
        selector.optional:c3 c4
        可选Channel写入失败不影响必选Channel
        提高系统的容错能力
    c.应用场景
        数据需要发送到多个目标系统
        实时和离线处理并存
        数据备份和容灾

04.Multiplexing Selector
    a.基本配置
        selector.type:multiplexing
        selector.header:指定用于路由的Header key
        selector.mapping.*:配置映射规则
        selector.default:默认Channel
    b.映射规则
        根据Header值映射到Channel
        selector.mapping.value1:c1
        selector.mapping.value2:c2
        支持配置多个映射规则
    c.默认Channel
        selector.default:c3
        Header值不匹配时使用默认Channel
        如果不配置默认Channel,不匹配的Event会被丢弃
    d.应用场景
        根据数据类型路由到不同Channel
        实现数据的分类处理
        不同业务线的数据隔离

05.Selector配置示例
    a.Replicating示例
        agent.sources.r1.selector.type:replicating
        agent.sources.r1.channels:c1 c2 c3
        agent.sources.r1.selector.optional:c3
        Event会被复制到c1、c2、c3
        c3写入失败不影响c1、c2
    b.Multiplexing示例
        agent.sources.r1.selector.type:multiplexing
        agent.sources.r1.selector.header:type
        agent.sources.r1.selector.mapping.login:c1
        agent.sources.r1.selector.mapping.logout:c2
        agent.sources.r1.selector.default:c3
        type为login的Event发送到c1
        type为logout的Event发送到c2
        其他Event发送到c3
    c.结合Interceptor
        使用Interceptor添加Header
        Selector根据Header路由
        实现灵活的数据分发

06.自定义Selector
    a.实现接口
        实现org.apache.flume.ChannelSelector接口
        实现getRequiredChannels方法返回必选Channel
        实现getOptionalChannels方法返回可选Channel
    b.实现Builder
        实现ChannelSelector.Builder接口
        实现build方法创建Selector实例
        实现configure方法读取配置参数
    c.配置使用
        将自定义Selector打包成jar
        放到Flume的lib目录
        在配置文件中指定完整类名
    d.最佳实践
        Selector应该是无状态的
        路由逻辑应该简单高效
        避免复杂的计算和IO操作
        异常处理要完善

2.7 SinkProcessor

01.基本概念
    a.定义
        SinkProcessor是Sink处理器
        管理一组Sink的运行
        实现负载均衡和故障转移
    b.工作原理
        将多个Sink组织成Sink Group
        SinkProcessor决定使用哪个Sink
        处理Sink的失败和恢复
        提高系统的可用性和性能
    c.配置方式
        定义Sink Group
        指定SinkProcessor类型
        配置Sink Group包含的Sink

02.SinkProcessor类型
    a.Default Sink Processor
        默认处理器,单个Sink
        不需要配置Sink Group
        适合简单场景
    b.Failover Sink Processor
        故障转移处理器
        配置Sink的优先级
        主Sink失败后切换到备Sink
        实现高可用
    c.Load Balancing Sink Processor
        负载均衡处理器
        支持轮询和随机两种策略
        将数据分发到多个Sink
        提高吞吐量

03.Failover Sink Processor
    a.基本配置
        processor.type:failover
        processor.priority.*:配置Sink优先级
        processor.maxpenalty:最大惩罚时间
    b.工作原理
        按优先级选择Sink
        优先级高的Sink优先使用
        Sink失败后进入惩罚期
        惩罚期内不使用该Sink
        惩罚期结束后重新尝试
    c.优先级配置
        processor.priority.sink1:10
        processor.priority.sink2:5
        数值越大优先级越高
        优先使用sink1,失败后使用sink2
    d.应用场景
        主备模式的数据传输
        保证数据传输的高可用性
        目标系统有主备节点

04.Load Balancing Sink Processor
    a.基本配置
        processor.type:load_balance
        processor.selector:选择策略
        processor.backoff:失败后的退避策略
    b.选择策略
        round_robin:轮询策略,默认
        random:随机策略
        自定义策略
    c.工作原理
        根据策略选择Sink
        轮询策略依次使用每个Sink
        随机策略随机选择Sink
        Sink失败后进入黑名单
        定期重试黑名单中的Sink
    d.应用场景
        多个目标系统并行处理
        提高整体吞吐量
        目标系统有多个节点

05.Sink Group配置
    a.定义Sink Group
        agent.sinkgroups:g1
        agent.sinkgroups.g1.sinks:k1 k2 k3
        指定Sink Group包含的Sink
    b.配置SinkProcessor
        agent.sinkgroups.g1.processor.type:failover
        agent.sinkgroups.g1.processor.priority.k1:10
        agent.sinkgroups.g1.processor.priority.k2:5
        agent.sinkgroups.g1.processor.maxpenalty:10000
    c.Sink配置
        每个Sink独立配置
        指定Sink类型和目标系统
        配置Channel关联

06.SinkProcessor最佳实践
    a.选择合适的处理器类型
        高可用场景使用Failover
        高吞吐场景使用Load Balance
        根据业务需求选择
    b.合理配置优先级和策略
        Failover配置合理的优先级
        Load Balance选择合适的策略
        配置合理的退避时间
    c.监控Sink运行状态
        监控每个Sink的成功率
        监控Sink的切换情况
        及时发现和处理问题
    d.测试故障切换
        定期测试故障转移功能
        验证备用Sink是否正常工作
        确保高可用性

3 Source详解

3.1 汇总:7个

00.总结
    a.Avro Source
        监听Avro端口接收数据
        用于Agent之间的数据传输
        支持高吞吐量和数据压缩
    b.Exec Source
        执行Unix命令获取数据
        常用于tail命令监控日志文件
        命令退出后Source也会停止
    c.Spooling Directory Source
        监控指定目录,读取新文件
        文件读取完成后会重命名
        适合批量文件采集
    d.Taildir Source
        监控多个文件,支持断点续传
        记录每个文件的读取位置
        推荐用于日志文件采集
    e.Kafka Source
        从Kafka Topic消费数据
        支持Kafka消费者组
        可以配置多个分区并行消费
    f.HTTP Source
        提供HTTP接口接收数据
        支持JSON格式数据
        适合应用程序直接推送数据
    g.自定义Source
        实现Source接口
        满足特殊的数据采集需求
        需要Java开发能力

3.2 Avro Source

01.基本概念
    a.定义
        Avro Source监听Avro端口接收数据
        使用Avro RPC协议进行数据传输
        主要用于Agent之间的数据传输
    b.工作原理
        启动Avro RPC Server监听指定端口
        接收来自Avro Client的数据
        将接收到的数据封装成Event
        批量写入Channel
    c.应用场景
        多层Agent架构中的数据传输
        跨网络的数据传输
        需要高吞吐量的场景

02.配置参数
    a.基本配置
        type:avro
        bind:绑定的主机名或IP,默认0.0.0.0
        port:监听端口,必须配置
    b.性能配置
        threads:工作线程数
        compression-type:压缩类型,none或deflate
        ssl:是否启用SSL
    c.高级配置
        ipFilter:IP过滤规则
        ipFilterRules:具体的过滤规则
        max-threads:最大线程数

03.配置示例
    a.基本配置
        agent.sources.r1.type:avro
        agent.sources.r1.bind:0.0.0.0
        agent.sources.r1.port:4141
        agent.sources.r1.channels:c1
        监听4141端口接收数据
    b.启用压缩
        agent.sources.r1.type:avro
        agent.sources.r1.bind:0.0.0.0
        agent.sources.r1.port:4141
        agent.sources.r1.compression-type:deflate
        agent.sources.r1.channels:c1
        使用deflate压缩减少网络传输
    c.多线程配置
        agent.sources.r1.type:avro
        agent.sources.r1.bind:0.0.0.0
        agent.sources.r1.port:4141
        agent.sources.r1.threads:10
        agent.sources.r1.channels:c1
        使用10个工作线程提高并发

04.最佳实践
    a.端口规划
        不同的Agent使用不同的端口
        避免端口冲突
        记录端口使用情况
    b.启用压缩
        网络带宽有限时启用压缩
        压缩可以减少网络传输量
        但会增加CPU使用率
    c.合理配置线程数
        根据数据量配置线程数
        线程过多会增加资源消耗
        一般设置为CPU核心数的1-2倍
    d.监控和告警
        监控端口是否正常监听
        监控数据接收速率
        监控连接数和线程使用情况

3.3 Exec Source

01.基本概念
    a.定义
        Exec Source执行Unix命令获取数据
        将命令的标准输出作为数据源
        常用于tail命令监控日志文件
    b.工作原理
        启动时执行指定的命令
        持续读取命令的标准输出
        将每行输出封装成一个Event
        命令退出后Source也会停止
    c.应用场景
        监控日志文件的增量内容
        执行脚本获取数据
        临时的数据采集任务

02.配置参数
    a.基本配置
        type:exec
        command:要执行的命令,必须配置
        shell:执行命令的shell,默认/bin/sh
    b.性能配置
        batchSize:批量大小,默认20
        batchTimeout:批量超时时间,默认3000ms
        restart:命令退出后是否重启,默认false
    c.高级配置
        restartThrottle:重启间隔时间,默认10000ms
        logStdErr:是否记录标准错误,默认false
        charset:字符编码,默认UTF-8

03.配置示例
    a.监控日志文件
        agent.sources.r1.type:exec
        agent.sources.r1.command:tail -F /var/log/app.log
        agent.sources.r1.channels:c1
        使用tail -F命令监控日志文件
    b.执行脚本
        agent.sources.r1.type:exec
        agent.sources.r1.command:/opt/scripts/collect.sh
        agent.sources.r1.restart:true
        agent.sources.r1.restartThrottle:5000
        agent.sources.r1.channels:c1
        执行脚本,退出后5秒重启
    c.自定义shell
        agent.sources.r1.type:exec
        agent.sources.r1.command:tail -F /var/log/app.log
        agent.sources.r1.shell:/bin/bash
        agent.sources.r1.channels:c1
        使用bash执行命令

04.注意事项
    a.数据可靠性
        命令异常退出会导致数据丢失
        不支持断点续传
        不推荐用于生产环境
    b.资源消耗
        命令会持续运行消耗资源
        需要监控命令的资源使用
        避免命令产生大量输出
    c.替代方案
        推荐使用Taildir Source代替
        Taildir Source支持断点续传
        可靠性更高

3.4 Spooling Directory Source

01.基本概念
    a.定义
        Spooling Directory Source监控指定目录
        自动读取目录中的新文件
        文件读取完成后会重命名或删除
    b.工作原理
        定期扫描指定目录
        按文件修改时间排序
        依次读取文件内容
        读取完成后添加.COMPLETED后缀
        支持递归扫描子目录
    c.应用场景
        批量文件采集
        离线日志文件采集
        定期生成的数据文件采集

02.配置参数
    a.基本配置
        type:spooldir
        spoolDir:监控的目录,必须配置
        fileSuffix:.COMPLETED,完成后的文件后缀
    b.文件过滤
        includePattern:包含的文件名正则表达式
        ignorePattern:忽略的文件名正则表达式
        fileHeader:是否添加文件名到Header
    c.性能配置
        batchSize:批量大小,默认100
        maxBackoff:最大退避时间,默认4000ms
        consumeOrder:文件消费顺序,oldest或youngest或random
    d.高级配置
        deletePolicy:删除策略,never或immediate
        decodeErrorPolicy:解码错误策略,FAIL或REPLACE或IGNORE
        deserializer:反序列化器,默认LINE

03.配置示例
    a.基本配置
        agent.sources.r1.type:spooldir
        agent.sources.r1.spoolDir:/var/log/flume/spool
        agent.sources.r1.channels:c1
        监控/var/log/flume/spool目录
    b.文件过滤
        agent.sources.r1.type:spooldir
        agent.sources.r1.spoolDir:/var/log/flume/spool
        agent.sources.r1.includePattern:^.*\\.log$
        agent.sources.r1.ignorePattern:^.*\\.tmp$
        agent.sources.r1.channels:c1
        只处理.log文件,忽略.tmp文件
    c.添加文件名
        agent.sources.r1.type:spooldir
        agent.sources.r1.spoolDir:/var/log/flume/spool
        agent.sources.r1.fileHeader:true
        agent.sources.r1.fileHeaderKey:file
        agent.sources.r1.channels:c1
        在Header中添加文件名

04.最佳实践
    a.目录规划
        使用专门的目录存放待采集文件
        不要在目录中修改文件
        避免文件名冲突
    b.文件命名
        使用有意义的文件名
        包含时间戳便于排序
        使用统一的命名规范
    c.定期清理
        定期清理.COMPLETED文件
        避免目录中文件过多
        可以使用脚本自动清理
    d.错误处理
        配置合理的错误策略
        监控错误日志
        及时处理异常文件

3.5 Taildir Source

01.基本概念
    a.定义
        Taildir Source监控多个文件
        支持断点续传和文件轮转
        是Exec Source的可靠替代方案
    b.工作原理
        使用position文件记录读取位置
        定期检查文件是否有新内容
        从上次读取位置继续读取
        支持文件重命名和轮转
        Agent重启后可以继续读取
    c.应用场景
        生产环境的日志文件采集
        需要保证数据不丢失的场景
        文件会轮转的场景
        多个日志文件同时采集

02.配置参数
    a.基本配置
        type:TAILDIR
        positionFile:位置文件路径,必须配置
        filegroups:文件组列表,必须配置
        filegroups.<group>:文件路径正则表达式
    b.性能配置
        batchSize:批量大小,默认100
        backoffSleepIncrement:退避增量,默认1000ms
        maxBackoffSleep:最大退避时间,默认5000ms
    c.高级配置
        skipToEnd:是否跳到文件末尾,默认false
        idleTimeout:空闲超时时间,默认120000ms
        writePosInterval:写入位置文件间隔,默认3000ms
        fileHeader:是否添加文件路径到Header
        fileHeaderKey:文件路径的Header key,默认file

03.配置示例
    a.单个文件
        agent.sources.r1.type:TAILDIR
        agent.sources.r1.positionFile:/var/log/flume/taildir_position.json
        agent.sources.r1.filegroups:f1
        agent.sources.r1.filegroups.f1:/var/log/app.log
        agent.sources.r1.channels:c1
        监控单个日志文件
    b.多个文件
        agent.sources.r1.type:TAILDIR
        agent.sources.r1.positionFile:/var/log/flume/taildir_position.json
        agent.sources.r1.filegroups:f1 f2
        agent.sources.r1.filegroups.f1:/var/log/app1.log
        agent.sources.r1.filegroups.f2:/var/log/app2.log
        agent.sources.r1.channels:c1
        同时监控两个日志文件
    c.使用正则表达式
        agent.sources.r1.type:TAILDIR
        agent.sources.r1.positionFile:/var/log/flume/taildir_position.json
        agent.sources.r1.filegroups:f1
        agent.sources.r1.filegroups.f1:/var/log/app.*\\.log
        agent.sources.r1.channels:c1
        监控所有匹配的日志文件
    d.添加文件路径
        agent.sources.r1.type:TAILDIR
        agent.sources.r1.positionFile:/var/log/flume/taildir_position.json
        agent.sources.r1.filegroups:f1
        agent.sources.r1.filegroups.f1:/var/log/app.log
        agent.sources.r1.fileHeader:true
        agent.sources.r1.channels:c1
        在Header中添加文件路径

04.position文件
    a.文件格式
        JSON格式存储每个文件的读取位置
        包含文件路径和inode信息
        记录当前读取的偏移量
    b.文件内容示例
        [{"inode":12345,"pos":1024,"file":"/var/log/app.log"}]
        inode:文件的inode号
        pos:当前读取位置
        file:文件路径
    c.文件管理
        定期备份position文件
        避免position文件损坏
        position文件损坏会导致重复读取

05.文件轮转处理
    a.轮转检测
        通过inode检测文件是否被轮转
        文件重命名后inode不变
        继续读取原文件直到结束
    b.新文件处理
        检测到新文件后开始读取
        从文件开头或末尾开始读取
        由skipToEnd参数控制
    c.轮转场景
        支持logrotate等工具的轮转
        支持文件重命名
        支持文件删除和重建

06.最佳实践
    a.position文件位置
        使用独立的目录存放position文件
        定期备份position文件
        避免position文件丢失
    b.文件路径配置
        使用绝对路径
        正则表达式要准确
        避免匹配到不需要的文件
    c.性能优化
        合理配置批量大小
        调整退避时间
        监控文件读取延迟
    d.监控告警
        监控position文件是否正常更新
        监控文件读取进度
        告警文件读取延迟过大

3.6 Kafka Source

01.基本概念
    a.定义
        Kafka Source从Kafka Topic消费数据
        将Kafka消息转换为Flume Event
        支持Kafka消费者组
    b.工作原理
        创建Kafka Consumer订阅Topic
        从Kafka拉取消息
        将消息封装成Event
        批量写入Channel
        提交Kafka offset
    c.应用场景
        从Kafka采集数据到HDFS
        Kafka数据同步到HBase
        Kafka数据转换和路由

02.配置参数
    a.基本配置
        type:org.apache.flume.source.kafka.KafkaSource
        kafka.bootstrap.servers:Kafka集群地址,必须配置
        kafka.topics:订阅的Topic列表,必须配置
        kafka.consumer.group.id:消费者组ID
    b.性能配置
        batchSize:批量大小,默认1000
        batchDurationMillis:批量超时时间,默认1000ms
        kafka.consumer.max.poll.records:单次拉取最大记录数
    c.高级配置
        kafka.consumer.auto.offset.reset:offset重置策略,earliest或latest
        kafka.consumer.enable.auto.commit:是否自动提交offset,默认false
        setTopicHeader:是否添加Topic到Header,默认true
        topicHeader:Topic的Header key,默认topic

03.配置示例
    a.基本配置
        agent.sources.r1.type:org.apache.flume.source.kafka.KafkaSource
        agent.sources.r1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sources.r1.kafka.topics:test-topic
        agent.sources.r1.kafka.consumer.group.id:flume-group
        agent.sources.r1.channels:c1
        从test-topic消费数据
    b.多个Topic
        agent.sources.r1.type:org.apache.flume.source.kafka.KafkaSource
        agent.sources.r1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sources.r1.kafka.topics:topic1,topic2,topic3
        agent.sources.r1.kafka.consumer.group.id:flume-group
        agent.sources.r1.channels:c1
        同时消费多个Topic
    c.使用正则表达式
        agent.sources.r1.type:org.apache.flume.source.kafka.KafkaSource
        agent.sources.r1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sources.r1.kafka.topics.regex:^log-.*
        agent.sources.r1.kafka.consumer.group.id:flume-group
        agent.sources.r1.channels:c1
        消费所有以log-开头的Topic
    d.性能优化
        agent.sources.r1.type:org.apache.flume.source.kafka.KafkaSource
        agent.sources.r1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sources.r1.kafka.topics:test-topic
        agent.sources.r1.kafka.consumer.group.id:flume-group
        agent.sources.r1.batchSize:5000
        agent.sources.r1.kafka.consumer.max.poll.records:5000
        agent.sources.r1.channels:c1
        增大批量大小提高吞吐量

04.offset管理
    a.offset提交
        Flume自动管理offset提交
        Event成功写入Channel后提交offset
        保证At-Least-Once语义
    b.offset重置
        auto.offset.reset:earliest从最早开始消费
        auto.offset.reset:latest从最新开始消费
        首次消费或offset丢失时生效
    c.offset监控
        监控消费lag
        lag过大说明消费速度慢
        需要优化性能或增加并行度

05.性能优化
    a.增大批量大小
        增大batchSize和max.poll.records
        减少网络通信次数
        提高吞吐量
    b.并行消费
        增加Topic分区数
        部署多个Flume Agent
        每个Agent消费部分分区
    c.网络优化
        Flume Agent部署在Kafka集群附近
        减少网络延迟
        使用高速网络

06.最佳实践
    a.消费者组规划
        使用有意义的消费者组ID
        不同的应用使用不同的消费者组
        避免消费者组冲突
    b.Topic规划
        合理规划Topic分区数
        分区数影响并行度
        一般设置为Flume Agent数的整数倍
    c.监控告警
        监控消费lag
        监控消费速率
        告警lag过大
    d.错误处理
        配置合理的重试策略
        记录消费失败的消息
        定期检查错误日志

3.7 HTTP Source

01.基本概念
    a.定义
        HTTP Source提供HTTP接口接收数据
        应用程序可以通过HTTP POST发送数据
        支持JSON格式数据
    b.工作原理
        启动HTTP Server监听指定端口
        接收HTTP POST请求
        解析请求Body为Event
        批量写入Channel
        返回HTTP响应
    c.应用场景
        应用程序直接推送数据
        Web服务日志采集
        移动应用数据采集
        IoT设备数据采集

02.配置参数
    a.基本配置
        type:http
        bind:绑定的主机名或IP,默认0.0.0.0
        port:监听端口,必须配置
    b.性能配置
        threads:工作线程数
        handler:请求处理器,默认JSONHandler
        handler.*:处理器的配置参数
    c.高级配置
        selector.type:Channel Selector类型
        interceptors:拦截器列表
        enableSSL:是否启用SSL
        keystore:SSL keystore文件路径
        keystorePassword:keystore密码

03.Handler类型
    a.JSONHandler
        解析JSON格式的请求Body
        支持单个Event和Event数组
        是默认的Handler
    b.BlobHandler
        将整个请求Body作为Event Body
        不做任何解析
        适合二进制数据
    c.自定义Handler
        实现HTTPSourceHandler接口
        自定义数据解析逻辑
        满足特殊需求

04.配置示例
    a.基本配置
        agent.sources.r1.type:http
        agent.sources.r1.bind:0.0.0.0
        agent.sources.r1.port:8080
        agent.sources.r1.channels:c1
        监听8080端口接收HTTP请求
    b.多线程配置
        agent.sources.r1.type:http
        agent.sources.r1.bind:0.0.0.0
        agent.sources.r1.port:8080
        agent.sources.r1.threads:10
        agent.sources.r1.channels:c1
        使用10个工作线程
    c.使用BlobHandler
        agent.sources.r1.type:http
        agent.sources.r1.bind:0.0.0.0
        agent.sources.r1.port:8080
        agent.sources.r1.handler:org.apache.flume.source.http.BlobHandler
        agent.sources.r1.channels:c1
        使用BlobHandler处理请求

05.数据格式
    a.单个Event
        POST /
        Content-Type:application/json
        [{"headers":{"key1":"value1"},"body":"event body"}]
        发送单个Event
    b.多个Event
        POST /
        Content-Type:application/json
        [{"headers":{"key1":"value1"},"body":"body1"},{"headers":{"key2":"value2"},"body":"body2"}]
        批量发送多个Event
    c.响应格式
        成功:HTTP 200 OK
        失败:HTTP 400 Bad Request或500 Internal Server Error
        响应Body包含错误信息

06.最佳实践
    a.安全性
        使用防火墙限制访问
        启用SSL加密传输
        添加认证机制
    b.性能优化
        批量发送数据
        合理配置线程数
        使用连接池
    c.错误处理
        客户端实现重试机制
        记录发送失败的数据
        监控HTTP错误率
    d.监控告警
        监控HTTP请求速率
        监控响应时间
        告警错误率过高

3.8 自定义Source

01.实现接口
    a.Source接口
        实现org.apache.flume.Source接口
        或继承AbstractSource抽象类
        实现start、stop、process等方法
    b.核心方法
        configure(Context):读取配置参数
        start():启动Source
        stop():停止Source
        process():处理数据,返回Status
    c.Status类型
        READY:成功处理数据
        BACKOFF:没有数据,需要退避
        用于控制Source的执行频率

02.实现步骤
    a.创建类
        public class MySource extends AbstractSource implements Configurable,PollableSource
        继承AbstractSource
        实现Configurable和PollableSource接口
    b.实现configure方法
        读取配置文件中的参数
        初始化Source的配置
        验证配置的有效性
    c.实现start方法
        初始化资源
        启动后台线程
        开始接收数据
    d.实现stop方法
        停止后台线程
        释放资源
        清理状态
    e.实现process方法
        获取数据
        封装成Event
        写入Channel
        返回Status

03.代码示例
    a.基本结构
        public class MySource extends AbstractSource implements Configurable,PollableSource{
            private String myProp;
            public void configure(Context context){
                myProp:context.getString("myProp","defaultValue");
            }
            public void start(){
                super.start();
            }
            public void stop(){
                super.stop();
            }
            public Status process() throws EventDeliveryException{
                return Status.READY;
            }
        }
    b.写入Channel
        Channel channel:getChannel();
        Transaction tx:channel.getTransaction();
        try{
            tx.begin();
            Event event:EventBuilder.withBody("data".getBytes());
            channel.put(event);
            tx.commit();
            return Status.READY;
        }catch(Exception e){
            tx.rollback();
            throw new EventDeliveryException(e);
        }finally{
            tx.close();
        }
    c.批量写入
        List<Event> events:new ArrayList<>();
        for(int i:0;i<batchSize;i++){
            Event event:EventBuilder.withBody("data".getBytes());
            events.add(event);
        }
        Channel channel:getChannel();
        Transaction tx:channel.getTransaction();
        try{
            tx.begin();
            for(Event event:events){
                channel.put(event);
            }
            tx.commit();
            return Status.READY;
        }catch(Exception e){
            tx.rollback();
            throw new EventDeliveryException(e);
        }finally{
            tx.close();
        }

04.打包部署
    a.打包
        使用Maven或Gradle打包成jar
        包含所有依赖
        或者使用provided scope排除Flume依赖
    b.部署
        将jar文件放到Flume的lib目录
        或者使用plugins.d目录
        重启Flume Agent
    c.配置
        agent.sources.r1.type:com.example.MySource
        agent.sources.r1.myProp:myValue
        agent.sources.r1.channels:c1
        使用完整类名指定Source类型

05.最佳实践
    a.异常处理
        捕获所有异常
        避免Source崩溃
        记录详细的错误日志
    b.资源管理
        及时释放资源
        避免资源泄漏
        使用try-finally确保资源释放
    c.性能优化
        批量处理数据
        使用合理的退避策略
        避免阻塞操作
    d.可配置性
        提供丰富的配置参数
        设置合理的默认值
        验证配置的有效性
    e.监控
        暴露JMX指标
        记录关键的运行指标
        便于监控和调试

06.常见场景
    a.数据库Source
        从数据库查询数据
        支持增量查询
        记录查询位置
    b.消息队列Source
        从消息队列消费数据
        支持多种消息队列
        保证消息不丢失
    c.自定义协议Source
        实现特定的网络协议
        解析协议数据
        转换为Flume Event
    d.定时任务Source
        定期执行任务
        获取数据
        发送到Channel

4 Channel详解

4.1 汇总:3个

00.总结
    a.Memory Channel
        数据存储在内存中
        性能高,吞吐量大
        进程崩溃会导致数据丢失
        适合对可靠性要求不高的场景
    b.File Channel
        数据持久化到磁盘
        可靠性高,支持故障恢复
        性能比Memory Channel低
        适合对可靠性要求高的场景
    c.Kafka Channel
        使用Kafka作为Channel
        兼顾性能和可靠性
        支持大容量缓存
        适合大规模数据采集

4.2 Memory Channel

01.基本概念
    a.定义
        Memory Channel将Event存储在内存队列中
        使用Java的LinkedBlockingQueue实现
        提供高性能的数据缓存
    b.工作原理
        Source将Event放入内存队列
        Sink从内存队列取出Event
        队列满时Source阻塞
        队列空时Sink阻塞
    c.应用场景
        对性能要求高的场景
        可以容忍少量数据丢失
        数据量不大的场景

02.配置参数
    a.容量配置
        type:memory
        capacity:队列最大容量,默认100
        transactionCapacity:事务容量,默认100
    b.性能配置
        keep-alive:线程存活时间,默认3秒
        byteCapacityBufferPercentage:字节容量缓冲百分比,默认20
        byteCapacity:字节容量限制,默认为JVM堆的80%
    c.高级配置
        channelFullCount:Channel满的计数器
        用于监控Channel使用情况

03.配置示例
    a.基本配置
        agent.channels.c1.type:memory
        agent.channels.c1.capacity:1000
        agent.channels.c1.transactionCapacity:100
        配置容量为1000,事务容量为100
    b.大容量配置
        agent.channels.c1.type:memory
        agent.channels.c1.capacity:10000
        agent.channels.c1.transactionCapacity:1000
        agent.channels.c1.byteCapacity:800000000
        配置大容量Channel,字节容量800MB
    c.性能优化
        agent.channels.c1.type:memory
        agent.channels.c1.capacity:5000
        agent.channels.c1.transactionCapacity:500
        agent.channels.c1.keep-alive:10
        增加keep-alive时间减少线程创建开销

04.性能特点
    a.优点
        读写速度快,延迟低
        实现简单,资源消耗小
        适合高吞吐量场景
    b.缺点
        数据不持久化,可靠性差
        进程崩溃导致数据丢失
        容量受内存限制
    c.性能指标
        吞吐量:每秒数万到数十万Event
        延迟:毫秒级
        内存使用:与容量和Event大小相关

05.最佳实践
    a.容量规划
        根据数据量和峰值流量配置容量
        capacity应该大于transactionCapacity
        预留一定的缓冲空间
    b.内存管理
        监控JVM堆内存使用
        配置合理的byteCapacity
        避免OOM错误
    c.适用场景
        测试和开发环境
        日志级别较低的场景
        配合其他可靠性机制使用
    d.监控告警
        监控Channel容量使用率
        告警Channel满
        监控Event丢失情况

4.3 File Channel

01.基本概念
    a.定义
        File Channel将Event持久化到磁盘
        使用WAL(Write-Ahead Log)机制
        提供高可靠性的数据缓存
    b.工作原理
        所有操作先写入日志文件
        使用checkpoint记录状态
        定期清理已消费的日志
        Agent重启后从checkpoint恢复
    c.应用场景
        生产环境
        不能容忍数据丢失
        需要故障恢复能力

02.配置参数
    a.基本配置
        type:file
        checkpointDir:checkpoint目录,必须配置
        dataDirs:数据目录,可以配置多个
        capacity:Channel容量,默认1000000
        transactionCapacity:事务容量,默认10000
    b.文件配置
        maxFileSize:单个日志文件最大大小,默认2146435071(约2GB)
        minimumRequiredSpace:最小可用空间,默认524288000(500MB)
        checkpointInterval:checkpoint间隔,默认30000ms
    c.性能配置
        write-timeout:写超时时间,默认3秒
        checkpoint-timeout:checkpoint超时时间,默认600秒
        use-log-replay-v1:是否使用v1版本的日志重放,默认false
        use-fast-replay:是否使用快速重放,默认false

03.配置示例
    a.基本配置
        agent.channels.c1.type:file
        agent.channels.c1.checkpointDir:/var/flume/checkpoint
        agent.channels.c1.dataDirs:/var/flume/data
        配置checkpoint和数据目录
    b.多数据目录
        agent.channels.c1.type:file
        agent.channels.c1.checkpointDir:/var/flume/checkpoint
        agent.channels.c1.dataDirs:/disk1/flume/data,/disk2/flume/data,/disk3/flume/data
        使用多个磁盘提高性能
    c.大容量配置
        agent.channels.c1.type:file
        agent.channels.c1.checkpointDir:/var/flume/checkpoint
        agent.channels.c1.dataDirs:/var/flume/data
        agent.channels.c1.capacity:10000000
        agent.channels.c1.transactionCapacity:10000
        agent.channels.c1.maxFileSize:2147483648
        配置大容量Channel

04.存储机制
    a.WAL机制
        所有操作先写入日志
        Put操作写入Put日志
        Take操作写入Take日志
        Commit操作标记事务完成
    b.Checkpoint机制
        定期将内存状态写入checkpoint文件
        记录每个Event的状态
        记录日志文件的位置
        用于故障恢复
    c.日志清理
        已消费的Event对应的日志可以删除
        通过checkpoint确定可删除的日志
        定期执行清理操作
        避免磁盘空间耗尽

05.故障恢复
    a.恢复流程
        Agent启动时读取checkpoint
        重放未完成的事务
        恢复Channel状态
        继续处理数据
    b.数据完整性
        使用校验和验证数据
        检测数据损坏
        损坏的数据会被跳过
    c.恢复时间
        取决于未完成事务的数量
        取决于日志文件的大小
        一般在几秒到几分钟

06.性能优化
    a.使用多个数据目录
        将数据目录分布在不同磁盘
        提高IO性能
        避免单个磁盘成为瓶颈
    b.使用SSD
        使用SSD存储checkpoint
        提高checkpoint性能
        减少恢复时间
    c.调整checkpoint间隔
        增大checkpoint间隔减少IO
        但会增加恢复时间
        需要权衡性能和恢复时间
    d.调整文件大小
        增大maxFileSize减少文件数量
        但会增加单个文件的管理开销
        一般使用默认值

07.最佳实践
    a.目录规划
        checkpoint和数据目录使用不同磁盘
        数据目录使用多个磁盘
        定期监控磁盘空间
    b.容量规划
        根据数据量和保留时间规划容量
        预留足够的磁盘空间
        配置minimumRequiredSpace避免磁盘满
    c.定期维护
        定期检查日志文件数量
        监控checkpoint执行情况
        清理异常的日志文件
    d.监控告警
        监控磁盘空间使用率
        监控checkpoint执行时间
        告警磁盘空间不足
        监控Channel容量使用率

4.4 Kafka Channel

01.基本概念
    a.定义
        Kafka Channel使用Kafka作为存储
        将Event写入Kafka Topic
        从Kafka Topic读取Event
    b.工作原理
        Source作为Kafka Producer写入数据
        Sink作为Kafka Consumer读取数据
        利用Kafka的高性能和高可靠性
        支持大容量数据缓存
    c.应用场景
        大规模数据采集
        需要高性能和高可靠性
        数据需要长时间保留
        多个下游系统消费数据

02.配置参数
    a.基本配置
        type:org.apache.flume.channel.kafka.KafkaChannel
        kafka.bootstrap.servers:Kafka集群地址,必须配置
        kafka.topic:Kafka Topic名称,必须配置
        kafka.consumer.group.id:消费者组ID,默认flume
    b.性能配置
        parseAsFlumeEvent:是否解析为Flume Event,默认true
        kafka.producer.acks:生产者acks配置,默认1
        kafka.producer.batch.size:批量大小,默认16384
        kafka.producer.linger.ms:延迟时间,默认0
    c.高级配置
        kafka.consumer.auto.offset.reset:offset重置策略,默认latest
        kafka.producer.compression.type:压缩类型,默认none
        kafka.topic.default.partitions:默认分区数,默认1
        kafka.topic.default.replication-factor:默认副本数,默认1

03.配置示例
    a.基本配置
        agent.channels.c1.type:org.apache.flume.channel.kafka.KafkaChannel
        agent.channels.c1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.channels.c1.kafka.topic:flume-channel
        agent.channels.c1.kafka.consumer.group.id:flume-group
        使用Kafka作为Channel
    b.高可靠性配置
        agent.channels.c1.type:org.apache.flume.channel.kafka.KafkaChannel
        agent.channels.c1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.channels.c1.kafka.topic:flume-channel
        agent.channels.c1.kafka.consumer.group.id:flume-group
        agent.channels.c1.kafka.producer.acks:all
        agent.channels.c1.kafka.producer.compression.type:snappy
        配置acks为all保证可靠性,启用压缩
    c.高性能配置
        agent.channels.c1.type:org.apache.flume.channel.kafka.KafkaChannel
        agent.channels.c1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.channels.c1.kafka.topic:flume-channel
        agent.channels.c1.kafka.consumer.group.id:flume-group
        agent.channels.c1.kafka.producer.batch.size:65536
        agent.channels.c1.kafka.producer.linger.ms:10
        增大批量大小和延迟时间提高吞吐量

04.Event格式
    a.parseAsFlumeEvent为true
        Event序列化为Flume格式
        包含Header和Body
        Sink可以正确解析Event
        适合Flume内部使用
    b.parseAsFlumeEvent为false
        Event Body直接作为Kafka消息
        Header信息丢失
        适合与其他系统集成
        其他系统可以直接消费Kafka数据

05.性能特点
    a.优点
        兼顾性能和可靠性
        支持大容量数据缓存
        数据可以长时间保留
        支持多个消费者
        利用Kafka的分布式特性
    b.缺点
        需要部署Kafka集群
        增加了系统复杂度
        延迟比Memory Channel高
        需要管理Kafka Topic
    c.性能指标
        吞吐量:每秒数万到数十万Event
        延迟:几毫秒到几十毫秒
        容量:取决于Kafka集群

06.Topic管理
    a.Topic创建
        可以自动创建Topic
        也可以手动创建Topic
        建议手动创建并配置分区数和副本数
    b.分区规划
        分区数影响并行度
        建议设置为Flume Agent数的整数倍
        过多分区会增加管理开销
    c.副本配置
        副本数影响可靠性
        建议设置为3
        副本数不能超过Broker数量
    d.数据保留
        配置数据保留时间
        根据需求设置保留策略
        定期清理过期数据

07.最佳实践
    a.Kafka集群规划
        部署高可用的Kafka集群
        配置合理的分区数和副本数
        监控Kafka集群状态
    b.性能优化
        启用压缩减少网络传输
        增大批量大小提高吞吐量
        使用多个分区提高并行度
    c.可靠性保证
        配置acks为all
        配置min.insync.replicas
        监控数据复制状态
    d.监控告警
        监控Kafka集群状态
        监控Topic的lag
        告警消费延迟过大
        监控磁盘空间使用

4.5 Channel选择策略

01.性能优先
    a.选择Memory Channel
        适合对性能要求极高的场景
        可以容忍少量数据丢失
        数据量不大
    b.配置建议
        配置较大的容量
        增大transactionCapacity
        监控内存使用
    c.适用场景
        测试和开发环境
        日志级别为DEBUG或INFO
        实时性要求高的场景

02.可靠性优先
    a.选择File Channel
        适合不能容忍数据丢失的场景
        对性能要求不是特别高
        单机部署
    b.配置建议
        使用多个数据目录
        配置合理的checkpoint间隔
        定期清理日志文件
    c.适用场景
        生产环境
        重要的业务数据
        需要故障恢复能力

03.平衡性能和可靠性
    a.选择Kafka Channel
        适合大规模数据采集
        需要高性能和高可靠性
        有Kafka集群
    b.配置建议
        配置合理的分区数
        启用压缩
        配置acks为all
    c.适用场景
        大规模分布式系统
        数据需要多个下游系统消费
        需要长时间保留数据

04.混合使用
    a.不同数据使用不同Channel
        重要数据使用File Channel
        普通数据使用Memory Channel
        大规模数据使用Kafka Channel
    b.配置方式
        配置多个Channel
        使用Selector路由到不同Channel
        不同Sink消费不同Channel
    c.适用场景
        数据有不同的重要性级别
        需要灵活的数据处理策略
        系统资源有限

05.选择决策树
    a.是否可以容忍数据丢失
        可以:考虑Memory Channel
        不可以:考虑File Channel或Kafka Channel
    b.是否有Kafka集群
        有:考虑Kafka Channel
        没有:考虑File Channel
    c.数据量大小
        小:Memory Channel或File Channel
        大:Kafka Channel
    d.性能要求
        高:Memory Channel或Kafka Channel
        一般:File Channel

06.性能对比
    a.吞吐量
        Memory Channel:最高
        Kafka Channel:高
        File Channel:中等
    b.延迟
        Memory Channel:最低
        Kafka Channel:低
        File Channel:中等
    c.可靠性
        File Channel:最高
        Kafka Channel:高
        Memory Channel:低
    d.容量
        Kafka Channel:最大
        File Channel:大
        Memory Channel:受内存限制

4.6 Channel容量规划

01.容量评估
    a.数据量评估
        评估每秒产生的Event数量
        评估每个Event的平均大小
        计算每秒的数据量
    b.峰值评估
        评估峰值流量
        考虑突发流量
        预留一定的缓冲空间
    c.延迟评估
        评估Source和Sink的处理速度
        计算数据在Channel中的停留时间
        根据延迟要求确定容量

02.Memory Channel容量规划
    a.Event数量容量
        capacity:根据峰值流量和延迟计算
        一般设置为每秒Event数×延迟秒数×2
        例如:每秒1000个Event,延迟5秒,capacity设置为10000
    b.字节容量
        byteCapacity:根据Event大小和数量计算
        byteCapacity:capacity×平均Event大小×1.2
        不能超过JVM堆内存的80%
    c.内存规划
        JVM堆内存:byteCapacity÷0.8
        预留一定的内存给其他组件
        监控内存使用情况

03.File Channel容量规划
    a.Event数量容量
        capacity:根据数据保留时间和流量计算
        capacity:每秒Event数×保留时间秒数
        一般设置为几百万到几千万
    b.磁盘空间
        磁盘空间:capacity×平均Event大小×1.5
        考虑日志文件和checkpoint的开销
        预留足够的空闲空间
    c.数据目录规划
        使用多个数据目录
        每个目录在不同磁盘
        平均分配磁盘空间

04.Kafka Channel容量规划
    a.Topic分区数
        分区数:Flume Agent数×N(N一般为2-4)
        分区数影响并行度
        过多分区会增加管理开销
    b.数据保留时间
        根据业务需求设置保留时间
        一般设置为几小时到几天
        保留时间影响磁盘空间需求
    c.磁盘空间
        磁盘空间:每秒数据量×保留时间秒数×副本数×1.2
        考虑Kafka的存储开销
        监控磁盘空间使用

05.容量监控
    a.监控指标
        Channel容量使用率
        Event数量
        字节使用量
        Channel满的次数
    b.告警阈值
        容量使用率超过80%告警
        Channel满告警
        磁盘空间不足告警
    c.容量调整
        根据监控数据调整容量
        定期评估容量需求
        预留一定的增长空间

06.容量优化
    a.Source优化
        增大Source的batchSize
        提高Source的处理速度
        减少数据在Channel中的积压
    b.Sink优化
        增大Sink的batchSize
        提高Sink的处理速度
        加快数据消费
    c.Channel优化
        选择合适的Channel类型
        配置合理的容量
        使用多个Channel分流
    d.系统优化
        增加Agent实例
        使用负载均衡
        优化网络和磁盘IO

07.容量规划示例
    a.场景1:小规模日志采集
        数据量:每秒100个Event,每个1KB
        延迟要求:5秒
        Memory Channel容量:1000
        byteCapacity:1MB
    b.场景2:中规模日志采集
        数据量:每秒1000个Event,每个2KB
        保留时间:1小时
        File Channel容量:3600000
        磁盘空间:10GB
    c.场景3:大规模日志采集
        数据量:每秒10000个Event,每个5KB
        保留时间:24小时
        Kafka Channel分区数:32
        磁盘空间:500GB

5 Sink详解

5.1 汇总:7个

00.总结
    a.HDFS Sink
        将数据写入HDFS
        支持文件滚动和压缩
        适合离线数据存储
    b.Kafka Sink
        将数据发送到Kafka
        支持分区和压缩
        适合实时数据流处理
    c.Avro Sink
        通过Avro协议发送数据
        用于Agent之间的数据传输
        支持负载均衡和故障转移
    d.HBase Sink
        将数据写入HBase
        支持批量写入
        适合NoSQL存储
    e.Elasticsearch Sink
        将数据写入Elasticsearch
        支持批量索引
        适合日志检索和分析
    f.Logger Sink
        将Event输出到日志
        用于调试和测试
        不适合生产环境
    g.自定义Sink
        实现Sink接口
        满足特殊的数据输出需求
        需要Java开发能力

5.2 HDFS Sink

01.基本概念
    a.定义
        HDFS Sink将Event写入HDFS
        支持文件滚动和压缩
        适合大规模数据存储
    b.工作原理
        从Channel读取Event
        将Event写入HDFS文件
        根据配置滚动文件
        支持文件压缩和序列化
    c.应用场景
        离线数据存储
        数据仓库
        日志归档

02.配置参数
    a.基本配置
        type:hdfs
        hdfs.path:HDFS路径,必须配置
        hdfs.filePrefix:文件前缀,默认FlumeData
        hdfs.fileSuffix:文件后缀
    b.文件滚动配置
        hdfs.rollInterval:时间滚动间隔,默认30秒,0表示不按时间滚动
        hdfs.rollSize:大小滚动阈值,默认1024字节,0表示不按大小滚动
        hdfs.rollCount:Event数量滚动阈值,默认10,0表示不按数量滚动
    c.压缩配置
        hdfs.fileType:文件类型,SequenceFile或DataStream或CompressedStream
        hdfs.codeC:压缩编码,gzip或bzip2或lzo或lzop或snappy
        hdfs.writeFormat:写入格式,Text或Writable
    d.性能配置
        hdfs.batchSize:批量大小,默认100
        hdfs.threadsPoolSize:线程池大小,默认10
        hdfs.callTimeout:调用超时时间,默认10000ms

03.配置示例
    a.基本配置
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/events/%y-%m-%d/%H
        agent.sinks.k1.hdfs.filePrefix:events-
        agent.sinks.k1.channel:c1
        按小时分目录存储
    b.文件滚动
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/events/%y-%m-%d
        agent.sinks.k1.hdfs.filePrefix:events-
        agent.sinks.k1.hdfs.rollInterval:3600
        agent.sinks.k1.hdfs.rollSize:134217728
        agent.sinks.k1.hdfs.rollCount:0
        agent.sinks.k1.channel:c1
        按1小时或128MB滚动文件
    c.启用压缩
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/events/%y-%m-%d
        agent.sinks.k1.hdfs.fileType:CompressedStream
        agent.sinks.k1.hdfs.codeC:snappy
        agent.sinks.k1.channel:c1
        使用snappy压缩

04.路径格式化
    a.时间转义序列
        %y:年份(2位)
        %Y:年份(4位)
        %m:月份
        %d:日期
        %H:小时
        %M:分钟
        %S:秒
    b.Header转义序列
        %{header}:使用Event Header的值
        例如:%{host}使用host Header的值
    c.示例
        /flume/%y-%m-%d/%H:按小时分目录
        /flume/%{type}/%y-%m-%d:按type Header分目录

05.最佳实践
    a.合理配置文件滚动
        根据数据量配置滚动策略
        避免产生过多小文件
        一般设置为128MB或256MB
    b.启用压缩
        使用snappy或lzo压缩
        减少存储空间
        提高后续处理效率
    c.路径规划
        按时间分目录便于管理
        使用Header信息分类存储
        避免单个目录文件过多
    d.监控告警
        监控HDFS写入速率
        监控文件数量
        告警写入失败

5.3 Kafka Sink

01.基本概念
    a.定义
        Kafka Sink将Event发送到Kafka
        作为Kafka Producer工作
        支持分区和压缩
    b.工作原理
        从Channel读取Event
        将Event转换为Kafka消息
        发送到指定的Topic
        支持批量发送
    c.应用场景
        实时数据流处理
        与Kafka生态集成
        数据分发和路由

02.配置参数
    a.基本配置
        type:org.apache.flume.sink.kafka.KafkaSink
        kafka.bootstrap.servers:Kafka集群地址,必须配置
        kafka.topic:Topic名称,必须配置
    b.性能配置
        batchSize:批量大小,默认100
        kafka.producer.acks:acks配置,默认1
        kafka.producer.batch.size:Producer批量大小,默认16384
        kafka.producer.linger.ms:延迟时间,默认0
    c.高级配置
        kafka.producer.compression.type:压缩类型,默认none
        useFlumeEventFormat:是否使用Flume Event格式,默认false
        defaultPartitionId:默认分区ID
        partitioner.class:分区器类

03.配置示例
    a.基本配置
        agent.sinks.k1.type:org.apache.flume.sink.kafka.KafkaSink
        agent.sinks.k1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sinks.k1.kafka.topic:test-topic
        agent.sinks.k1.channel:c1
        发送数据到test-topic
    b.高可靠性配置
        agent.sinks.k1.type:org.apache.flume.sink.kafka.KafkaSink
        agent.sinks.k1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sinks.k1.kafka.topic:test-topic
        agent.sinks.k1.kafka.producer.acks:all
        agent.sinks.k1.kafka.producer.compression.type:snappy
        agent.sinks.k1.channel:c1
        配置acks为all,启用压缩
    c.高性能配置
        agent.sinks.k1.type:org.apache.flume.sink.kafka.KafkaSink
        agent.sinks.k1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sinks.k1.kafka.topic:test-topic
        agent.sinks.k1.batchSize:1000
        agent.sinks.k1.kafka.producer.batch.size:65536
        agent.sinks.k1.kafka.producer.linger.ms:10
        agent.sinks.k1.channel:c1
        增大批量大小和延迟时间

04.最佳实践
    a.Topic规划
        合理规划Topic分区数
        分区数影响并行度
        一般设置为Sink数的整数倍
    b.性能优化
        启用压缩减少网络传输
        增大批量大小提高吞吐量
        配置合理的linger.ms
    c.可靠性保证
        配置acks为all
        配置重试机制
        监控发送失败率
    d.监控告警
        监控发送速率
        监控发送失败次数
        告警Topic不可用

5.4 Avro Sink

01.基本概念
    a.定义
        Avro Sink通过Avro协议发送数据
        作为Avro Client工作
        用于Agent之间的数据传输
    b.工作原理
        从Channel读取Event
        通过Avro RPC发送到Avro Source
        支持批量发送
        支持压缩和SSL
    c.应用场景
        多层Agent架构
        跨网络数据传输
        Agent级联

02.配置参数
    a.基本配置
        type:avro
        hostname:目标主机名,必须配置
        port:目标端口,必须配置
    b.性能配置
        batch-size:批量大小,默认100
        compression-type:压缩类型,none或deflate
        connect-timeout:连接超时时间,默认20000ms
        request-timeout:请求超时时间,默认20000ms
    c.高级配置
        ssl:是否启用SSL
        trust-all-certs:是否信任所有证书
        maxIoWorkers:最大IO工作线程数

03.配置示例
    a.基本配置
        agent.sinks.k1.type:avro
        agent.sinks.k1.hostname:192.168.1.100
        agent.sinks.k1.port:4141
        agent.sinks.k1.channel:c1
        发送数据到192.168.1.100:4141
    b.启用压缩
        agent.sinks.k1.type:avro
        agent.sinks.k1.hostname:192.168.1.100
        agent.sinks.k1.port:4141
        agent.sinks.k1.compression-type:deflate
        agent.sinks.k1.channel:c1
        使用deflate压缩
    c.大批量配置
        agent.sinks.k1.type:avro
        agent.sinks.k1.hostname:192.168.1.100
        agent.sinks.k1.port:4141
        agent.sinks.k1.batch-size:1000
        agent.sinks.k1.channel:c1
        增大批量大小

04.最佳实践
    a.网络规划
        确保网络连通性
        配置合理的超时时间
        监控网络延迟
    b.启用压缩
        网络带宽有限时启用压缩
        减少网络传输量
        注意CPU使用率
    c.配置Sink Group
        使用多个Avro Sink
        配置负载均衡或故障转移
        提高可用性
    d.监控告警
        监控连接状态
        监控发送速率
        告警连接失败

5.5 HBase Sink

01.基本概念
    a.定义
        HBase Sink将Event写入HBase
        支持批量写入
        适合NoSQL存储
    b.工作原理
        从Channel读取Event
        将Event转换为HBase Put操作
        批量写入HBase
        支持异步写入
    c.应用场景
        实时数据写入HBase
        日志数据存储
        时序数据存储

02.配置参数
    a.基本配置
        type:hbase
        table:HBase表名,必须配置
        columnFamily:列族名,必须配置
        zookeeperQuorum:ZooKeeper地址,必须配置
    b.性能配置
        batchSize:批量大小,默认100
        serializer:序列化器,默认org.apache.flume.sink.hbase.SimpleHBaseEventSerializer
        serializer.*:序列化器配置参数
    c.高级配置
        kerberosPrincipal:Kerberos principal
        kerberosKeytab:Kerberos keytab文件
        coalesceIncrements:是否合并增量操作

03.配置示例
    a.基本配置
        agent.sinks.k1.type:hbase
        agent.sinks.k1.table:flume_table
        agent.sinks.k1.columnFamily:cf
        agent.sinks.k1.zookeeperQuorum:zk1:2181,zk2:2181
        agent.sinks.k1.channel:c1
        写入HBase表flume_table
    b.大批量配置
        agent.sinks.k1.type:hbase
        agent.sinks.k1.table:flume_table
        agent.sinks.k1.columnFamily:cf
        agent.sinks.k1.zookeeperQuorum:zk1:2181,zk2:2181
        agent.sinks.k1.batchSize:1000
        agent.sinks.k1.channel:c1
        增大批量大小提高吞吐量
    c.自定义序列化器
        agent.sinks.k1.type:hbase
        agent.sinks.k1.table:flume_table
        agent.sinks.k1.columnFamily:cf
        agent.sinks.k1.zookeeperQuorum:zk1:2181,zk2:2181
        agent.sinks.k1.serializer:com.example.MyHBaseSerializer
        agent.sinks.k1.channel:c1
        使用自定义序列化器

04.序列化器
    a.SimpleHBaseEventSerializer
        默认序列化器
        使用Event Body作为RowKey
        将整个Event写入一列
    b.RegexHBaseEventSerializer
        使用正则表达式解析Event
        提取字段作为列
        灵活但性能较低
    c.自定义序列化器
        实现HBaseEventSerializer接口
        自定义RowKey和列的生成逻辑
        满足特殊需求

05.最佳实践
    a.表设计
        合理设计RowKey
        避免热点问题
        预分区提高性能
    b.批量写入
        增大batchSize
        减少RPC次数
        提高吞吐量
    c.监控告警
        监控写入速率
        监控写入延迟
        告警写入失败

5.6 Elasticsearch Sink

01.基本概念
    a.定义
        Elasticsearch Sink将Event写入Elasticsearch
        支持批量索引
        适合日志检索和分析
    b.工作原理
        从Channel读取Event
        将Event转换为JSON文档
        批量索引到Elasticsearch
        支持动态索引名称
    c.应用场景
        日志检索系统
        实时数据分析
        全文搜索

02.配置参数
    a.基本配置
        type:elasticsearch
        hostNames:ES集群地址,必须配置
        indexName:索引名称,必须配置
        indexType:索引类型,默认logs
    b.性能配置
        batchSize:批量大小,默认100
        ttl:文档TTL
        serializer:序列化器
    c.高级配置
        clusterName:集群名称
        client:客户端类型,transport或rest
        client.*:客户端配置参数

03.配置示例
    a.基本配置
        agent.sinks.k1.type:elasticsearch
        agent.sinks.k1.hostNames:es1:9300,es2:9300
        agent.sinks.k1.indexName:flume
        agent.sinks.k1.indexType:log
        agent.sinks.k1.channel:c1
        写入ES索引flume
    b.动态索引名称
        agent.sinks.k1.type:elasticsearch
        agent.sinks.k1.hostNames:es1:9300,es2:9300
        agent.sinks.k1.indexName:flume-%{+YYYY.MM.dd}
        agent.sinks.k1.indexType:log
        agent.sinks.k1.channel:c1
        按日期创建索引
    c.大批量配置
        agent.sinks.k1.type:elasticsearch
        agent.sinks.k1.hostNames:es1:9300,es2:9300
        agent.sinks.k1.indexName:flume
        agent.sinks.k1.batchSize:1000
        agent.sinks.k1.channel:c1
        增大批量大小

04.最佳实践
    a.索引规划
        按时间分索引
        控制索引大小
        定期清理旧索引
    b.性能优化
        增大批量大小
        使用bulk API
        监控索引性能
    c.监控告警
        监控索引速率
        监控索引失败次数
        告警ES集群不可用

5.7 Logger Sink

01.基本概念
    a.定义
        Logger Sink将Event输出到日志
        用于调试和测试
        不适合生产环境
    b.工作原理
        从Channel读取Event
        将Event信息输出到日志
        支持配置日志级别
    c.应用场景
        开发调试
        测试验证
        问题排查

02.配置参数
    a.基本配置
        type:logger
        maxBytesToLog:最大输出字节数,默认16
    b.日志级别
        通过log4j配置日志级别
        一般设置为INFO或DEBUG

03.配置示例
    a.基本配置
        agent.sinks.k1.type:logger
        agent.sinks.k1.channel:c1
        输出Event到日志
    b.输出完整Body
        agent.sinks.k1.type:logger
        agent.sinks.k1.maxBytesToLog:1024
        agent.sinks.k1.channel:c1
        输出1024字节的Body

04.最佳实践
    a.仅用于调试
        不要在生产环境使用
        会产生大量日志
        影响性能
    b.限制输出大小
        配置maxBytesToLog
        避免日志过大
        便于查看

5.8 自定义Sink

01.实现接口
    a.Sink接口
        实现org.apache.flume.Sink接口
        或继承AbstractSink抽象类
        实现start、stop、process等方法
    b.核心方法
        configure(Context):读取配置参数
        start():启动Sink
        stop():停止Sink
        process():处理数据,返回Status
    c.Status类型
        READY:成功处理数据
        BACKOFF:Channel为空,需要退避
        用于控制Sink的执行频率

02.实现步骤
    a.创建类
        public class MySink extends AbstractSink implements Configurable
        继承AbstractSink
        实现Configurable接口
    b.实现configure方法
        读取配置文件中的参数
        初始化Sink的配置
        验证配置的有效性
    c.实现start方法
        初始化资源
        建立连接
        准备发送数据
    d.实现stop方法
        关闭连接
        释放资源
        清理状态
    e.实现process方法
        从Channel读取Event
        处理Event
        发送到目标系统
        返回Status

03.代码示例
    a.基本结构
        public class MySink extends AbstractSink implements Configurable{
            private String myProp;
            public void configure(Context context){
                myProp:context.getString("myProp","defaultValue");
            }
            public void start(){
                super.start();
            }
            public void stop(){
                super.stop();
            }
            public Status process() throws EventDeliveryException{
                return Status.READY;
            }
        }
    b.从Channel读取
        Channel channel:getChannel();
        Transaction tx:channel.getTransaction();
        try{
            tx.begin();
            Event event:channel.take();
            if(event::null){
                tx.commit();
                return Status.BACKOFF;
            }
            tx.commit();
            return Status.READY;
        }catch(Exception e){
            tx.rollback();
            throw new EventDeliveryException(e);
        }finally{
            tx.close();
        }
    c.批量读取
        Channel channel:getChannel();
        Transaction tx:channel.getTransaction();
        List<Event> events:new ArrayList<>();
        try{
            tx.begin();
            for(int i:0;i<batchSize;i++){
                Event event:channel.take();
                if(event::null){
                    break;
                }
                events.add(event);
            }
            tx.commit();
            if(events.isEmpty()){
                return Status.BACKOFF;
            }
            return Status.READY;
        }catch(Exception e){
            tx.rollback();
            throw new EventDeliveryException(e);
        }finally{
            tx.close();
        }

04.打包部署
    a.打包
        使用Maven或Gradle打包成jar
        包含所有依赖
        或者使用provided scope排除Flume依赖
    b.部署
        将jar文件放到Flume的lib目录
        或者使用plugins.d目录
        重启Flume Agent
    c.配置
        agent.sinks.k1.type:com.example.MySink
        agent.sinks.k1.myProp:myValue
        agent.sinks.k1.channel:c1
        使用完整类名指定Sink类型

05.最佳实践
    a.异常处理
        捕获所有异常
        避免Sink崩溃
        记录详细的错误日志
    b.资源管理
        及时释放资源
        避免资源泄漏
        使用try-finally确保资源释放
    c.性能优化
        批量处理数据
        使用合理的退避策略
        避免阻塞操作
    d.事务处理
        正确使用事务
        发送成功后才提交事务
        发送失败时回滚事务
    e.监控
        暴露JMX指标
        记录关键的运行指标
        便于监控和调试

6 高级特性

6.1 拦截器详解

01.拦截器链
    a.链式处理
        多个Interceptor按顺序执行
        前一个的输出是后一个的输入
        任何一个Interceptor都可以修改或丢弃Event
    b.配置方式
        agent.sources.r1.interceptors:i1 i2 i3
        agent.sources.r1.interceptors.i1.type:timestamp
        agent.sources.r1.interceptors.i2.type:host
        agent.sources.r1.interceptors.i3.type:static
        按i1→i2→i3顺序执行
    c.执行顺序
        严格按照配置顺序执行
        顺序很重要,影响最终结果
        需要仔细规划拦截器顺序

02.Timestamp Interceptor详解
    a.功能
        在Event Header中添加时间戳
        key为timestamp,值为当前时间的毫秒数
        用于记录Event产生或接收的时间
    b.配置参数
        type:timestamp
        preserveExisting:是否保留已有时间戳,默认false
        如果为true且Header中已有timestamp,则不覆盖
    c.应用场景
        HDFS Sink按时间分目录
        数据分析时需要时间信息
        监控数据延迟
    d.配置示例
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:timestamp
        agent.sources.r1.interceptors.i1.preserveExisting:true

03.Host Interceptor详解
    a.功能
        在Event Header中添加主机名或IP
        key为host,值为主机名或IP地址
        用于标识数据来源
    b.配置参数
        type:host
        useIP:使用IP还是主机名,默认true
        preserveExisting:是否保留已有host,默认false
        hostHeader:Header的key,默认host
    c.应用场景
        标识数据来源服务器
        按服务器分类存储数据
        问题排查和定位
    d.配置示例
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:host
        agent.sources.r1.interceptors.i1.useIP:false
        agent.sources.r1.interceptors.i1.hostHeader:hostname

04.Static Interceptor详解
    a.功能
        在Event Header中添加静态键值对
        可以添加多个键值对
        用于标记数据类型或来源
    b.配置参数
        type:static
        key:Header的key
        value:Header的value
        可以配置多个key-value对
    c.应用场景
        标记数据类型
        标记数据来源应用
        用于数据路由和分类
    d.配置示例
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:static
        agent.sources.r1.interceptors.i1.key:datacenter
        agent.sources.r1.interceptors.i1.value:dc1

05.Regex Filtering Interceptor详解
    a.功能
        根据正则表达式过滤Event
        匹配的Event保留或丢弃
        用于过滤无效或不需要的数据
    b.配置参数
        type:regex_filter
        regex:正则表达式,必须配置
        excludeEvents:true表示排除匹配的Event,false表示保留匹配的Event,默认false
    c.应用场景
        过滤无效日志
        过滤敏感信息
        只保留特定格式的数据
    d.配置示例
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:regex_filter
        agent.sources.r1.interceptors.i1.regex:^ERROR.*
        agent.sources.r1.interceptors.i1.excludeEvents:false
        只保留以ERROR开头的日志

06.Regex Extractor Interceptor详解
    a.功能
        从Event Body中提取信息到Header
        使用正则表达式和捕获组提取
        用于数据解析和路由
    b.配置参数
        type:regex_extractor
        regex:正则表达式,必须配置
        serializers:序列化器列表,必须配置
        serializers.<name>.name:Header的key
    c.应用场景
        从日志中提取字段
        提取信息用于路由
        数据解析和转换
    d.配置示例
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:regex_extractor
        agent.sources.r1.interceptors.i1.regex:^(\\d{4}-\\d{2}-\\d{2})\\s+(\\w+)\\s+(.*)$
        agent.sources.r1.interceptors.i1.serializers:s1 s2 s3
        agent.sources.r1.interceptors.i1.serializers.s1.name:date
        agent.sources.r1.interceptors.i1.serializers.s2.name:level
        agent.sources.r1.interceptors.i1.serializers.s3.name:message
        提取日期、级别和消息到Header

07.拦截器最佳实践
    a.性能考虑
        拦截器会影响性能
        避免复杂的正则表达式
        避免在拦截器中做重计算
    b.顺序规划
        先添加信息,后过滤
        先提取字段,后路由
        合理规划拦截器顺序
    c.错误处理
        拦截器异常会导致Event丢失
        需要充分测试
        记录异常日志
    d.监控
        监控拦截器处理时间
        监控Event丢弃数量
        及时发现问题

6.2 选择器详解

01.Replicating Selector详解
    a.工作原理
        将Event复制到所有Channel
        每个Channel都会收到相同的Event
        实现数据的多路复用
    b.必选和可选Channel
        必选Channel:必须写入成功
        可选Channel:写入失败不影响必选Channel
        提高系统的容错能力
    c.配置参数
        selector.type:replicating
        selector.optional:可选Channel列表
    d.应用场景
        数据需要发送到多个目标系统
        实时和离线处理并存
        数据备份和容灾
    e.配置示例
        agent.sources.r1.channels:c1 c2 c3
        agent.sources.r1.selector.type:replicating
        agent.sources.r1.selector.optional:c3
        Event复制到c1、c2、c3,c3为可选

02.Multiplexing Selector详解
    a.工作原理
        根据Event Header的值选择Channel
        不同的Header值路由到不同的Channel
        实现数据的条件路由
    b.映射规则
        selector.mapping.<value>:Channel列表
        根据Header值映射到Channel
        支持配置多个映射规则
    c.默认Channel
        selector.default:默认Channel列表
        Header值不匹配时使用默认Channel
        如果不配置,不匹配的Event会被丢弃
    d.配置参数
        selector.type:multiplexing
        selector.header:用于路由的Header key,必须配置
        selector.mapping.*:映射规则
        selector.default:默认Channel
    e.应用场景
        根据数据类型路由
        根据日志级别路由
        不同业务线的数据隔离
    f.配置示例
        agent.sources.r1.channels:c1 c2 c3
        agent.sources.r1.selector.type:multiplexing
        agent.sources.r1.selector.header:level
        agent.sources.r1.selector.mapping.ERROR:c1
        agent.sources.r1.selector.mapping.WARN:c2
        agent.sources.r1.selector.default:c3
        ERROR级别到c1,WARN级别到c2,其他到c3

03.选择器组合使用
    a.Interceptor+Selector
        使用Interceptor添加Header
        使用Selector根据Header路由
        实现灵活的数据分发
    b.配置示例
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:regex_extractor
        agent.sources.r1.interceptors.i1.regex:^(\\w+)\\s+.*$
        agent.sources.r1.interceptors.i1.serializers:s1
        agent.sources.r1.interceptors.i1.serializers.s1.name:level
        agent.sources.r1.channels:c1 c2 c3
        agent.sources.r1.selector.type:multiplexing
        agent.sources.r1.selector.header:level
        agent.sources.r1.selector.mapping.ERROR:c1
        agent.sources.r1.selector.mapping.WARN:c2
        agent.sources.r1.selector.default:c3
        提取日志级别并路由到不同Channel
    c.应用场景
        复杂的数据路由逻辑
        动态的数据分类
        灵活的数据处理

04.选择器最佳实践
    a.合理规划Channel
        根据业务需求规划Channel数量
        避免Channel过多
        每个Channel对应一个处理流程
    b.配置默认Channel
        Multiplexing Selector建议配置默认Channel
        避免数据丢失
        记录不匹配的数据
    c.监控
        监控每个Channel的数据量
        监控数据分布是否均衡
        及时调整路由规则
    d.测试
        充分测试路由规则
        验证数据路由的正确性
        测试异常情况

6.3 负载均衡

01.Load Balancing Sink Processor
    a.工作原理
        将数据分发到多个Sink
        支持轮询和随机两种策略
        提高整体吞吐量
    b.选择策略
        round_robin:轮询策略,依次使用每个Sink
        random:随机策略,随机选择Sink
        自定义策略:实现LoadBalancingSelector接口
    c.配置参数
        processor.type:load_balance
        processor.selector:选择策略,默认round_robin
        processor.backoff:失败后的退避策略,默认false
    d.应用场景
        多个目标系统并行处理
        提高整体吞吐量
        目标系统有多个节点

02.轮询策略
    a.工作原理
        按顺序依次使用每个Sink
        每个Sink处理的数据量基本相同
        实现负载均衡
    b.优点
        负载分布均匀
        实现简单
        可预测性强
    c.缺点
        不考虑Sink的实际负载
        不考虑Sink的处理能力
        可能导致某些Sink过载
    d.配置示例
        agent.sinkgroups:g1
        agent.sinkgroups.g1.sinks:k1 k2 k3
        agent.sinkgroups.g1.processor.type:load_balance
        agent.sinkgroups.g1.processor.selector:round_robin

03.随机策略
    a.工作原理
        随机选择一个Sink
        长期来看负载基本均衡
        短期可能不均衡
    b.优点
        实现简单
        避免顺序依赖
        适合动态环境
    c.缺点
        短期负载可能不均衡
        不可预测
        可能导致某些Sink空闲
    d.配置示例
        agent.sinkgroups:g1
        agent.sinkgroups.g1.sinks:k1 k2 k3
        agent.sinkgroups.g1.processor.type:load_balance
        agent.sinkgroups.g1.processor.selector:random

04.退避机制
    a.工作原理
        Sink失败后进入黑名单
        黑名单中的Sink暂时不使用
        定期重试黑名单中的Sink
        重试成功后移出黑名单
    b.配置参数
        processor.backoff:是否启用退避,默认false
        processor.selector.maxTimeOut:最大退避时间,默认30000ms
    c.应用场景
        目标系统可能临时不可用
        提高系统的容错能力
        自动恢复
    d.配置示例
        agent.sinkgroups:g1
        agent.sinkgroups.g1.sinks:k1 k2 k3
        agent.sinkgroups.g1.processor.type:load_balance
        agent.sinkgroups.g1.processor.selector:round_robin
        agent.sinkgroups.g1.processor.backoff:true
        agent.sinkgroups.g1.processor.selector.maxTimeOut:10000

05.负载均衡最佳实践
    a.选择合适的策略
        一般使用轮询策略
        动态环境使用随机策略
        根据实际情况选择
    b.启用退避机制
        提高容错能力
        自动处理临时故障
        配置合理的退避时间
    c.监控
        监控每个Sink的负载
        监控Sink的成功率
        监控退避情况
    d.容量规划
        确保每个Sink有足够的处理能力
        避免单个Sink成为瓶颈
        预留一定的冗余

6.4 故障转移

01.Failover Sink Processor
    a.工作原理
        配置Sink的优先级
        优先使用优先级高的Sink
        Sink失败后切换到备用Sink
        实现高可用
    b.优先级配置
        processor.priority.<sinkName>:优先级值
        数值越大优先级越高
        优先使用优先级高的Sink
    c.配置参数
        processor.type:failover
        processor.priority.*:Sink优先级
        processor.maxpenalty:最大惩罚时间,默认30000ms
    d.应用场景
        主备模式的数据传输
        保证数据传输的高可用性
        目标系统有主备节点

02.故障检测
    a.检测机制
        Sink处理失败时触发故障检测
        将失败的Sink标记为不可用
        切换到下一个优先级的Sink
    b.惩罚机制
        失败的Sink进入惩罚期
        惩罚期内不使用该Sink
        惩罚期结束后重新尝试
        惩罚时间逐渐增加
    c.恢复机制
        定期尝试使用失败的Sink
        尝试成功后恢复使用
        惩罚时间重置

03.配置示例
    a.主备配置
        agent.sinkgroups:g1
        agent.sinkgroups.g1.sinks:k1 k2
        agent.sinkgroups.g1.processor.type:failover
        agent.sinkgroups.g1.processor.priority.k1:10
        agent.sinkgroups.g1.processor.priority.k2:5
        agent.sinkgroups.g1.processor.maxpenalty:10000
        k1为主Sink,k2为备Sink
    b.多备份配置
        agent.sinkgroups:g1
        agent.sinkgroups.g1.sinks:k1 k2 k3
        agent.sinkgroups.g1.processor.type:failover
        agent.sinkgroups.g1.processor.priority.k1:10
        agent.sinkgroups.g1.processor.priority.k2:5
        agent.sinkgroups.g1.processor.priority.k3:1
        agent.sinkgroups.g1.processor.maxpenalty:10000
        k1为主,k2为第一备份,k3为第二备份

04.故障转移最佳实践
    a.合理配置优先级
        主Sink优先级最高
        备Sink优先级递减
        确保故障转移顺序正确
    b.配置合理的惩罚时间
        惩罚时间不宜过短
        避免频繁切换
        一般设置为10-30秒
    c.监控
        监控Sink的状态
        监控故障转移次数
        监控Sink的恢复情况
    d.测试
        定期测试故障转移功能
        验证备用Sink是否正常工作
        确保高可用性

6.5 多路复用

01.多路复用概念
    a.定义
        一个Source将数据发送到多个Channel
        不同的Channel连接不同的Sink
        实现数据的多路分发
    b.实现方式
        使用Replicating Selector复制数据
        使用Multiplexing Selector条件路由
        结合Interceptor实现复杂逻辑
    c.应用场景
        数据需要发送到多个目标系统
        实时和离线处理并存
        不同的下游系统处理不同的数据

02.数据复制
    a.使用Replicating Selector
        将Event复制到所有Channel
        每个Channel都会收到相同的数据
        适合数据需要多份处理的场景
    b.配置示例
        agent.sources.r1.channels:c1 c2 c3
        agent.sources.r1.selector.type:replicating
        agent.channels.c1.type:memory
        agent.channels.c2.type:memory
        agent.channels.c3.type:memory
        agent.sinks.k1.channel:c1
        agent.sinks.k2.channel:c2
        agent.sinks.k3.channel:c3
        数据复制到三个Channel,分别由三个Sink处理
    c.应用场景
        数据同时写入HDFS和Kafka
        实时分析和离线存储
        数据备份

03.条件路由
    a.使用Multiplexing Selector
        根据Header值路由到不同Channel
        不同的数据发送到不同的目标
        实现数据分类处理
    b.配置示例
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:regex_extractor
        agent.sources.r1.interceptors.i1.regex:^(\\w+)\\s+.*$
        agent.sources.r1.interceptors.i1.serializers:s1
        agent.sources.r1.interceptors.i1.serializers.s1.name:type
        agent.sources.r1.channels:c1 c2 c3
        agent.sources.r1.selector.type:multiplexing
        agent.sources.r1.selector.header:type
        agent.sources.r1.selector.mapping.login:c1
        agent.sources.r1.selector.mapping.logout:c2
        agent.sources.r1.selector.default:c3
        根据类型路由到不同Channel
    c.应用场景
        不同类型的日志分别处理
        根据优先级路由
        业务数据隔离

04.混合模式
    a.复制+路由
        部分数据复制到所有Channel
        部分数据条件路由
        实现复杂的数据分发逻辑
    b.实现方式
        使用多个Source
        每个Source配置不同的Selector
        组合实现复杂逻辑
    c.应用场景
        复杂的业务需求
        灵活的数据处理
        多种数据流并存

05.多路复用最佳实践
    a.合理规划数据流
        明确每个Channel的用途
        避免数据重复处理
        优化资源使用
    b.监控数据流
        监控每个Channel的数据量
        监控数据分布
        及时发现问题
    c.性能优化
        避免不必要的数据复制
        使用条件路由减少数据量
        合理配置Channel容量
    d.可靠性保证
        使用可选Channel提高容错
        配置合理的重试策略
        监控数据丢失情况

6.6 聚合模式

01.聚合概念
    a.定义
        多个Source将数据发送到一个Channel
        或多个Agent将数据发送到一个中心Agent
        实现数据的汇聚
    b.实现方式
        多个Source配置同一个Channel
        使用Avro Source接收多个Agent的数据
        使用Kafka作为中间层
    c.应用场景
        多个服务器的日志汇总
        分布式系统的数据聚合
        减少下游系统的连接数

02.单Agent聚合
    a.多Source单Channel
        多个Source将数据写入同一个Channel
        一个Sink从Channel读取数据
        实现简单的数据聚合
    b.配置示例
        agent.sources:r1 r2 r3
        agent.sources.r1.type:exec
        agent.sources.r1.command:tail -F /var/log/app1.log
        agent.sources.r1.channels:c1
        agent.sources.r2.type:exec
        agent.sources.r2.command:tail -F /var/log/app2.log
        agent.sources.r2.channels:c1
        agent.sources.r3.type:exec
        agent.sources.r3.command:tail -F /var/log/app3.log
        agent.sources.r3.channels:c1
        agent.channels.c1.type:memory
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.channel:c1
        三个Source的数据聚合到一个Channel
    c.应用场景
        单机多个日志文件聚合
        多个数据源统一处理
        简化配置

03.多Agent聚合
    a.扇入架构
        多个Agent将数据发送到中心Agent
        中心Agent负责数据的聚合和统一处理
        减少目标系统的连接数
    b.配置示例
        采集Agent配置:
        agent1.sources.r1.type:exec
        agent1.sources.r1.command:tail -F /var/log/app.log
        agent1.sources.r1.channels:c1
        agent1.channels.c1.type:memory
        agent1.sinks.k1.type:avro
        agent1.sinks.k1.hostname:center-agent
        agent1.sinks.k1.port:4141
        agent1.sinks.k1.channel:c1
        中心Agent配置:
        center.sources.r1.type:avro
        center.sources.r1.bind:0.0.0.0
        center.sources.r1.port:4141
        center.sources.r1.channels:c1
        center.channels.c1.type:memory
        center.sinks.k1.type:hdfs
        center.sinks.k1.channel:c1
        多个采集Agent发送数据到中心Agent
    c.应用场景
        多个服务器的日志汇总
        跨数据中心数据聚合
        统一的数据处理

04.使用Kafka聚合
    a.Kafka作为中间层
        多个Agent将数据发送到Kafka
        中心Agent从Kafka消费数据
        利用Kafka的高性能和高可靠性
    b.配置示例
        采集Agent配置:
        agent1.sources.r1.type:exec
        agent1.sources.r1.command:tail -F /var/log/app.log
        agent1.sources.r1.channels:c1
        agent1.channels.c1.type:memory
        agent1.sinks.k1.type:org.apache.flume.sink.kafka.KafkaSink
        agent1.sinks.k1.kafka.bootstrap.servers:kafka1:9092
        agent1.sinks.k1.kafka.topic:flume-data
        agent1.sinks.k1.channel:c1
        中心Agent配置:
        center.sources.r1.type:org.apache.flume.source.kafka.KafkaSource
        center.sources.r1.kafka.bootstrap.servers:kafka1:9092
        center.sources.r1.kafka.topics:flume-data
        center.sources.r1.kafka.consumer.group.id:flume-center
        center.sources.r1.channels:c1
        center.channels.c1.type:memory
        center.sinks.k1.type:hdfs
        center.sinks.k1.channel:c1
        使用Kafka作为中间层聚合数据
    c.优点
        解耦采集Agent和中心Agent
        提高系统的可靠性和扩展性
        利用Kafka的缓存能力

05.聚合模式最佳实践
    a.选择合适的聚合方式
        小规模使用单Agent聚合
        大规模使用多Agent聚合
        高可靠性使用Kafka聚合
    b.容量规划
        中心Agent需要足够的处理能力
        Channel容量要足够大
        网络带宽要充足
    c.监控
        监控中心Agent的负载
        监控数据聚合速率
        监控数据延迟
    d.高可用
        部署多个中心Agent
        使用负载均衡
        配置故障转移

7 实战

7.1 安装部署

01.环境准备
    a.系统要求
        Linux操作系统,CentOS 7或Ubuntu 16.04以上
        Java 1.8或以上版本
        足够的磁盘空间和内存
    b.下载Flume
        从Apache官网下载Flume二进制包
        当前稳定版本为1.9.0或1.11.0
        解压到指定目录
    c.环境变量
        配置JAVA_HOME环境变量
        配置FLUME_HOME环境变量
        将$FLUME_HOME/bin添加到PATH

02.单机部署
    a.解压安装
        tar -zxvf apache-flume-1.9.0-bin.tar.gz
        mv apache-flume-1.9.0-bin /opt/flume
        配置环境变量
    b.验证安装
        flume-ng version
        查看版本信息
        确认安装成功
    c.配置文件
        创建配置文件flume-conf.properties
        配置Source、Channel、Sink
        启动Agent测试

03.集群部署
    a.规划部署
        确定Agent数量和角色
        规划网络拓扑
        分配服务器资源
    b.分发安装包
        在所有节点安装Flume
        配置统一的环境变量
        同步配置文件
    c.启动集群
        按顺序启动各个Agent
        先启动中心Agent
        再启动采集Agent

7.2 配置文件

01.配置文件结构
    a.基本格式
        使用properties格式
        key-value对配置
        支持注释
    b.命名规范
        agent.sources:Source列表
        agent.channels:Channel列表
        agent.sinks:Sink列表
        agent为Agent名称,可自定义
    c.组件配置
        agent.sources.<name>.*:Source配置
        agent.channels.<name>.*:Channel配置
        agent.sinks.<name>.*:Sink配置
        agent.sinkgroups.<name>.*:Sink Group配置

02.配置示例
    a.简单示例
        agent.sources:r1
        agent.channels:c1
        agent.sinks:k1
        agent.sources.r1.type:netcat
        agent.sources.r1.bind:localhost
        agent.sources.r1.port:44444
        agent.sources.r1.channels:c1
        agent.channels.c1.type:memory
        agent.channels.c1.capacity:1000
        agent.sinks.k1.type:logger
        agent.sinks.k1.channel:c1
    b.完整示例
        agent.sources:r1
        agent.channels:c1
        agent.sinks:k1
        agent.sources.r1.type:exec
        agent.sources.r1.command:tail -F /var/log/app.log
        agent.sources.r1.channels:c1
        agent.sources.r1.interceptors:i1 i2
        agent.sources.r1.interceptors.i1.type:timestamp
        agent.sources.r1.interceptors.i2.type:host
        agent.channels.c1.type:file
        agent.channels.c1.checkpointDir:/var/flume/checkpoint
        agent.channels.c1.dataDirs:/var/flume/data
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/events/%y-%m-%d/%H
        agent.sinks.k1.hdfs.filePrefix:events-
        agent.sinks.k1.hdfs.rollInterval:3600
        agent.sinks.k1.channel:c1

03.启动命令
    a.基本启动
        flume-ng agent -n agent -c conf -f conf/flume-conf.properties
        -n指定Agent名称
        -c指定配置目录
        -f指定配置文件
    b.后台启动
        nohup flume-ng agent -n agent -c conf -f conf/flume-conf.properties > /var/log/flume.log 2>&1 &
        使用nohup后台运行
        重定向日志到文件
    c.JVM参数
        export JAVA_OPTS="-Xms2g -Xmx4g"
        flume-ng agent -n agent -c conf -f conf/flume-conf.properties
        配置JVM堆内存

7.3 常用案例

01.日志文件采集到HDFS
    a.需求
        采集应用日志文件
        按小时分目录存储到HDFS
        文件大小超过128MB滚动
    b.配置
        agent.sources:r1
        agent.channels:c1
        agent.sinks:k1
        agent.sources.r1.type:TAILDIR
        agent.sources.r1.positionFile:/var/flume/taildir_position.json
        agent.sources.r1.filegroups:f1
        agent.sources.r1.filegroups.f1:/var/log/app.log
        agent.sources.r1.channels:c1
        agent.channels.c1.type:file
        agent.channels.c1.checkpointDir:/var/flume/checkpoint
        agent.channels.c1.dataDirs:/var/flume/data
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/logs/%y-%m-%d/%H
        agent.sinks.k1.hdfs.filePrefix:app-
        agent.sinks.k1.hdfs.rollInterval:0
        agent.sinks.k1.hdfs.rollSize:134217728
        agent.sinks.k1.hdfs.rollCount:0
        agent.sinks.k1.hdfs.fileType:DataStream
        agent.sinks.k1.channel:c1

02.多服务器日志汇总
    a.需求
        多个Web服务器的日志汇总
        发送到中心Agent
        统一存储到HDFS
    b.Web服务器Agent配置
        agent.sources:r1
        agent.channels:c1
        agent.sinks:k1
        agent.sources.r1.type:exec
        agent.sources.r1.command:tail -F /var/log/nginx/access.log
        agent.sources.r1.channels:c1
        agent.channels.c1.type:memory
        agent.channels.c1.capacity:10000
        agent.sinks.k1.type:avro
        agent.sinks.k1.hostname:center-server
        agent.sinks.k1.port:4141
        agent.sinks.k1.channel:c1
    c.中心Agent配置
        agent.sources:r1
        agent.channels:c1
        agent.sinks:k1
        agent.sources.r1.type:avro
        agent.sources.r1.bind:0.0.0.0
        agent.sources.r1.port:4141
        agent.sources.r1.channels:c1
        agent.channels.c1.type:file
        agent.channels.c1.checkpointDir:/var/flume/checkpoint
        agent.channels.c1.dataDirs:/var/flume/data
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/nginx/%y-%m-%d
        agent.sinks.k1.channel:c1

03.Kafka到HDFS
    a.需求
        从Kafka消费数据
        写入HDFS
        按天分目录
    b.配置
        agent.sources:r1
        agent.channels:c1
        agent.sinks:k1
        agent.sources.r1.type:org.apache.flume.source.kafka.KafkaSource
        agent.sources.r1.kafka.bootstrap.servers:kafka1:9092,kafka2:9092
        agent.sources.r1.kafka.topics:app-logs
        agent.sources.r1.kafka.consumer.group.id:flume-consumer
        agent.sources.r1.channels:c1
        agent.channels.c1.type:file
        agent.channels.c1.checkpointDir:/var/flume/checkpoint
        agent.channels.c1.dataDirs:/var/flume/data
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/kafka/%y-%m-%d
        agent.sinks.k1.hdfs.filePrefix:kafka-
        agent.sinks.k1.hdfs.rollInterval:3600
        agent.sinks.k1.channel:c1

04.日志分类存储
    a.需求
        根据日志级别分类存储
        ERROR日志单独存储
        其他日志正常存储
    b.配置
        agent.sources:r1
        agent.channels:c1 c2
        agent.sinks:k1 k2
        agent.sources.r1.type:exec
        agent.sources.r1.command:tail -F /var/log/app.log
        agent.sources.r1.channels:c1 c2
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:regex_extractor
        agent.sources.r1.interceptors.i1.regex:^(\\w+)\\s+.*$
        agent.sources.r1.interceptors.i1.serializers:s1
        agent.sources.r1.interceptors.i1.serializers.s1.name:level
        agent.sources.r1.selector.type:multiplexing
        agent.sources.r1.selector.header:level
        agent.sources.r1.selector.mapping.ERROR:c1
        agent.sources.r1.selector.default:c2
        agent.channels.c1.type:memory
        agent.channels.c2.type:memory
        agent.sinks.k1.type:hdfs
        agent.sinks.k1.hdfs.path:/flume/error/%y-%m-%d
        agent.sinks.k1.channel:c1
        agent.sinks.k2.type:hdfs
        agent.sinks.k2.hdfs.path:/flume/normal/%y-%m-%d
        agent.sinks.k2.channel:c2

7.4 性能调优

01.Source调优
    a.增大批量大小
        配置batchSize参数
        减少Channel写入次数
        提高吞吐量
    b.多线程处理
        配置threads参数
        提高并发处理能力
        注意CPU使用率
    c.选择合适的Source类型
        Taildir Source优于Exec Source
        支持断点续传
        可靠性更高

02.Channel调优
    a.Memory Channel优化
        增大capacity和transactionCapacity
        配置合理的byteCapacity
        监控内存使用
    b.File Channel优化
        使用多个数据目录
        数据目录分布在不同磁盘
        使用SSD存储checkpoint
        调整checkpoint间隔
    c.Kafka Channel优化
        增加Topic分区数
        启用压缩
        配置合理的批量大小

03.Sink调优
    a.增大批量大小
        配置batchSize参数
        减少目标系统写入次数
        提高吞吐量
    b.并行Sink
        使用Sink Group
        配置多个Sink并行处理
        提高整体性能
    c.目标系统优化
        HDFS Sink配置合理的滚动策略
        HBase Sink增大批量大小
        Kafka Sink启用压缩

04.JVM调优
    a.堆内存配置
        根据Channel容量配置堆内存
        Memory Channel需要较大堆内存
        一般配置2-4GB
    b.GC配置
        使用G1 GC
        配置合理的GC参数
        监控GC情况
    c.JVM参数示例
        export JAVA_OPTS="-Xms2g -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=200"

05.系统调优
    a.网络优化
        增加网络带宽
        减少网络延迟
        使用高速网络
    b.磁盘优化
        使用SSD
        使用RAID提高性能
        监控磁盘IO
    c.操作系统优化
        增加文件描述符限制
        调整网络参数
        优化内核参数

7.5 监控运维

01.JMX监控
    a.启用JMX
        export JAVA_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port:9999 -Dcom.sun.management.jmxremote.authenticate:false -Dcom.sun.management.jmxremote.ssl:false"
        启动Flume时启用JMX
    b.监控指标
        Source接收速率
        Channel容量使用率
        Sink发送速率
        Event处理延迟
    c.监控工具
        JConsole
        VisualVM
        Prometheus+Grafana

02.HTTP监控
    a.启用HTTP监控
        在配置文件中添加
        agent.sources.r1.interceptors:i1
        agent.sources.r1.interceptors.i1.type:org.apache.flume.interceptor.HostInterceptor$Builder
    b.访问监控接口
        http://hostname:port/metrics
        查看JSON格式的监控数据
    c.监控指标
        ChannelSize:Channel中Event数量
        ChannelFillPercentage:Channel使用率
        EventPutSuccessCount:成功写入Event数量
        EventTakeSuccessCount:成功读取Event数量

03.日志监控
    a.日志配置
        配置log4j.properties
        设置合理的日志级别
        配置日志滚动
    b.关键日志
        Source启动和停止日志
        Channel满的日志
        Sink失败的日志
        异常堆栈信息
    c.日志分析
        定期分析日志
        统计错误信息
        发现潜在问题

04.告警配置
    a.告警指标
        Channel使用率超过80%
        Sink连续失败
        Agent进程退出
        磁盘空间不足
    b.告警方式
        邮件告警
        短信告警
        企业微信/钉钉告警
    c.告警处理
        及时响应告警
        分析告警原因
        采取处理措施

05.运维操作
    a.启动停止
        启动:flume-ng agent命令
        停止:kill -15 <pid>
        优雅停止,等待数据处理完成
    b.配置更新
        修改配置文件
        重启Agent生效
        或使用动态配置更新
    c.数据备份
        备份position文件
        备份checkpoint文件
        定期备份配置文件
    d.故障恢复
        检查日志定位问题
        恢复position文件
        重启Agent

7.6 常见问题

01.数据丢失
    a.原因
        使用Memory Channel且进程崩溃
        Exec Source命令异常退出
        配置错误导致数据被过滤
    b.解决方案
        使用File Channel或Kafka Channel
        使用Taildir Source代替Exec Source
        检查Interceptor和Selector配置
        启用事务机制
    c.预防措施
        选择可靠的组件
        配置合理的重试策略
        监控数据流量
        定期备份关键文件

02.数据重复
    a.原因
        Sink失败后重试
        Agent重启后重新读取数据
        网络抖动导致重复发送
    b.解决方案
        在下游系统去重
        使用唯一ID标识Event
        配置合理的重试策略
    c.说明
        Flume保证At-Least-Once语义
        不保证Exactly-Once
        需要在应用层处理重复

03.性能问题
    a.原因
        Channel容量不足
        批量大小配置过小
        JVM内存不足
        磁盘IO瓶颈
    b.解决方案
        增大Channel容量
        增大批量大小
        增加JVM堆内存
        使用SSD或多个磁盘
        增加Agent实例
    c.优化建议
        监控系统资源使用
        分析性能瓶颈
        逐步优化配置
        压力测试验证

04.内存溢出
    a.原因
        Memory Channel容量过大
        JVM堆内存配置过小
        Event过大
        内存泄漏
    b.解决方案
        减小Memory Channel容量
        增加JVM堆内存
        使用File Channel
        分析堆dump文件
    c.预防措施
        合理配置内存
        监控内存使用
        定期重启Agent
        使用内存分析工具

05.连接失败
    a.原因
        网络不通
        端口被占用
        防火墙阻止
        目标系统不可用
    b.解决方案
        检查网络连通性
        检查端口是否监听
        配置防火墙规则
        检查目标系统状态
        配置重试和故障转移
    c.排查步骤
        ping测试网络
        telnet测试端口
        查看防火墙规则
        查看目标系统日志
        查看Flume日志

06.文件权限问题
    a.原因
        Flume进程没有读取权限
        没有写入HDFS的权限
        position文件权限错误
    b.解决方案
        修改文件权限
        使用正确的用户启动Flume
        配置HDFS权限
        检查目录权限
    c.预防措施
        使用专门的用户运行Flume
        配置合理的文件权限
        定期检查权限配置