《Flink应用开发技术》课件-项目2：DataStream API基础编程（流处理核心）

项目2-1DatastreamAPI初步使用1目录CONTENTS01.Flume概述02.Flume组件03.Flume安装04.Kafka基础05.Kafka命令2目录CONTENTS01.Shell脚本02.任务实现301Flume概述4Flume：海量日志收集的利器分布式、可靠、高效的日志收集系统分布式架构基于流式架构，实时处理数据流，可水平扩展应对海量数据。可靠传输提供多种容错与恢复机制，确保数据在传输过程中不丢失。无缝对接Hadoop、Kafka、HBase等下游生态，构建统一数据通道5Flume的架构演进之路FlumeOG(OriginalGeneration)集中式架构，扩展性与容错能力有限。架构跃迁FlumeNG(NextGeneration)分布式架构，增强扩展性、容错与资源隔离，成为Apache顶级项目。6Flume核心概念模型Event数据单元，由Header和Body组成。FlowEvent从源点到目的点的迁移抽象。Agent独立进程，是包含S/C/S的Flume基本运行单元。702Flume组件8Source:数据收集组件负责对接各种外部数据源，是Flume数据管道的入口，决定了数据的来源和初步格式。AvroSource监听Avro端口，接收RPC调用数据。ExecSource执行Unix命令（如tail），采集其标准输出。SpoolingDirectory监听指定目录，自动采集新增文件。NetCatSource监听TCP/UDP端口，将每行文本转为Event。9Channel:临时存储与缓冲连接Source和Sink的桥梁，提供数据缓冲和容错能力，是保障数据可靠性的关键。Memory速度快，但数据易失。适用于对性能要求高、可容忍数据丢失的场景。File持久化存储，可靠性高。适用于对数据安全要求高的场景。JDBC支持事务，可靠性最高。适用于对数据一致性要求极高的场景。Kafka高吞吐，兼具性能与可靠。适用于大数据量、高并发场景。10Sink:数据输出组件HDFSSink写入HDFS，支持按时间/大小滚动。LoggerSink输出到日志，用于调试。KafkaSink转发到Kafka，实现多级流处理。HBaseSink写入HBase，支持列映射。1103Flume安装12Flume1.9.0快速安装指南1解压安装包tar-zxvf...2配置环境变量FLUME_HOME,PATH3修改flume-env.sh设置JAVA_HOME4验证安装flume-ngversion13最小化测试：NetCat+Logger通过最简配置验证Flume安装是否成功，并直观感受数据流处理过程。1.配置Agent:使用netcatsource监听端口，loggersink输出到控制台。2.启动Agent:运行Flumeagent进程。3.发送数据:使用telnet或nc命令连接端口并输入文本。4.观察结果:在Agent控制台查看输出的Event信息。1404Kafka基础15为何引入Kafka？解耦削峰通过队列模式将生产者和消费者分离，有效应对数据流量峰值，避免系统过载。高吞吐能力采用顺序写磁盘和分区并行技术，实现极高的数据吞吐量，专为大数据场景设计。ProducerBrokerZookeeperConsumerGroupConsumerTopic&Partition16Kafka工作流程：发布与订阅Producer发布消息Topic(Partition)消息按Topic分类，以Partition为单位追加写入Log文件Consumer订阅并拉取Offset机制：Consumer通过记录Offset来控制消费进度，实现消息的按需重放和故障恢复。1705Kafka命令18Kafka常用命令行速览#创建Topickafka-topics.sh--create--topicmy_topic--bootstrap-serverlocalhost:9092#查看Topic列表kafka-topics.sh--list--bootstrap-serverlocalhost:9092#启动生产者kafka-console-producer.sh--topicmy_topic--bootstrap-serverlocalhost:9092#启动消费者kafka-console-consumer.sh--topicmy_topic--from-beginning--bootstrap-serverlocalhost:90921906Shell脚本20Shell脚本：自动化基石通过命令行接口编写脚本，实现任务自动化，是数据工程师的必备技能。变量定义与控制循环(if/for/while)输入输出与文件操作(read/echo)管道与重定向(|,>,>>)例2.10:按行读取文件whilereadlinedoecho"$line"done<file.txt例2.11:写入文件并加时间戳whilereadlinedoecho"$(date):$line">>log.txtdoneShell脚本实例：按行读写2107任务实现22任务实现：数据准备使用阿里天池用户行为数据集，通过Shell脚本构建数据源。1.数据集(UserBehavior.csv)包含用户ID、商品ID、行为类型(pv,buy,cart,fav)等字段。2.编写Shell脚本使用whileread循环按行读取CSV，并追加到file.log。23任务实现：Flume对接Kafkafile.logFlumeAgentSource:tail-ffile.logChannel:memorySink:KafkaKafkaTopic配置FlumeAgent，实现从file.log实时采集数据并发送至Kafka。24任务实现：Flink消费程序编写Flink程序，从Kafka中消费用户行为数据流并进行处理。添加依赖:flink-connector-kafka设置Kafka消费者属性(bootstrap.servers,group.id)创建FlinkKafkaConsumer添加数据源并执行作业(print/业务逻辑)25任务实现：测试与注意事项测试步骤启动Zookeeper服务启动Kafka服务启动FlumeAgent运行Shell脚本抽取数据运行Flink程序观察输出注意事项确保Flume配置正确，实现实时采集。Shell脚本需处理数据去重，避免重复消费。Kafka配置需保证消息不丢失。数据量大，脚本运行时间可能较长。多次调试可能导致Kafka中数据重复。26项目2-2DataStreamAPI实战27目录CONTENTS01.核心转换算子02.Sink体系与连接器03.HBase环境部署04.数据流处理流程05.HBase数据存储28目录CONTENTS01.系统集成测试02.技术要点回顾2901核心转换算子30Map一对一转换DataStream最基础的转换操作，实现“一条进，一条出”的数据处理，并可通过RichMapFunction扩展生命周期管理。数据格式转换如JSON字符串转为Java对象，为后续操作提供结构化数据。字段提取与计算从原始数据中提取关键字段或进行衍生计算。数据清洗与格式化过滤无效数据，统一数据格式，提升数据质量。31数据精炼：过滤与展开Filter过滤基于条件判断，返回布尔值决定数据是否保留，是数据清洗的关键步骤。valevenNumbers:DataStream[Int]=numbers.filter(_%2==0)RichFilterFunction支持计数器统计和生命周期管理。FlatMap扁平映射实现“一对多”的数据转换，通过Collector收集多个输出结果。logEntries.flatMap{entry=>entry.activities.split(",").map{activity=>(entry.userId,activity.trim)}典型应用：句子拆词、数据展平。32分流机制的演进从类型耦合到灵活多路输出Split+Select早期方案，因类型耦合已弃用。SideOutputs(推荐)支持多类型输出，主流与侧流数据类型可不同，为异常、日志等提供独立通道。33合流策略：同型合并vs异型协同Union将多个数据类型完全一致的流进行物理拼接，输出类型不变。适用场景：同源数据流扩容，简单合并处理。Connect连接两个不同类型的流，输出ConnectedStreams，支持状态共享。适用场景：规则流与数据流匹配、双流Join。分组聚合与极值计算KeyBy按Key逻辑分区，生成KeyedStreamReduce/Aggregate在分组或窗口内进行增量聚合计算Min/Max仅计算极值，保留首个记录的其他字段。MinBy/MaxBy计算极值，替换为完整极值记录。3402Sink体系与连接器35RedisSink:高速缓存层利用Redis内存数据库，实现计算结果的低延迟、高并发写入，为下游提供快速查询能力。内存存储:极致读写性能，支持多种数据结构(String,Hash,List等)。集成要点:添加依赖、配置连接、实现RedisMapper接口。MySQLSink:可靠的事务写入通过继承RichSinkFunction，在生命周期方法中

管理数据库连接，实现高效、可靠的数据落地。open()建立数据库连接invoke()执行批量写入close()释放连接资源36HBaseSink:分布式NoSQL存储面向列存储的分布式数据库，为Flink提供高吞吐量、可扩展的数据Sink方案。分布式架构:高可用，水平扩展高吞吐写入:适合大数据量实时写入面向列存储:高效压缩与查询数据模型:表→行→列族TableRowColumnFamily每行由唯一行键(RowKey)标识，数据按时间戳存储多版本。3703HBase环境部署38HBase2.2.3伪分布式部署1环境准备配置JDK,Hadoop,Zookeeper2修改配置hbase-env.sh,hbase-site.xml3启动服务先HDFS，再HBase4验证部署WebUI(16010)查看进程核心配置项hbase.rootdir(HDFS路径)hbase.zookeeper.quorum(ZK地址)3904数据流处理流程40数据流处理流程(一):模型与过滤Event(原始数据)userId:LongproductId:LongcategoryId:Longaction:Stringtimestamp:Long解析&过滤(action="pv")ProductClickCount(结果)productId:Longcount:LongwindowStart:StringwindowEnd:String核心操作:Map(解析)→Filter(action="pv")→KeyBy(productId)41数据流处理流程(二):时间语义与窗口事件时间提取`timestamp`字段+水印策略允许3秒乱序→滚动窗口60分钟，无重叠窗口聚合结果封装为ProductClickCount对象，包含窗口起止时间。4205HBase数据存储43HBase表结构设计表名:default:result列族:info(单列族设计，紧凑高效)行键(RowKey):UUID(全局唯一，避免热点)字段映射(String类型)productIdcountwindowStartwindowEnd统一用String类型解决HBase原生Long字节数组显示问题。44HBaseSink实现：生命周期管理open()建立HBase连接(Connection)invoke()组装Put对象并执行写表操作close()释放连接资源，防止泄漏关键优化类型处理:将Long型count转为String存储，解决HBase显示问题。行键设计:使用UUID作为RowKey，确保全局唯一性，避免写入热点。异常处理:增加重试机制，防止数据丢失，保证端到端一致性。45数据验证与一致性检查Shell命令行验证使用HBaseShell的`scan'result'`命令，直观查看表中的字符串数据，确认写入成功。>scan'result'ROWCOLUMN+CELL...程序读取验证通过代码读取数据，使用`Bytes.toLong()`等方法转换，校验数据准确性。byte[]val=result.getValue(...);longcount=Byt