Spark:连续三天登录的用户

package Restartimport java.text.SimpleDateFormat
import java.util.{Calendar, Date}import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
//查询连续三天登录的用户
object DataOne {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[*]")val sc = new SparkContext(conf)val lines = sc.textFile("C:\\Users\\admin\\Desktop\\data1.txt")val uidAndData: RDD[(String, String)] = lines.map(i => {val strings = i.split(",")val uid = strings(0)val data = strings(1)(uid, data)})val groupUid: RDD[(String, Iterable[String])] = uidAndData.groupByKey()//分组ArrayBuffer((guid01,CompactBuffer(2018-02-28, 2018-03-01, 2018-03-02, 2018-03-04, 2018-03-05, 2018-03-06, 2018-03-07)),// (guid02,CompactBuffer(2018-03-01, 2018-03-02, 2018-03-03, 2018-03-06)))//定义一个index,如果2018-02-28-0  = 2018-03-01-1= 2018-03-02-2......依次类推连续相等3天,就满足条件//本题难点就在于,怎么让时间连续相减val uidAndDate: RDD[(String, (String, String))] = groupUid.flatMapValues(it => {val sorted = it.toSet.toList.sorted //去重,变为数组val sdf = new SimpleDateFormat("yyyy-MM-dd")val calendar = Calendar.getInstance() //获得一个索引var index = 0sorted.map(it => {val date: Date = sdf.parse(it) //将value的时间转为时间戳calendar.setTime(date) //获取时间calendar.add(Calendar.DATE, -index) //时间  负的-index.index += 1(it, sdf.format(calendar.getTime)) //将时间戳,转化为正常时间})})//ArrayBuffer((guid01,(2018-02-28,2018-02-28)), (guid01,(2018-03-01,2018-02-28)), (guid01,(2018-03-02,2018-02-28)), (guid01,(2018-03-04,2018-03-01)), (guid01,(2018-03-05,2018-03-01)), (guid01,(2018-03-06,2018-03-01)), (guid01,(2018-03-07,2018-03-01)),// (guid02,(2018-03-01,2018-03-01)), (guid02,(2018-03-02,2018-03-01)), (guid02,(2018-03-03,2018-03-01)), (guid02,(2018-03-06,2018-03-03)))//如何处理这个时间呢,我们观察,(2018-02-28,2018-02-28) key-0=2018-02-28 没错//2018-03-01  -1=2018-02-28没错,数据是对的那么怎么获取起始时间和结束时间呢//我们将value._2取出来,和guid组成一个元组,在分组,那么只要是连续的就会在一个分区//然后得到---ArrayBuffer(((guid02,2018-03-01),CompactBuffer(2018-03-01, 2018-03-02, 2018-03-03)),// ((guid01,2018-03-01),CompactBuffer(2018-03-04, 2018-03-05, 2018-03-06, 2018-03-07)), // ((guid01,2018-02-28),CompactBuffer(2018-02-28, 2018-03-01, 2018-03-02)),// ((guid02,2018-03-03),CompactBuffer(2018-03-06)))//我们对value进行处理,value.head就是起始时间,last结束时间,size是连续几天//处理后的结果//ArrayBuffer(((guid02,2018-03-01),(3,2018-03-01,2018-03-03)), ((guid01,2018-03-01),(4,2018-03-04,2018-03-07)), ((guid01,2018-02-28),(3,2018-02-28,2018-03-02)), ((guid02,2018-03-03),(1,2018-03-06,2018-03-06)))//余下的工作只需要map将uid和起始时间,结束时间连续天数进行过滤就可以得到需求val value: RDD[(String, String, String, Int)] = uidAndDate.map(it => ((it._1, it._2._2), it._2._1)).groupByKey().mapValues(it => {val sorted = it.toList.sortedval head = sorted.headval last = sorted.lastval size = sorted.size(head, last, size)}).filter(i=>i._2._3>=3).map(it => {(it._1._1, it._2._1, it._2._2, it._2._3)})
//ArrayBuffer((guid02,2018-03-01,2018-03-03,3), (guid01,2018-03-04,2018-03-07,4), (guid01,2018-02-28,2018-03-02,3))println(uidAndDate.collect().toBuffer)
sc.stop()}}