-- 这个问题可以扩展到很多相似的问题：连续几个月充值会员、连续天数有商品卖出、连续打滴滴、连续逾期。
数据如下:
guid01,2018-02-28
guid01,2018-03-01
guid01,2018-03-02
guid01,2018-03-04
guid01,2018-03-05
guid01,2018-03-06
guid01,2018-03-07
guid02,2018-03-01
guid02,2018-03-02
guid02,2018-03-03
guid02,2018-03-06
uid| 连续登录天数 | 起始日期 | 结束日期 |
±--------±-------±------------±------------±-+
| guid01 | 4 | 2018-03-04 | 2018-03-07 |
| guid02 | 3 | 2018-03-01 | 2018-03-03 |
def main(args: Array[String]): Unit = {val isLocal = args(0).toBooleanval conf = new SparkConf().setAppName(this.getClass.getSimpleName)if(isLocal){conf.setMaster("local[*]")}val sc = new SparkContext(conf)//指定从哪里读取数据,并创建RDDval lines: RDD[String] = sc.textFile(args(1))val uidAndDate: RDD[(String, String)] = lines.map(line => {val fields: Array[String] = line.split(",")val uid = fields(0)val date = fields(1)(uid, date)})//根据uid进行分组,将同一个用户的登录数据,搞到同一个组内val grouped: RDD[(String, Iterable[String])] = uidAndDate.groupByKey()//在组内进行排序val uidAndDateDatedif = grouped.flatMapValues(it => {//toSet可以将日期去重,再tolist,进行排序//留意如果数据特别大,toset 或者toList 可能会内存溢出val sorted = it.toSet.toList.sorted//定义一个日期的工具类val calendar = Calendar.getInstance()val sdf = new SimpleDateFormat("yyyy-MM-dd")var index = 0sorted.map(dateStr => {val date = sdf.parse(dateStr)calendar.setTime(date)val difdate = calendar.add(Calendar.DATE, -index)index += 1(dateStr, sdf.format(calendar.getTime))})})//println(uidAndDateDatedif.collect().toBuffer)val result = uidAndDateDatedif.map(t => {((t._1, t._2._2), t._2._1)}).groupByKey().mapValues(it => {val list: List[String] = it.toList.sortedval times = list.sizeval begintime = list.headval endtime = list.last(times, begintime, endtime)}).filter(t=>{t._2._1>=3}).map(t => {(t._1._1, t._2._1, t._2._2, t._2._3)})val res = result.collect()
println(res.toBuffer)sc.stop()}