sessionisation_in_spark.py

sqlloggik4_df = """
SELECT *
  , CAST(id as BIGINT) *10000 + SUM(new_session) 
    OVER (PARTITION BY id ORDER BY starttid)  
     AS session_id
FROM(
SELECT *,
    unix_timestamp(l.starttid) - LAG(unix_timestamp(l.starttid)) OVER (PARTITION BY l.id ORDER BY  l.starttid) timesincelast,
    CASE
    WHEN unix_timestamp(l.starttid) - LAG(unix_timestamp(l.starttid)) OVER (PARTITION BY l.id ORDER BY  l.starttid)  >= 30 * 60 
    THEN 1
    ELSE 0
    END as new_session 
    from loggik_df l
) s1
    """

sql4df = sqlHContext.sql(sqlloggik4_df)