flume+kafka+slipstream实现黑名单用户访问实时监测

本是古典何须时尚 2022-05-19 07:29 184阅读 0赞

# 说明 #

之前说过，关于flume和kafka的实践操作就不单独拿出来讲了，打算用一个案例将两者结合，同流式计算一起在具体案例中讲述。

本篇博文案例背景：用户访问系统会留下一条条访问记录（除了姓名和身份证号外，还包括ip地址、登陆地点、设备等一系列详情信息），在数据库中我们有一份黑名单用户数据（姓名和身份证号），现在业务需求是我们需要实时监测访问系统的黑名单用户，将访问信息实时写入数据库中，并在前台展示。

此项目可用于公安系统通缉，比如犯罪嫌疑人在网吧登录、在宾馆登记等等都会留下访问记录（对接公安系统），只要检测到黑名单人员相关信息，便会实时反馈到公安系统，以便实时处理。

# 技术实现方案 #

flume的sink类型指定为kafka类型，采集指定日志目录送至kafka，slipstream的流表对接kafka消费数据。在stream job业务逻辑中编写黑名单比对业务流程，此后流任务启动起来后，每出现一条黑名单人员的访问记录，它便会被实时监测到，并写入结果表中记录下来。

# Flume对接Kafka配置 #

解压flume，编写 flume2kafka.conf 配置文件：

tar -zxvf apache-flume-1.7.0-bin.tar.gz 
    cd apache-flume-1.7.0-bin/conf/
    vi flume2kafka.conf

添加如下配置：

flume2kafka.sources=r3
    flume2kafka.sinks=k3
    flume2kafka.channels=c3
    
    # 配置source
    flume2kafka.sources.r3.type = spooldir
    flume2kafka.sources.r3.spoolDir = /home/jbw/log/
    flume2kafka.sources.r3.fileHeader = true
    
    # 配置channel，将buffer事件放在内存中
    flume2kafka.channels.c3.type = memory
    flume2kafka.channels.c3.capacity = 10000
    flume2kafka.channels.c3.transactionCapacity = 1000
    
    # 配置sink
    flume2kafka.sinks.k3.type=org.apache.flume.sink.kafka.KafkaSink
    flume2kafka.sinks.k3.brokerList=172.18.xxx.xxx:9092,172.18.xxx.xxx:9092,172.18.xxx.xxx:9092
    flume2kafka.sinks.k3.topic=peopleVisitTopic
    flume2kafka.sinks.k3.serializer.class=kafka.serializer.StringEncoder
    flume2kafka.sinks.k3.requiredAcks = 1
    flume2kafka.sinks.k3.batchSize = 20
    
    # 把source和sink绑定在channel上
    flume2kafka.sources.r3.channels=c3
    flume2kafka.sinks.k3.channel=c3

# 创建Kafka Topic #

cd kafka/bin
    # 开启一个kafka的Topic
    ./kafka-topics.sh --create --topic people_visit_topic --zookeeper 172.18.xxx.xxx:2181, 172.18.xxx.xxx:2181, 172.18.xxx.xxx:2181 --partitions 1 --replication-factor 2

![70][]

可以看到名为peopleVisitTopic的topic创建成功，此topic即为flume端将采集到的数据送入的topic，也是stream消费的topic。

（需要注意，kafka的topic命名规范指出，名称中不能含有下划线"\_"或者点"."，以避免名称冲突）

./kafka-topics.sh --list --zookeeper 172.18.xxx.xxx:2181, 172.18.xxx.xxx:2181, 172.18.xxx.xxx:2181

![70 1][]

# 黑名单数据导入数据库，以便后续比对 #

将黑名单数据上传至HDFS，用于后续导入至HBase中进行后续比对。

hadoop fs -mkdir -p /tmp/data_bowen/blacklist
    hadoop fs -put people_visit.log /tmp/data_bowen/blacklist

![70 2][]

看一下blacklist黑名单数据格式：

![70 3][]

根据blacklist黑名单数据格式创建外表，指定HDFS上数据文件路径：

CREATE EXTERNAL TABLE blacklists (
       name string,
       sex string,
       nation string,
       id string)
    row format delimited fields terminated by ','
    LOCATION '/tmp/data_bowen/blacklist';

创建HBase表，并从外表中导入黑名单人员数据。（此黑名单表用于与实时流表作比对，若发现当前访问的人员在黑名单列表里，则会将其插入到最终结果表中）

CREATE TABLE hyper_blacklist(
       name string,
       sex string,
       nation string,
       id string)
    STORED AS HYPERDRIVE;
    
    INSERT INTO hyper_blacklist SELECT * FROM blacklists;

# 创建流表（从kafka中实时获取用户访问数据） #

在此之前，我们先看下用户访问日志记录数据格式，可以发现黑名单用户表只包含了访问日志记录中的部分信息：

![70 4][]

创建流表与kafka对接，flume采集访问记录送到kafka中，在创建并开启流任务后，kafka中来的数据会被传入SlipStream中预先定义好的流表中。如下即为流表的定义。创建并开启流任务的操作在下一步讲述：

CREATE STREAM people_visit_stream
    (name string
      , sex string
      , nation string
      , id string
      , birthdate string
      , phone string
      , email string
      , address string)
    ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
    TBLPROPERTIES("topic" = "peopleVisitTopic"
                  , "kafka.zookeeper" = "172.18.xxx.xxx:2181, 172.18.xxx.xxx:2181, 172.18.xxx.xxx:2181"
                  , "kafka.broker.list" = "172.18.xxx.xxx:9092,172.18.xxx.xxx:9092, 172.18.xxx.xxx:9092");

# 创建结果表 #

结果表用于存放实时监测到的黑名单人员访问记录：

CREATE TABLE blacklst_visit_result (
       name string,
       sex string,
       nation string,
       id string,
       phone string,
       address string)
    STORED AS HYPERDRIVE;

# 创建并开启流任务 #

CREATE STREAMJOB monitor_blacklist AS
    ("INSERT INTO blacklst_visit_result SELECT a.name,a.sex,a.nation,a.id,a.phone,a.address FROM people_visit_stream a JOIN hyper_blacklist b ON a.id == b.id");

开启流任务Job，查看运行中的StreamJob，如下：

START STREAMJOB monitor_blacklist;
    LIST STREAMJOBS;

![70 5][]

此StreamJob所做的任务即将从kafka中得到的访问数据，与黑名单数据进行比对，若发其存在于黑名单中，则将此条访问记录插入到结果表中。

# 准备日志文件，模拟访问数据（待采集的数据） #

开启flume客户端，开始监控指定日志目录 /

# 进入flume目录
    cd /home/apache-flume-1.7.0-bin
    # 运行flume
    bin/flume-ng agent -n flume2kafka -c conf -f /home/apache-flume-1.7.0-bin/conf/flume2kafka.conf 
    -Dflume.root.logger=INFO,console

![70 6][]

往日志目录放people\_visit.log文件，模拟真实访问数据。

mkdir -p /home/jbw/log
    cp people_visit.log jbw/log/

可以看到瞬间采集完毕：

![70 7][]

查询结果表中数据如下：

![70 8][]

以上显示的便是黑名单人员访问的记录了，当然我们还可以加上时间字段等其他信息，更精准地描述黑名单人员的访问记录。

[70]: /images/20220519/76e5899e040f4ebe9b3d4bc921a4984c.png
[70 1]: /images/20220519/92c6121df8974302abcef8d445ba4aa2.png
[70 2]: /images/20220519/ec9411842ed249a9a8541790fc809211.png
[70 3]: /images/20220519/aea63f35936b4505a030912449df5e69.png
[70 4]: /images/20220519/fc4d44f47a604c9cbbe3a90538577336.png
[70 5]: /images/20220519/92a6870493854a469ddc6a91c56ab05e.png
[70 6]: /images/20220519/fd9e752ca0b94cf78e38a5a99df39499.png
[70 7]: /images/20220519/b52a9f01a8d04dd7976e8b908522da45.png
[70 8]: /images/20220519/a454eb2a680c45a7964b05ab7c030b7a.png