PL/PgSQL 从大表数据抽取统计数据

--Extract Data from views to stats table v2.0
CREATE OR REPLACE FUNCTION (IN endTs TIMESTAMP,IN tab_name VARCHAR) RETURNS VARCHAR AS $$
DECLARE
 t_res       		   		VARCHAR;
 t_endTs	 		   		TIMESTAMP;
 t_startTs   		   		TIMESTAMP;
 t_imprs_country_stats_dt   TIMESTAMP;
 t_imprs_stats_dt			TIMESTAMP;
 t_imprs_audit_id           BIGINT;
 t_rec_views 				views%ROWTYPE;
 t_rec_imprs_country_stats  RECORD;
 t_rec_imprs_stats 			RECORD;
 t_rec_audit 				s_ext_stats_audit%ROWTYPE;
 t_tabName   				VARCHAR;
 -------Define id variable for saving the s_ads_imprs_country_stats primary id
 v_rec_imprs_country_stats_id BIGINT;
 -------Define id variable for saving the s_ads_imprs_stats primary id
v_rec_imprs_stats_id 		BIGINT; 

BEGIN
	t_tabName := tab_name;
	--- Define the function return chars 
	t_res := 'Ok. It is successfull to extract data.';
	--- Check views_audit table whether contains data
	SELECT * INTO t_rec_audit FROM s_ext_stats_audit ORDER BY id DESC LIMIT 1;
	------ Get the audit table primary key id
	t_imprs_audit_id := t_rec_audit.views_pid;
	-------Get the start time from audit table 
	t_startTs := t_rec_audit.views_cdt;
	-------Get the end time from audit table
	t_endTs := endTs;
	RAISE NOTICE 'The start time is %',t_startTs;
	RAISE NOTICE 'The end time is %',t_endTs;
	-------The loop for getting ads impressions country stats data
	FOR t_rec_imprs_country_stats IN EXECUTE 'SELECT ad_id,COALESCE(country,''OTHER'') AS country,SUM(impressions) AS impressions,dates,source  FROM '||
		'(((SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,a.client_ip,COUNT(a.fb_uid) AS impressions, 0 AS source FROM '||t_tabname||
		' a,ads b WHERE a.ad_id = b.id '||'AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.fb_uid IS NOT NULL GROUP BY b.ad_id,a.client_ip,dates ORDER BY b.ad_id)'||
		' UNION ALL '||
		'(SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,a.client_ip,COUNT(a.bebo_uid) AS impressions, 1 AS source FROM '||t_tabname||
			' a,ads b WHERE a.ad_id = b.id 	AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.bebo_uid IS NOT NULL GROUP BY b.ad_id,a.client_ip,dates ORDER BY b.ad_id)'||
		' UNION ALL '||
		'(SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,a.client_ip,COUNT(a.ms_uid) AS impressions, 2 AS source FROM '||t_tabname||
			' a,ads b WHERE a.ad_id = b.id 	AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.ms_uid IS NOT NULL GROUP BY b.ad_id,a.client_ip,dates ORDER BY b.ad_id)'||
		' UNION ALL '||
		'(SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,a.client_ip,COUNT(a.hi5_uid) AS impressions, 3 AS source FROM '||t_tabname||
			' a,ads b WHERE a.ad_id = b.id 	AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.hi5_uid IS NOT NULL GROUP BY b.ad_id,a.client_ip,dates ORDER BY b.ad_id)) '||
		' AS tmp_ads_imprs_country_stats '||
		' LEFT JOIN geoip G ON ip4(tmp_ads_imprs_country_stats.client_ip) <<= G.network '||
		' LEFT JOIN geoip_location AS geoip_loc ON G.location_id = geoip_loc.id ) AS tmp_ads_imprs_country_stats_t '||
		' GROUP BY ad_id,country,dates,source'
		LOOP
			SELECT id INTO v_rec_imprs_country_stats_id FROM s_ads_imprs_country_stats WHERE ad_id = t_rec_imprs_country_stats.ad_id 
				AND country = t_rec_imprs_country_stats.country AND crdate = t_rec_imprs_country_stats.dates AND data_platform = t_rec_imprs_country_stats.source;
			RAISE NOTICE 't_rec_imprs_country_stats.crdate is %',t_rec_imprs_country_stats.dates;
			IF NOT FOUND THEN
				INSERT INTO s_ads_imprs_country_stats (crdate,ad_id,country,impressions,data_platform) VALUES (t_rec_imprs_country_stats.dates
								,t_rec_imprs_country_stats.ad_id,t_rec_imprs_country_stats.country,t_rec_imprs_country_stats.impressions,t_rec_imprs_country_stats.source);
			ELSE
				UPDATE s_ads_imprs_country_stats SET impressions = impressions + t_rec_imprs_country_stats.impressions WHERE id = v_rec_imprs_country_stats_id;
			END IF;	
	END LOOP;
	
	--- The loop for getting ads impressions stats data
	FOR t_rec_imprs_stats IN EXECUTE 'SELECT *  FROM'||
		'((SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,COUNT(a.fb_uid) AS impressions, 0 AS data_platform FROM '||t_tabname||
			' a,ads b WHERE a.ad_id = b.id AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.fb_uid IS NOT NULL GROUP BY b.ad_id,dates ORDER BY b.ad_id)'||
		' UNION ALL '||
		'(SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,COUNT(a.bebo_uid) AS impressions, 1 AS data_platform FROM '||t_tabname||
			' a,ads b WHERE a.ad_id = b.id AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.bebo_uid IS NOT NULL GROUP BY b.ad_id,dates ORDER BY b.ad_id)'||
		' UNION ALL '||
		'(SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,COUNT(a.ms_uid) AS impressions, 2 AS data_platform FROM '||t_tabname||
			' a,ads b WHERE a.ad_id = b.id AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.ms_uid IS NOT NULL GROUP BY b.ad_id,dates ORDER BY b.ad_id)'||
		' UNION ALL '||
		'(SELECT DISTINCT(DATE_TRUNC(''DAY'',a.crdate)) AS dates,b.ad_id,COUNT(a.hi5_uid) AS impressions, 3 AS data_platform FROM '||t_tabname||
			' a,ads b WHERE a.ad_id = b.id AND a.id > '||t_imprs_audit_id||' AND a.crdate > '''||t_startTs||''' AND a.crdate < '''||t_endTs||
			''' AND a.hi5_uid IS NOT NULL GROUP BY b.ad_id,dates ORDER BY b.ad_id))'||
		' AS tmp_ads_imprs_stats'
	LOOP
			SELECT id INTO v_rec_imprs_stats_id FROM s_ads_imprs_stats WHERE ad_id = t_rec_imprs_stats.ad_id 
				AND crdate = t_rec_imprs_stats.dates AND data_platform = t_rec_imprs_stats.data_platform;
			
			IF NOT FOUND THEN
				INSERT INTO s_ads_imprs_stats (crdate,ad_id,impressions,data_platform) VALUES (t_rec_imprs_stats.dates
								,t_rec_imprs_stats.ad_id,t_rec_imprs_stats.impressions,t_rec_imprs_stats.data_platform);
			ELSE
				UPDATE s_ads_imprs_stats SET impressions = impressions + t_rec_imprs_country_stats.impressions WHERE id = v_rec_imprs_stats_id;
			END IF;	
	END LOOP;
	
	--- To insert the latest data into audit table, that means which data will be extracted and where it is from
		
	EXECUTE 'INSERT INTO s_ext_stats_audit (views_pid,views_cdt,opr_dt) SELECT id,crdate,now() FROM '||t_tabName||' WHERE crdate >= '''||t_startTs||
				''' and crdate <'''|| t_endTs||''' ORDER BY id DESC LIMIT 1';	
				
	RETURN t_res;

	RAISE EXCEPTION 'It is failed.';
END;
$$ LANGUAGE plpgsql;

评论

1 楼 martri 2008-10-31  

分步抽取麽？