9.2 广告日志数据实时传输

广告日志数据实时传输主讲：李强任务描述Flume作为日志收集系统，可以从不同的数据源将数据源源不断收集，但Flume不会持久地保存数据，需要使用Sink将数据存储到外部存储系统，如HDFS、HBase、Kafka等。Flume与HDFS、HBase的结合一般用于离线批处理。而Flume与Kafka的整合一般用于数据实时流处理，通过Flume的Agent代理收集日志数据，再由Flume的Sink将数据传送到Kafka集群，完成数据的生产流程，最后交给Storm、Flink、SparkStreaming等进行实时消费计算。本案例将基于项目8中的case_data_new.csv广告日志数据，使用Flume和Kafka整合实现广告日志数据的实时传输。首先用脚本模拟实时生成的日志数据并存入MySQL，再使用Flume实时监视MySQL中增加的数据，采集到Kafka集群的主题中，并启动消费者消费主题数据，最终实现数据的实时传输。任务分析广告日志数据的实时传输的实现步骤如下。（1）脚本定时抽取数据到指定目录，模拟系统日志产生，并将其存入MySQL数据表中。（2）创建Kafka主题，开启消费者以消费数据。（3）编写conf采集配置文件，将存入MySQL数据库中的数据传入Kafka主题。9.2.1创建脚本文件创建脚本文件在master节点下运行“mysql-uroot-pPassword123$”进入MySQL数据库中，创建数据库kafka，并在kafka下创建表用于存储数据，如代码9-13所示。代码9-13创建数据表createdatabasekafka;usekafka;createtablecase_data(`rank`int,dtint,cookievarchar(200),ipvarchar(200),idfavarchar(200),imeivarchar(200),androidvarchar(200),openudidvarchar(200),macvarchar(200),......省略创建脚本文件打开一个新的master节点终端窗口，运行“vi/data/datamysql.sh”创建一个脚本文件，如代码9-14所示。脚本内容为一个whiletrue循环，每分钟在case_data_new.csv随机提取100条数据，存入“/data/datamysql/mysqltmp.txt”文件中，再将文件中的数据存入MySQL数据库中。代码9-14脚本datamysql.sh#!/bin/bashwhiletruedotime=$(date"+%Y%m%d_%H%M%S")shuf-n100/opt/case_data_new.csv>/data/datamysql/mysqltmp.txtmysql-uroot-pPassword123$--local-infile-e"useKafka;loaddatalocalinfile'/data/datamysql/mysqltmp.txt'intotablecase_datafieldsterminatedby','OPTIONALLYENCLOSEDBY'\"';"sleep60done创建脚本文件脚本创建完成后，赋予脚本权限，然后将其启动，如代码9-15所示。启动成功后可能会发出如图9-10所示的警报信息，表示在命令行中直接输入密码账户信息是不安全的，该警报信息是在MySQL5.6版本后有的，并不影响运行结果，可以选择忽视。代码9-15关于脚本的命令//脚本权限chmod777/data/data2mysql.sh//脚本启动命令sh/data/datamysql.sh&//脚本中断命令psaux|grep"datamysql.sh"|grep-vgrep|cut-c9-15|xargskill-9图9-10执行脚本文件创建脚本文件在MySQL数据库中查看数据是否成功存入表中，如代码9-16所示。结果如图9-11所示。可以看出已经有数据存入表中，并正在实时更新中。代码9-16查看数据是否存入//进入数据库kafkausekafka;//查看表中有几行selectcount(*)fromcase_data;图9-11数据已存入9.2.2创建Kafka主题创建Kafka主题分别在slave1、slave2中开启ZooKeeper、Kafka集群。在slave1节点创建一个Kafka主题RealTime，设置2个副本，2个分区。创建成功后，开启消费者消费，如代码9-17所示。目前，该消费者并没有在指定主题中消费到数据。代码9-17创建Kafka主题并开启消费//创建RealTime主题kafka-topics.sh-create--topicRealTime--bootstrap-serverslave1:9092,slave2:9092--partitions2--replication-factor2//开启消费者kafka-console-consumer.sh--topicRealTime--bootstrap-serverslave1:9092,slave2:90929.2.3Flume采集日志Flume采集日志在Flume的conf目录下创建一个“datamysql.conf”文件，实现从MySQL中采集数据，并传入RealTime主题中，如代码9-18所示。代码9-18Flume脚本datamysql.confagent.sources=sql-sourceagent.sinks=k1agent.channels=chagent.sources.sql-source.type=org.keedio.flume.source.SQLSourceagent.sources.sql-source.hibernate.connection.url=jdbc:mysql://192.168.88.181:3306/kafka?&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=GMTagent.sources.sql-source.hibernate.connection.user=rootagent.sources.sql-source.hibernate.connection.password=Password123$agent.sources.sql-source.hibernate.dialect=org.hibernate.dialect.MySQLDialectagent.sources.sql-source.hibernate.driver_class=com.mysql.cj.jdbc.Driveragent.sources.sql-source.hibernate.connection.autocommit=true......省略Flume采集日志启动FlumeAgent开始采集MySQL数据库中的数据，如代码9-19所示。切换到Kafka消费者的终端，可以看到主题上已经有数据被消费者消费，如图9-12所示。观察消费者终端，可以看到消费者每过一分钟，就会有新数据消费，因为脚本文件一直在模拟用户产生数据，而Flume在实时采集并传入到Kafka主题上。至此，Flume和Kafka的整合实现实时传输已经完成。在现实应用中，一般实时传输将应用于实时计算的场景，因此，Flume与Kafka还可以与实时计算框架，如SparkStreaming