任务10.2 使用Flink实现实时计算农村电商数据

使用Flink实现实时计算农村电商数据任务描述部署好Flink集群后，小数还需要通过Scala语言在Flink中进行农村电子商务商品数据的实时计算，包括统计每日销售额、每日访问次数、每日每家门店总销售额、每日商品销量等。任务环境已部署Hadoop集群、ZooKeeper集群、Kafka集群和Flink集群（独立集群模式），且master节点已部署MySQL和Flume的3台CentOS7操作系统的虚拟机。任务描述与任务环境农村电商网站的商品信息和用户信息数据的相关字段及字段说明如下。数据字段说明编号字段字段说明1rowkey行键2userId用户名3age年龄4gender性别5goods_no商品ID6goods_price价格7store_id门店ID8shopping_type购买行为9tel电话号码10shopping_date购买日期11email邮箱目录1搭建Flink开发环境采集农村电子商务商品数据2创建MySQL存储表3统计每日销售额和每日访问次数4采集农村电子商务商品数据的实现流程如下。采集农村电子商务商品数据依次启动ZooKeeper集群、Kafka集群、Flink集群。在所有节点执行。在master节点执行。创建Kafka主题在master节点创建shop主题。若要创建的主题已存在。可先删除主题，然后再执行创建命令创建主题shop，确保主题不包含任何可能的干扰性内容。创建Kafka主题在master节点中，执行“vim$FLUME_HOME/conf/spool-memory-kafka.conf”命令编写Flume数据采集方案spool-memory-kafka.conf。编写Flume采集方案在slave1节点上启动消费者。启动消费者通过Xftp文件传输工具，将data.txt文件上传至master节点的/opt/data目录下。在master节点，使用“vim/opt/createData.sh”命令创建抽取数据脚本。使用“mkdir/opt/flinkproject”命令，在/opt目录下创建flinkproject文件夹，用于存放抽取后的数据文件。数据采集准备工作目录1搭建Flink开发环境采集农村电子商务商品数据2创建MySQL存储表3统计每日销售额和每日访问次数4Flink开发环境的搭建流程如下。搭建Flink开发环境IntelliJScala插件是为Scala语言开发的一个工具，它提供了许多有用的功能和工具，可以帮助开发人员更高效地编写和调试Scala代码。打开IDEA，单击“Plugins”选项，在Marketplace页面下的搜索框中输入“scala”。单击“Install”按钮进行下载安装Scala插件。安装Scala插件下载好后单击“RestartIDE”按钮，重启IDEA。安装Scala插件在IDEA启动初始界面中，单击“+”图标、“NewProject”选项，弹出“NewProject”对话框。在左侧列表栏中选择“MavenArchetype”，填写工程的名称并选择工程保存的位置，选择JDK的版本为“1.8”，“Archetype”选择“org.apache.maven.archetypes:maven-archetype-quickstart”。创建Flink工程Flink工程创建完成后，在IntelliJIDEA的主界面中可以查看该工程的目录结构。创建Flink工程修改pom文件，添加依赖。添加Flink依赖配置完pom.xml文件后，需要单击右侧边栏的“Maven”，同时单击刷新图标重新加载所有的Maven项目所需的依赖包。Junit是一个开源的Java测试框架，用于编写和运行单元测试。主要是使用Scala语言实现Flink实时统计功能，由于Junit主要是用于Java环境的单元测试，因此Junit并非是此Flink工程所必需的依赖项。实际上，在创建Maven工程时，Junit可能是由于默认配置而自动引入的。如果用户不打算使用Java单元测试，或项目完全是基于Scala的，那么完全可以选择将Junit依赖从项目中移除，以保持依赖关系的简洁和相关性。添加Flink依赖在工程的src下面，鼠标右键“main”，依次选择“New”→“Directory”。输入“scala”，单击Enter键完成创建。鼠标右键“scala”，依次选择“MarkDirectoryas”→“SourceRoot”，将其改成源码包。创建scala目录选择菜单栏中的“File”→“ProjectStructure”→“Libraries”→“+”→“ScalaSDK”选项，添加SDK设置。添加Scala支持在弹出的对话框中，单击“Download”按钮。选定版本为“2.12.6”，单击“OK”按钮。添加Scala支持下载完成后，再次单击“OK”按钮。添加Scala依赖鼠标右键单击main文件夹，依次选择“New”→“Directory”。在弹出的“NewDirectory”窗口中，单击下方的“resources”选项，单击Enter键完成创建。配置Log4j日志文件单击main/resources文件夹，依次选择“New”→“File”，新建文件，命名为“perties”。从$FLINK_HOME/conf/perties中复制文本内容，粘贴到perties文件。配置Log4j日志文件目录1搭建Flink开发环境采集农村电子商务商品数据2创建MySQL存储表3统计每日销售额和每日访问次数4为了高效地存储和查询数据，将每日访问次数、每日销售额的统计结果存储在MySQL中。所需的MySQL数据表结构说明下。创建MySQL存储表表名表说明字段字段说明字段类型visitcount_everyday每日访问次数sale_date销售日期varcharvisitcount用户访问次数intsalevolume每日销售额sale_date销售日期varcharsalevolume销售量double启动MySQL。创建MySQL存储表创建shop数据库。在shop数据库下，创建visitcount_everyday和salevolume表。创建MySQL存储表执行后使用“showtables;”查看数据表。注意：操作完成后建议不要关闭此终端，涉及终端其他操作则另开一个终端进行。创建MySQL存储表目录1搭建Flink开发环境采集农村电子商务商品数据2创建MySQL存储表3统计每日销售额和每日访问次数4统计每日销售额和每日访问次数的整体流程如下。统计每日销售额和每日访问次数鼠标右键scala目录，依次选择“New”→“Package”，在弹出的窗口输入“mysqlsink”，单击Enter键完成创建mysqlsink包。鼠标右键mysqlsink包，依次选择“New”→“ScalaClass”，在弹出的窗口输入“SaleSQLSink”，单击Enter键完成创建SaleSQLSink类，由于此定义类非用于执行程序，故默认选择“Class”类型即可，自定义MySQLSinkSaleSQLSink类定义代码如下。自定义MySQLSink在scala工程下创建objectclass包，包下创建StringAndDouble类，同样默认选择“Class”类型。自定义MySQLSink在scala工程下创建flinkclass包，包下创建SaleVolumn类，因此类用于本地运行，故设置为“Object”类型。编写实时数据处理程序编写实时数据处理程序在master节点上，通过“crontab-e”命令，创建定时任务。在定时任务框中编写脚本“*****sh/opt/createData.