场景说明

场景说明

假定用户有某个周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Spark应用程序实现如下功能：

• 实时统计连续网购时间超过半个小时的女性网民信息。
• 周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为“,”。

  log1.txt：周六网民停留日志

  LiuYang,female,20
  YuanJing,male,10
  GuoYijun,male,5
  CaiXuyu,female,50
  Liyuan,male,20
  FangBo,female,50
  LiuYang,female,20
  YuanJing,male,10
  GuoYijun,male,50
  CaiXuyu,female,50
  FangBo,female,60

  log2.txt：周日网民停留日志

  LiuYang,female,20
  YuanJing,male,10
  CaiXuyu,female,50
  FangBo,female,50
  GuoYijun,male,5
  CaiXuyu,female,50
  Liyuan,male,20
  CaiXuyu,female,50
  FangBo,female,50
  LiuYang,female,20
  YuanJing,male,10
  FangBo,female,50
  GuoYijun,male,50
  CaiXuyu,female,50
  FangBo,female,60

数据规划

Spark Streaming样例工程的数据存储在Kafka组件中（需要有Kafka权限用户）。

1. 本地新建两个文本文件input_data1.txt和input_data2.txt，将log1.txt的内容复制保存到input_data1.txt，将log2.txt的内容复制保存到input_data2.txt。
2. 在客户端安装节点下创建文件目录：“/home/data”。将上述两个文件上传到此“/home/data”目录下。
3. 将Kafka的Broker配置参数“allow.everyone.if.no.acl.found”值设置为“true”(普通集群不需配置)。
4. 启动样例代码的Producer，向Kafka发送数据。

   java -cp $SPARK_HOME/jars/*:$SPARK_HOME/jars/streamingClient/*:{JAR_PATH} com.huawei.bigdata.spark.examples.StreamingExampleProducer {BrokerList} {Topic}

   

   • JAR_PATH为程序jar包所在路径。
   • brokerlist格式为brokerIp:9092。

开发思路

统计日志文件中本周末网购停留总时间超过半个小时的女性网民信息。

主要分为四个部分：

• 接收Kafka中数据，生成相应DStream。
• 筛选女性网民上网时间数据信息。
• 汇总在一个时间窗口内每个女性上网时间。
• 筛选连续上网时间超过阈值的用户，并获取结果。