R, Python 분석과 프로그래밍의 친구 (by R Friend)

  (1) abalone 데이터셋으로 테이블 만들기

---------------------------------

-- Linear Regression in Parallel 

-- using PL/R

---------------------------------

-- Dataset for example: abalone dataset from the UC Irvine Machine Learning Repository

-- url: http://archive.ics.uci.edu/ml/machine-learning-databases/abalone/abalone.data

-- Create an external web table

DROP EXTERNAL TABLE IF EXISTS abalone_external;

CREATE EXTERNAL WEB TABLE abalone_external(

sex text 

, length float8

, diameter float8

, height float8

, whole_weight float8

, shucked_weight float8

, viscera_weight float8

, shell_weight float8

, rings integer -- target variable to predict

) LOCATION('http://archive.ics.uci.edu/ml/machine-learning-databases/abalone/abalone.data') 

FORMAT 'CSV' 

(null as '?');

-- Create a table of abalone

DROP TABLE IF EXISTS abalone;

CREATE TABLE abalone AS 

SELECT * FROM abalone_external

DISTRIBUTED BY (sex);

-- Viewing data distribution

SELECT gp_segment_id, COUNT(*) AS row_cnt

FROM abalone

GROUP BY gp_segment_id;

-- Check data

SELECT * FROM abalone LIMIT 5;

  (2) Train, Test set 테이블 분할 (train, test set split)

---------------------------

-- Train, Test set split

---------------------------

-- Check the number of observations per sex group(F, I, M)
SELECT sex, COUNT(*) FROM abalone GROUP BY sex;

-- Train, Test set split

DROP TABLE IF EXISTS out_train, out_test;

SELECT madlib.train_test_split(

'abalone',    -- Source table

'out',     -- Output table

    0.8,       -- train_proportion

    NULL,      -- Default = 1 - train_proportion = 0.5

    'sex', -- Strata definition

    'rings, shucked_weight, diameter', -- Columns to output

    FALSE,      -- Sample with replacement

    TRUE);     -- Separate output tables

SELECT * FROM out_train LIMIT 5;

SELECT sex, count(*) FROM out_train GROUP BY sex;

SELECT sex, count(*) FROM out_test GROUP BY sex;

  (3) array aggregation 하여 PL/R에서 사용할 데이터셋 준비하기

-- Data Preparation

-- : array aggregation using array_agg()

DROP TABLE IF EXISTS abalone_array;

CREATE TABLE abalone_array AS 

SELECT

sex::text -- group

, array_agg(rings::float8) as rings             -- y

, array_agg(shucked_weight::float8) as s_weight -- x1

, array_agg(diameter::float8) as diameter       -- x2

FROM out_train

GROUP BY sex

DISTRIBUTED BY (sex);

SELECT * FROM abalone_array;

  (4) 선형회귀모형 적합하는 PL/R 사용자 정의 함수 정의하기 (Define PL/R UDF)

----------------------------------------------------------------

-- (4-1) PL/R : Linear Regression Model's Coefficients --> Predict

----------------------------------------------------------------

-- Return Types

DROP TYPE IF EXISTS lm_abalone_type CASCADE;

CREATE TYPE lm_abalone_type AS (

variable text

, coef_est float

, std_error float

, t_stat float

, p_value float

);

-- PL/R User Defined Function

DROP FUNCTION IF EXISTS plr_lm_train(float8[], float8[], float8[]);

CREATE OR REPLACE FUNCTION plr_lm_train(

rings float8[]

, s_weight float8[]

, diameter float8[]

) RETURNS SETOF lm_abalone_type AS

$$

m1 <- lm(rings ~ s_weight + diameter)

m1_s <- summary(m1)$coef

temp_m1 <- data.frame(rownames(m1_s), m1_s)

return(temp_m1)

$$

LANGUAGE 'plr';

 (5) PL/R 실행하기 (execute PL/R UDF in Greenplum DB)

-- Execution of Linear Regression PL/R 

DROP TABLE IF EXISTS lm_abalone_model_coef;

CREATE TABLE lm_abalone_model_coef AS (

SELECT sex, (plr_lm_train(rings, s_weight, diameter)).* 

FROM abalone_array

) DISTRIBUTED BY (sex);

SELECT * FROM lm_abalone_model_coef;

  (6) 적합한 선형회귀모형을 사용해 test set에 대해 예측하기 (prediction on test set)

------------------------------------------------

-- Prediction and Model Evaluation for Test set

------------------------------------------------

-- Preparation of test set in aggregated array

DROP TABLE IF EXISTS test_array;

CREATE TABLE test_array AS 

SELECT

sex::text

, array_agg(rings::float8) as rings             -- y

, array_agg(shucked_weight::float8) as s_weight -- x1

, array_agg(diameter::float8) as diameter       --x2

FROM out_test

GROUP BY sex

DISTRIBUTED BY (sex);

SELECT * FROM test_array;

-- Define composite data type for predicted return values

DROP TYPE IF EXISTS lm_predicted_type CASCADE;

CREATE TYPE lm_predicted_type AS (

id int

, actual float

, pred float

);

-- Define PL/R UDF of prediction per groups using linear regression coefficients

DROP FUNCTION IF EXISTS plr_lm_coef_predict(float8[], float8[], float8[], float8[]);

CREATE OR REPLACE FUNCTION plr_lm_coef_predict(

rings float8[]

, s_weight float8[]

, diameter float8[]

, coef_est float8[]

) RETURNS SETOF lm_predicted_type AS

$$

actual <- rings # y

intercept <- 1

X <- cbind(intercept, s_weight, diameter) # X matrix

coef_est <- matrix(coef_est) # coefficients matrix

predicted <- X %*% coef_est  # matrix multiplication

df_actual_pred <- data.frame(actual, predicted)

id <- as.numeric(rownames(df_actual_pred))

return(data.frame(id, df_actual_pred))

$$

LANGUAGE 'plr';

-- Execute PL/R Prediction UDF

DROP TABLE IF EXISTS out_coef_predict;

CREATE TABLE out_coef_predict AS (

SELECT sex, (plr_lm_coef_predict(c.rings, c.s_weight, c.diameter, c.coef_est)).*

FROM (

SELECT a.*, b.coef_est

FROM test_array a, 

(SELECT sex, array_agg(coef_est) AS coef_est FROM lm_abalone_model_coef GROUP BY sex) b

WHERE a.sex = b.sex

) c

) DISTRIBUTED BY (sex);

-- Compare 'actual' vs. 'predicted'

SELECT * FROM out_coef_predict WHERE sex = 'F' ORDER BY sex, id LIMIT 10;

  (7) 회귀모형 자체를 Serialize 해서 DB에 저장하고, Unserialize 해서 예측하기

--------------------------------------------------------------------

-- (2) PL/R : Linear Model --> Serialize --> Deserialize --> Predict

--------------------------------------------------------------------

-- PL/R User Defined Function Definition

DROP FUNCTION IF EXISTS plr_lm_model(float8[], float8[], float8[]);

CREATE OR REPLACE FUNCTION plr_lm_model(

    rings float8[]

, s_weight float8[]

, diameter float8[]

) RETURNS bytea -- serialized model as a byte array

AS

$$

lr_model <- lm(rings ~ s_weight + diameter)

return (serialize(lr_model, NULL))

$$

LANGUAGE 'plr';

-- Execution of Linear Regression PL/R 

DROP TABLE IF EXISTS lm_abalone_model;

CREATE TABLE lm_abalone_model AS (

SELECT sex, plr_lm_model(rings, s_weight, diameter) AS serialized_model

FROM abalone_array

) DISTRIBUTED BY (sex);

-- We can not read serialized model

SELECT * FROM lm_abalone_model;

DROP TYPE IF EXISTS lm_predicted_interval_type CASCADE;

CREATE TYPE lm_predicted_interval_type AS (

id int

, actual float

, pred float

, lwr float

, upr float

);

-- PL/R function to read a serialized PL/R model

DROP FUNCTION IF EXISTS plr_lm_model_predict(float8[], float8[], float8[], bytea);

CREATE OR REPLACE FUNCTION plr_lm_model_predict(

rings float8[]

, s_weight float8[]

, diameter float8[]

, serialized_model bytea

) RETURNS SETOF lm_predicted_interval_type 

AS

$$

model <- unserialize(serialized_model)

actual <- rings # y

X <- data.frame(s_weight, diameter) # new data X

predicted <- predict(model, newdata = X, interval = "confidence")

df_actual_pred <- data.frame(actual, predicted)

id <- as.numeric(rownames(df_actual_pred))

return (data.frame(id, df_actual_pred))

$$

LANGUAGE 'plr';

-- Predict

DROP TABLE IF EXISTS out_model_predict;

CREATE TABLE out_model_predict AS (

SELECT sex, (plr_lm_model_predict(c.rings, c.s_weight, c.diameter, c.serialized_model)).*

FROM (

SELECT a.*, b.serialized_model

FROM test_array a, lm_abalone_model b

WHERE a.sex = b.sex

) c

) DISTRIBUTED BY (sex);

SELECT * FROM out_model_predict WHERE sex = 'F' ORDER BY sex, id LIMIT 10;

ihongdon-ui-MacBook-Pro:~ ihongdon$ psql postgres

psql (11.5)

Type "help" for help.

postgres=#

$ psql -h host_name  -d db_name  -u user_id

$ Password ********

postgres=#

postgres=# select pid, now() - query_start as "runtime", usename, datname, state, query from pg_stat_activity where now() - query_start > '2 minutes'::interval and state = 'active' order by runtime desc;

 pid  |     runtime     | usename  | datname  | state  |                                      query

------+-----------------+----------+----------+--------+----------------------------------------

 4941 | 00:34:00.595933 | ihongdon | postgres | active |              +

   |          |          |        | DROP TABLE IF EXISTS my_table;              +      

   |          |          |        | CREATE TABLE my_table AS (              +     

   |          |          |        |     SELECT              +

   |          |          |        |         a.*              +

   |          |          |        |         , b.age             +

   |          |          |        |         , b.pch_amt              +

   |          |          |        |         , b.pch_cnt              +

   |          |          |        |         , b.age              +

   |          |          |        |     FROM              +

   |          |          |        |         cust_master AS a              +

   |          |          |        |         LEFT OUTER JOIN cust_pch AS b              +

   |          |          |        |             ON  a.cust_id = b.cust_id            +

   |          |          |        | );              +

   |          |          |        |

(1 row)

postgres=#

postgres=# SELECT pg_terminate_backend(4941); 

pg_terminate_backend

----------------------

 t

(1 row)

postgres=#

postgres=#

-- ssh로 GPDB 접속

> ssh gpadmin@xxx.xxx.xxx.xxx
> password: xxxxxxxxx

-- 현재 수행 중인 쿼리 조회
> select datname, username, proceed, sess_id, current_query from pg_stat_activity;

-- 현재 수행중인 쿼리 강제 종료
> select pg_cancel_backend(**put pid here**)

> select gp_segment_id, count(*) from mytable group by gp_segment_id;

> select * from pg_catalog.gp_segment_configration;

select column_name, data_type

from information_schema.columns
where table_name ='table_name'

select

    ctime, 

    hostname, 

    avg(cpu_user + cpu_sys) cpu, 

    avg(mem_actual_used/1024/1024/1024) memogy_gigabyte, 

    avg(dist_rb_rate/1024/1024) dist_rb_megabyte, 

    avg(disk_wb_rate/1024/1024) disk_wb_megabyte, 

    avg(net_rb_rate/1024/1024) net_rb_megabyte, 

    avg(net_wb_rate/1024/1024) net_wb_megabyte

 from system_hitory

 where ctime between '2019-01-01 12:00:00' and '2019-01-01 23:59:59' 

and hostname not in ('dev1.gphd.local', 'dev2.gphd.local')

group by ctime, hostname

order by 1, 2 asc

http://localhost:28080/main/querymonitor/hostmetrics

----------------------------------------------

-- Graph Analytics with Greenplum and MADlib

----------------------------------------------

-- CREATE VERTEX TABLE

DROP TABLE IF EXISTS vertex;

CREATE TABLE vertex(

id INTEGER

) DISTRIBUTED RANDOMLY;

INSERT INTO vertex 

VALUES

(0), 

(1), 

(2), 

(3), 

(4),

(5),

(6);

select * from vertex

-- CREATE EDGE TABLE

DROP TABLE IF EXISTS edge;

CREATE TABLE edge(

src INTEGER,

dest INTEGER,

user_id INTEGER

)

DISTRIBUTED BY (user_id);

INSERT INTO edge 

VALUES 

(0, 1, 1), (0, 2, 1), -- user id 1

(0, 4, 1), (1, 2, 1),

(1, 3, 1), (2, 3, 1),

(2, 5, 1), (2, 6, 1),

(3, 0, 1), (4, 0, 1),

(5, 6, 1), (6, 3, 1),

(0, 1, 2), (0, 2, 2), -- user id 2

(0, 4, 2), (1, 2, 2),

(1, 3, 2), (2, 3, 2),

(3, 0, 2), (4, 0, 2),

(5, 6, 2), (6, 3, 2);

select * from edge;

-- (1) Compute the PageRank with All IDs

DROP TABLE IF EXISTS pagerank_out, pagerank_out_summary;

SELECT madlib.pagerank(

'vertex' -- Vertex table

, 'id' -- Vertex id column

, 'edge' -- Edge table

, 'src=src, dest=dest' -- Comma delimited string of edge arguments

, 'pagerank_out' -- Output table of RageRank

, NULL -- Default damping factor (0.85)

); 

SELECT * FROM pagerank_out ORDER BY pagerank DESC;

-- (2) Compute the PageRank of vertices associated with each user using the grouping feature

DROP TABLE IF EXISTS pagerank_gr_out, pagerank_gr_out_summary;

SELECT madlib.pagerank(

'vertex' -- Vertex table

, 'id' -- Vertex id column

, 'edge' -- Edge table

, 'src=src, dest=dest' -- Comma delimited string of edge arguments

, 'pagerank_gr_out' -- Output table of PageRank

, NULL -- Default damping factor (0.85)

, NULL -- Default max iterations (100)

, 0.00000001 -- Threshold

, 'user_id'); -- Grouping column name

SELECT * FROM pagerank_gr_out ORDER BY user_id, pagerank DESC;

-- (3) Personalized PageRank of vertices {2, 4}

DROP TABLE IF EXISTS pagerank_pers_out, pagerank_pers_out_summary;

SELECT madlib.pagerank(

'vertex' -- Vertex table

, 'id' -- Vertex id column

, 'edge' -- Edge table

, 'src=src, dest=dest' -- Comma delimited string of edge arguments

, 'pagerank_pers_out' -- Output table of PageRank

, NULL -- Default damping factor (0.85)

, NULL

-- Default max iterations (100)

, NULL -- Default Threshold (1/number of vertices*1000)

, NULL -- No Grouping

, '{2, 4}' -- Personalization vertices

);

SELECT * FROM pagerank_pers_out ORDER BY pagerank DESC;

SELECT * FROM pagerank_pers_out_summary;

#!/usr/bin/env python2

# -*- coding: utf-8 -*-

"""

Created on Tuesday Jau 29 2019

@author: Hongdon Lee

"""

#%% network visualization using Graphviz

import numpy as np

import pandas as pd

import pygraphviz as pgv

def run_query(query):

     import pandas as pd

     import psycopg2 as pg

    # DB Connection

     conn = pg.connect(host='localhost',

                       port='5432', 

                       dbname='gpadmin', 

                       user='gpadmin', 

                       password='pivotal')

     # Get a DataFrame

     query_result = pd.read_sql(query, conn)

     conn.close()

     return query_result

#%% Network Edge Table

query = """

select a.*, b.pagerank 

    from edge a

    left outer join pagerank_out b on a.src = b.id

;

"""

edge_pagerank = run_query(query)

#%% PageRank Values

query = """

select * from pagerank_out;

"""

pagerank_out  = run_query(query)

#%% NW Visualization using Graphviz with different size proportional to PageRank

import pygraphviz as pgv

# Generating the output flow_graph with PyGraphviz

flow_graph = pgv.AGraph(strict=False, directed=True) # directed graph

# Flow Direction(Left to Right, or Top to Bottom)

flow_graph.graph_attr['rankdir'] = 'LR' # from Left to Right

# Node Shape

flow_graph.node_attr['shape'] = 'circle'

# Making node with different size proportional to PageRank

for i in range(len(pagerank_out)):

    label_text = str(pagerank_out.id[i]) + '\n(' + str(pagerank_out.pagerank[i].round(decimals=2)) + ')'

    node_width = pagerank_out.pagerank[i]*10

    node_height = pagerank_out.pagerank[i]*10

    flow_graph.add_node(str(pagerank_out.id[i]), 

                        label=label_text, 

                        **{'width': str(node_width), 

                           'height': str(node_height)})

# Adding edge with different color by user_id

colors = ['blue', 'red']

for i in range(len(edge_pagerank)):

    if edge_pagerank.user_id[i] == 1:

        color_text = colors[0]

    else:

        color_text = colors[1]

    flow_graph.add_edge(str(edge_pagerank.src[i])

                        , str(edge_pagerank.dest[i])

                        , color = color_text)

#----- Finally, Draw the Network Diagram using dot program :-)

flow_graph.draw("/Users/ihongdon/Documents/nw_diagram.png", prog='dot')