快捷導(dǎo)航

Hive實(shí)現(xiàn)連續(xù)N天登陸語法實(shí)例代碼

更新時(shí)間：2026年04月04日 08:54:09 作者：小小草臺班子

在用戶行為數(shù)據(jù)分析中,常常需要分析用戶的連續(xù)登錄行為,下面這篇文章主要介紹了Hive實(shí)現(xiàn)連續(xù)N天登陸語法的相關(guān)資料,文中通過代碼介紹的非常詳細(xì),需要的朋友可以參考下

Sql方式實(shí)現(xiàn)連續(xù)N天登陸

構(gòu)造測試數(shù)據(jù)

create table dwd.login_log as
select 1 as user_id, "2020-01-01" as login_date
union all
select 1 as user_id, "2020-01-02" as login_date
union all
select 1 as user_id, "2020-01-07" as login_date
union all
select 1 as user_id, "2020-01-08" as login_date
union all
select 1 as user_id, "2020-01-09" as login_date
union all
select 1 as user_id, "2020-01-10" as login_date
union all
select 2 as user_id, "2020-01-01" as login_date
union all
select 2 as user_id, "2020-01-02" as login_date
union all
select 2 as user_id, "2020-01-04" as login_date

如果日期格式不規(guī)范，可以將其轉(zhuǎn)換為標(biāo)準(zhǔn)格式

create table dwd.login_log as
select user_id,to_date(from_unixtime(UNIX_TIMESTAMP(login_date,'yyyy-MM-dd'))) as login_date
from tmp.login_log; -- tmp庫為原始數(shù)據(jù)

1.使用lag&lead+datediff窗口函數(shù)

比如求連續(xù)三天登陸，可以將當(dāng)天上一條數(shù)據(jù)和下一條數(shù)據(jù)都拿到，然后保證now-lag=lead-now=1即可;
如果是連續(xù)多天，可以取更多的數(shù)據(jù)，或者將數(shù)據(jù)全部更改為lag或者lead函數(shù);
datediff(date1, date2) - Returns the number of days between date1 and date2

select user_id 
from 
  (select user_id
  from
      (select user_id,
            lag(login_date,1) over(partition by user_id order by login_date) as lag_login_date,
            login_date,
            lead(login_date,1) over(partition by user_id order by login_date) as lead_login_date
      from dwd.login_log)t1
  where datediff(login_date,lag_login_date)=1 and datediff(lead_login_date,login_date)=1)t2
group by user_id;

2.使用date_add函數(shù)

通用的，先對user_id分區(qū)排序，然后將日期減去rank天，查看有多少條數(shù)據(jù)即可;
優(yōu)點(diǎn)在于可以統(tǒng)計(jì)具體連續(xù)登陸多少天,以及連續(xù)登陸的實(shí)際情況；
date_add(start_date, num_days) - Returns the date that is num_days after start_date

select user_id,con_login_date,count(*) nums
from
    (select user_id,login_date,rk,date_add(login_date,1 - rk) as con_login_date
    from 
        (select user_id,login_date,rank() over(partition by user_id order by login_date) rk
        from dwd.login_log)t1
    )t2
group by user_id,con_login_date
having count(*) >= 3;

t1表的查詢結(jié)果

用戶id	登陸時(shí)間	按照登陸時(shí)間組內(nèi)排序
1	2020-01-01	1
1	2020-01-02	2
1	2020-01-07	3
1	2020-01-08	4
1	2020-01-09	5
1	2020-01-10	6
2	2020-01-01	1
2	2020-01-02	2
2	2020-01-04	3

t2表的查詢結(jié)果，歸一化的日期(也就是上述取前1 - rk)可以自己定義

用戶id	登陸時(shí)間	連續(xù)登陸的日期歸一化的日期
1	2020-01-01	2020-01-01
1	2020-01-02	2020-01-01
1	2020-01-07	2020-01-05
1	2020-01-08	2020-01-05
1	2020-01-09	2020-01-05
1	2020-01-10	2020-01-05
2	2020-01-1	2020-01-01
2	2020-01-2	2020-01-01
2	2020-01-4	2020-01-02

group by后的查詢結(jié)果，第三列可以按照session內(nèi)統(tǒng)計(jì)來理解，就是這批連續(xù)登陸內(nèi)連續(xù)登陸的天數(shù)

用戶id	連續(xù)登陸的日期歸一化的日期	用戶此次連續(xù)登陸天數(shù)
1	2020-01-01	2
1	2020-01-05	4
2	2020-01-01	2
2	2020-01-02	1

代碼實(shí)現(xiàn)思路

使用代碼來實(shí)現(xiàn)連續(xù)N天登陸，核心邏輯就是按照日期排序，新日期如果和舊日期相差1天就保留在HashMap里面，Size超過N即可輸出user_id,否則清空

package cn.lang.spark_core
import java.text.{ParseException, SimpleDateFormat}
import java.util.Calendar
import org.apache.spark.sql.SparkSession
object ContinuousLoginDays {
  def main(args: Array[String]): Unit = {
    // env
    val spark: SparkSession = SparkSession
      .builder()
      .appName("ContinuousLoginDays")
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    // source,可以是load hive(開啟hive支持)或者parquet列式文件(定義好schema)
    val source = sc.textFile("/user/hive/warehouse/dwd/login_log")
    case class Login(uid: Int, loginTime: String) // 可以kryo序列化
    /** get date last `abs(n)` days defore or after biz_date   *
     * example biz_date = 20200101 ,last_n = 1,return 20191231 */
    def getLastNDate(biz_date: String,
                     date_format: String = "yyyyMMdd",
                     last_n: Int = 1): String = {
      val calendar: Calendar = Calendar.getInstance()
      val sdf = new SimpleDateFormat(date_format)
      try
        calendar.setTime(sdf.parse(biz_date))
      catch {
        case e: ParseException => // omit
      }
      calendar.set(Calendar.DATE, calendar.get(Calendar.DATE) - last_n)
      sdf.format(calendar.getTime)
    }
    // transform
    val result = source
      .map(_.split("\t"))
      .map(iterm => Login(iterm(0).toInt, iterm(1)))
      .groupBy(_.uid) // RDD[(Int, Iterable[Login])]
      .map(iterm => {
        // 用于給此uid標(biāo)記是否符合要求
        var CONTINUOUS_LOGIN_N = false
        val logins = iterm._2
          .toSeq
          .sortWith((v1, v2) => v1.loginTime.compareTo(v2.loginTime) > 0)
        var lastLoginTime: String = ""
        var loginDays: Int = 0
        logins
          .foreach(iterm => {
            if (lastLoginTime == "") {
              lastLoginTime = iterm.loginTime
              loginDays = 1
            } else if (getLastNDate(iterm.loginTime) == lastLoginTime) {
              lastLoginTime = iterm.loginTime
              loginDays = 2
            } else {
              lastLoginTime = iterm.loginTime
              loginDays = 1
            }
          })
        if (loginDays > 3) CONTINUOUS_LOGIN_N = true
        /** 此處可以使用集合將連續(xù)登陸的情況保留，
         * 也可以直接按照是否連續(xù)登陸N天進(jìn)行標(biāo)記
         */
        (iterm._1, CONTINUOUS_LOGIN_N)
      })
      .filter(_._2)
      .map(_._1)
    // sink
    result.foreach(println(_))
  }
}