如何实现基于Flink实时数据处理

小编给大家分享一下如何实现基于Flink实时数据处理，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！

成都创新互联于2013年开始，先为陆港等服务建站，陆港等地企业，进行企业商务咨询服务。为陆港企业网站制作PC+手机+微官网三网同步一站式服务解决您的所有建站问题。

基于Flink 1.11的网络流量实时解析，主要针对基于Pcap的原始网络流量数据进行TCP/UDP/ICMP的协议数据实时解析，并将解析数据装成数据帧Frame，以便进行实时网络流量分析。

为完成以上功能，需要了解Pcap数据解析、TCP/UDP层协议解析、Flink的序列化和反序列化、Flink自定义函数以及基于Stream sql的Flink实时数据分析。

1、Pcap数据解析

要进行基于Pcap格式的网络流量数据解析，就必须了解Pcap文件格式定义：

如上所示，标准Pcap数据由Pcap文件头、数据桢Frame头、数据桢Frame组成。

在Pcap文件头中：Magic ：0x1A2B3C 4D，用于表示Pcap数据的开始；Major：用于标示Pcap数据主版本号；Minor：用于标示Pcap数据次版本号；ThisZone：本地标准时间；SigFigs: 时间戳精度；SnapLen：最大的存储长度；LinkType：链路类型。

在数据桢头中：Timestamp1：时间戳高位，精确到S；Timestamp2：时间戳低位，精确到ms；CapLen：当前数据桢长度；

Len：网络中实际数据桢的长度。

注意：目前LinkType链路类型，支持EN10MB、RAW、LOOP、LINUX_SLI；通过以上基本结构，在Pcap文件头中，我们获取最有用的信息即时LinkType，后面我们需要根据不同的LinkType类型，进行数据桢Frame的解析。

除此之外，根据数据桢头，可以获得数据桢的封装时间；

这里根据以太网数据桢类型为例：也就是Ipv4、Ipv6、ARP数据桢，如上图所示，该类型的数据桢数据部分的偏移是14。如果是Ipv4或者Ipv6的协议类型，可以解析获取Mac地址。接下来，其实就是解析TCP/IP层的协议。

2、TCP/UDP协议解析

（1）、TCP协议

// 获取TCP头大小tcpOrUdpHeaderSize = getTcpHeaderLength(packetData, ipStart + ipHeaderLen);packet.put(Packet.TCP_HEADER_LENGTH, tcpOrUdpHeaderSize);// Store the sequence and acknowledgement numbers --M// 获取TCP 请求序列号packet.put(Packet.TCP_SEQ, PcapReaderUtil.convertUnsignedInt(packetData, ipStart + ipHeaderLen + PROTOCOL_HEADER_TCP_SEQ_OFFSET));// 获取TCP 确认序列号packet.put(Packet.TCP_ACK, PcapReaderUtil.convertUnsignedInt(packetData, ipStart + ipHeaderLen + PROTOCOL_HEADER_TCP_ACK_OFFSET));// Flags stretch two bytes starting at the TCP header offsetint flags = PcapReaderUtil.convertShort(new byte[] { packetData[ipStart + ipHeaderLen + TCP_HEADER_DATA_OFFSET],packetData[ipStart + ipHeaderLen + TCP_HEADER_DATA_OFFSET + 1] })& 0x1FF; // Filter first 7 bits. First 4 are the data offset and the other 3 reserved for future use.packet.put(Packet.TCP_FLAG_NS, (flags & 0x100) == 0 ? false : true);packet.put(Packet.TCP_FLAG_CWR, (flags & 0x80) == 0 ? false : true);packet.put(Packet.TCP_FLAG_ECE, (flags & 0x40) == 0 ? false : true);packet.put(Packet.TCP_FLAG_URG, (flags & 0x20) == 0 ? false : true);packet.put(Packet.TCP_FLAG_ACK, (flags & 0x10) == 0 ? false : true);packet.put(Packet.TCP_FLAG_PSH, (flags & 0x8) == 0 ? false : true);packet.put(Packet.TCP_FLAG_RST, (flags & 0x4) == 0 ? false : true);packet.put(Packet.TCP_FLAG_SYN, (flags & 0x2) == 0 ? false : true);packet.put(Packet.TCP_FLAG_FIN, (flags & 0x1) == 0 ? false : true);

2、UDP协议

tcpOrUdpHeaderSize = UDP_HEADER_SIZE;if (ipProtocolHeaderVersion == 4) {int cksum = getUdpChecksum(packetData, ipStart, ipHeaderLen);if (cksum >= 0)packet.put(Packet.UDP_SUM, cksum);}int udpLen = getUdpLength(packetData, ipStart, ipHeaderLen);packet.put(Packet.UDP_LENGTH, udpLen);

3、Kafka的序列化和反序列化

基于分布式消息队列Kafka作为网络流量数据的中间临时缓存，通过FlinkKafkaConsumer进行网络流数据的解析，这里我们自定义了PcapResover的解析器，使用自定义的解序列化函数PcapDataDeSerializer。

Kafka Producer，负责转发已采集的网络流量，这里配置使用了Kafka内部的序列化类

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());this.consumer = new FlinkKafkaConsumer<>(this.topic,(KafkaDeserializationSchema)new PcapDataDeSerializer(Object.class),props);public class PcapDataDeSerializer implements KafkaDeserializationSchema