Sessionize in Hive 0.12 (to Parquet table)

gistfile1.txt

ADD JAR /opt/cloudera/parcels/CDH/lib/hive/lib/hive-contrib.jar;

DROP TABLE raw_log;

CREATE EXTERNAL TABLE raw_log(
	IP		STRING,
	timestamp	STRING,
	URL		STRING,
	referrer	STRING,
	user_agent	STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
	"input.regex" = "(\\d+.\\d+.\\d+.\\d+).*\\[(.*)\\].*GET (\\S*).*\\d+ \\d+ (\\S+) \"(.*)\""
)
location '/etl/bikeshop/clickstream/raw/year=2014/month=5/day=6'
;

--select * from raw_log limit 5;

CREATE TABLE if not exists apache_log_parquet(
        IP              STRING,
        timestamp       STRING,
        URL             STRING,
        referrer        STRING,
        user_agent      STRING,
	ts		BIGINT,
	session_id	INT)
PARTITIONED BY (year int, month int)
ROW FORMAT SERDE 'parquet.hive.serde.ParquetHiveSerDe'
STORED AS
INPUTFORMAT "parquet.hive.DeprecatedParquetInputFormat"
OUTPUTFORMAT "parquet.hive.DeprecatedParquetOutputFormat";


set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.dynamic.partition=true;

insert into table apache_log_parquet
partition(year,month)
select IP,timestamp,URL,referrer,user_agent,ts,
coalesce(sum(case when new='Y' then 1 end) over (PARTITION BY IP ORDER BY ts),0) session_id
, year(from_unixtime(ts)) year, month(from_unixtime(ts)) month
from
(select IP,timestamp,URL,referrer,user_agent,ts,case when ts-lag(ts,1) OVER (PARTITION BY IP ORDER BY ts) > 30*60 then 'Y' else 'N' end new
from (
        select IP,timestamp,URL,referrer,user_agent,unix_timestamp(timestamp,'dd/MMM/yyyy:HH:mm:ss') as ts
        from raw_log
) t) s;


select * from apache_log_parquet limit 5;
select count(*) from apache_log_parquet;