PostgreSQL

PostgreSQL下查找与去除重复数据

有时候难免会导入重复的数据（一行所有的列都重复），问题是如何找出来呢？假设有表 tbl，结构如下，

CREATE TABLE tbl (
 id serial,
 href text NULL,
 title text NULL,
 image text NULL
)
WITH (
 OIDS=FALSE
);

CREATE TABLE tbl (

id serial,

href text NULL,

title text NULL,

image text NULL

)

WITH (

OIDS=FALSE

);

如何判断是否存在id, href，title和image都重复的数据呢(这种情况一般发生在导入重复）？

select * from (
  SELECT id,
  ROW_NUMBER() OVER(PARTITION BY id, href, title, image) AS Row
  FROM tbl
) dups
where 
dups.Row > 1

select * from (

SELECT id,

ROW_NUMBER() OVER(PARTITION BY id, href, title, image) AS Row

FROM tbl

) dups

where

dups.Row > 1

如何判断是否存在重复href，title和image，但id不同的数据呢（这种一般是OLTP的新增重复）？

select * from (
  SELECT id,
  ROW_NUMBER() OVER(PARTITION BY  href, title, image ORDER BY id asc) AS Row
  FROM tbl
) dups
where 
dups.Row > 1

select * from (

SELECT id,

ROW_NUMBER() OVER(PARTITION BY href, title, image ORDER BY id asc) AS Row

FROM tbl

) dups

where

dups.Row > 1

接下来就是这样的数据如何只保留最早插入的数据，而删除后面重复插入的数据？

DELETE FROM tbl
WHERE id IN (SELECT id
              FROM (SELECT id,
                             ROW_NUMBER() OVER (partition BY href, title, image ORDER BY id) AS rnum
                     FROM tbl) t
              WHERE t.rnum > 1);

DELETE FROM tbl

WHERE id IN (SELECT id

FROM (SELECT id,

ROW_NUMBER() OVER (partition BY href, title, image ORDER BY id) AS rnum

FROM tbl) t

WHERE t.rnum > 1);

reference, https://wiki.postgresql.org/wiki/Deleting_duplicates

PostgreSQL与交叉表查询

PostgreSQL提供了一个tablefunc 模块，内置了多个函数，其中就有crosstab（交叉表，又叫行转列，或者长表转宽表），具体使用参见PostgreSQL文档（中文，英文）。如果不清楚生成的宽表有多少列，或者列太多，手工敲很容易吃力不讨好，那么可以借助这个函数（pivotcode）来简化工作，

-- PL/pgSQL code to create pivot tables with automatic column names
-- Eric Minikel, CureFFI.org - 2013-03-19

-- prerequisite: install the tablefunc module
create extension tablefunc;

-- tablename: name of source table you want to pivot
-- rowc: the name of the column in source table you want to be the rows
-- colc: the name of the column in source table you want to be the columns
-- cellc: an aggregate expression determining how the cell values will be created
-- celldatatype: desired data type for the cells
create or replace function pivotcode (tablename varchar, rowc varchar, colc varchar, cellc varchar, celldatatype varchar) returns varchar language plpgsql as $$
declare
    dynsql1 varchar;
    dynsql2 varchar;
    columnlist varchar;
begin
    -- 1. retrieve list of column names.
    dynsql1 = 'select string_agg(distinct ''_''||'||colc||'||'' '||celldatatype||''','','' order by ''_''||'||colc||'||'' '||celldatatype||''') from '||tablename||';';
    execute dynsql1 into columnlist;
    -- 2. set up the crosstab query
    dynsql2 = 'select * from crosstab (
 ''select '||rowc||','||colc||','||cellc||' from '||tablename||' group by 1,2 order by 1,2'',
 ''select distinct '||colc||' from '||tablename||' order by 1''
 )
 as newtable (
 '||rowc||' varchar,'||columnlist||'
 );';
    return dynsql2;
end
$$

-- PL/pgSQL code to create pivot tables with automatic column names

-- Eric Minikel, CureFFI.org - 2013-03-19

-- prerequisite: install the tablefunc module

create extension tablefunc;

-- tablename: name of source table you want to pivot

-- rowc: the name of the column in source table you want to be the rows

-- colc: the name of the column in source table you want to be the columns

-- cellc: an aggregate expression determining how the cell values will be created

-- celldatatype: desired data type for the cells

create or replace function pivotcode (tablename varchar, rowc varchar, colc varchar, cellc varchar, celldatatype varchar) returns varchar language plpgsql as $$

declare

dynsql1 varchar;

dynsql2 varchar;

columnlist varchar;

begin

-- 1. retrieve list of column names.

dynsql1 = 'select string_agg(distinct ''_''||'||colc||'||'' '||celldatatype||''','','' order by ''_''||'||colc||'||'' '||celldatatype||''') from '||tablename||';';

execute dynsql1 into columnlist;

-- 2. set up the crosstab query

dynsql2 = 'select * from crosstab (

''select '||rowc||','||colc||','||cellc||' from '||tablename||' group by 1,2 order by 1,2'',

''select distinct '||colc||' from '||tablename||' order by 1''

)

as newtable (

'||rowc||' varchar,'||columnlist||'

);';

return dynsql2;

end

但这个函数，有如下的缺点，如果返回的列有空格，需要自己手工修改可能会有大量的Null值，而不是0 返回的是一个SQL，你需要copy出来后自己再执行（虽然也可以改进下，直接生成表）下面这个plpython函数，pivotmytable，可以很好地解决上面的问题，python 2和python3（只需讲函数里的plpythonu替换为plpython3u即可）均支持，

/* pivotMyTable An improved crosstab function for PostgreSQL.
 
 Copyright 2014 Jorge Tornero Nunez http://imasdemase.com
 
 pivotMyTable is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 (at your option) any later version.
 
 pivotMyTable is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.
 
 You should have received a copy of the GNU General Public License
 along with pivotMyTable.  If not, see <http://www.gnu.org/licenses/>.

FUNCTION DESCRIPTION
====================
pivotMyTablee is a PL/Python function for use in PostgreSQL servers. Its aim is
to get crosstab/pivoted tables in a more friendly way that PostgreSQL module
tablefunc does with its crosstab series functions and in fact it behaves in the
end as a proxy for tablefunc functions.

pivotMyTablee makes it possible in the same way that other available solutions,
automating the creation the queries that the tablefunc crosstab functions need to work.

Also, pivotMyTablee makes possible to directly get percentages,
as well as get rid of null values in the oputput tables.

FUNCTION PARAMETERS/OPTIONS
===========================
input_table (varchar):  Name of the table to get data to pivot.
output_table(varchar):  Name for the output table.
group_fields(varchar):  Name of the column(s) for categorizing the data. Unlike the
                        native tablefunc module functions crosstab, it accepts multiple
                        columns.
pivot_field(varchar):   Column to be pivoted, must be a single column.
value_field(varchar):   Data to aggregate for the pivot field. It is expected for
                        value_field to be a numeric (int, float, etc) field.
agg_func(varchar):      Aggregate function to apply to the data from value_field. It
                        must be specified without brackets.
                        It's user responsability to check for function/data type
                        compatibility and resulting data. So far it has been tested
                        with sum.
as_percentage(bool):    This option enables the output as percentage of each output
                        columns over its overall sum BY ROW. It defaults to FALSE, so                        to enable percentage calculation set it to TRUE.
sort_order(varchar):    This option enables sorting the resulting pivoted columns.
                        You can specify 'asc', 'desc'. It defaults to 'no_sorting'
drop_ex_tbl (boolean):  pivotMyTablee checks if the output table specified with the
                        parameter output_table exists before further processing.
                        Setting this option to TRUE makes possible to automatically
                        drop the existing table. By default, drop_ex_tbl is set to
                        FALSE so the function exists with a warning if output_table
                        already exists in the database.
as_view (boolean):       This option enables the output to a view instead of to a table.
                        To enable view output, set it to TRUE. Defaults to FALSE.
                        
*/

CREATE OR REPLACE FUNCTION pivotmytable (input_table varchar,
                                   output_table varchar,
                                   group_fields varchar,
                                   pivot_field varchar,
                                   value_field varchar,
                                   agg_func varchar,
                                   as_percentage bool default false,
                                   sort_order varchar default 'no_sorting',
                                   drop_ex_tbl boolean default false,
                                   as_view boolean default false)
  RETURNS varchar
  LANGUAGE plpythonu

AS $$
    """Function pivotMyTable"""
    
    import math
    
    
    # Time to check if input_table,group_fields, pivot_field and value_field parameters are OK
    
        
    tablesQuery = plpy.execute("select relname table_name from pg_catalog.pg_stat_all_tables")
    tableList = [tablename["table_name"] for tablename in tablesQuery]
    
    inputTableColumnsQuery = plpy.execute("select attname column_name from pg_attribute where attrelid = (current_schema()||'.'||'%s')::regclass and attnum > 0 AND NOT attisdropped" %input_table.replace(" ",""))
    inputTableColumnsList = [col["column_name"] for col in inputTableColumnsQuery]
    
    if input_table.replace(" ","") not in tableList:
        return "Your input table %s does not exist.  Please check and try again." %input_table
    
    for parameter in group_fields.split(','):
        if parameter.replace(" ", "") not in inputTableColumnsList:
            return ("The column %s specified in group_fields does not exist in the table %s. Please check and try again." %(parameter,input_table))
    
     
    if pivot_field.replace(" ","") not in inputTableColumnsList:
        return ("The column %s specified in group_field does not exist in the table %s. Please check and try again." %(pivot_field,input_table))
    if value_field.replace(" ","") not in inputTableColumnsList:
        return ("The column %s specified in value_field does not exist in the table %s. Please check and try again." %(value_field,input_table))
    
    # Checking for percentage and aggregate function compatibility
    
    if agg_func.replace(" ","") not in ('sum','count','avg'):
        return "Aggregate function %s is not compatible/tested. Aborting" %agg_func
    
    if as_percentage and (agg_func.replace(" ","")!='sum'):
            return "Percentage calculation and aggregate function %s are not compatible. Aborting" %agg_func
    
    
    # Checks if the output table name already exists in the database
    if as_view:
        table_type = 'view'
    else:
        table_type = 'table'
        
    if (output_table.replace(" ","") in tableList):
        if (drop_ex_tbl):
            plpy.execute("drop %s %s" %(table_type, output_table))
        else:
            return ("The %s %s already exists in the database.  Please check and try again." %(table_type, output_table))
        
        
    # Management of sort order parameter
    
    if sort_order.replace(" ","") == 'asc':
        ordering="(select %s, %s(%s) as ordervalue from %s group by %s order by 2 asc) as t2"%(pivot_field,agg_func,value_field,input_table,pivot_field)
    elif sort_order.replace(" ","") == 'desc':
        ordering="(select %s, %s(%s) as ordervalue from %s group by %s order by 2 desc) as t2"%(pivot_field,agg_func,value_field,input_table,pivot_field)
    elif sort_order.replace(" ","") == 'no_sorting':
        ordering="(select distinct %s from %s order by 1) as t2"%(pivot_field,input_table)
    else:
        return "Wrong sort parameter specification. It must be 'asc', 'desc' or no. Please check and try again."
        
    # GATHERING OF THE PIVOTED COLUMN NAMES AND THEIR DATA TYPES.
    
    # First we get the names of the destination fields, which are the values present in
    # the pivot_field column
    
    destColumns = plpy.execute("select %s as columns from %s " %(pivot_field, ordering))
    
    # We need this for the final field splitting
    destColumns2 = plpy.execute("select distinct %s as column from %s order by 1" %(pivot_field,input_table))
    
    # Now we get the column type. Because all the output pivoted columns
    # will have the same data type than input columns, its definition is
    # simple in the case of integer/double columns, but for numeric columns
    # we need to get both the precision (number of decimals) and the maximum
    # value of its aggregate, to prevent overflows when aggregating the data
    
    # First we make the query to pg_attribute to get columns details. 
    
    columnPropertiesQuery = plpy.execute("SELECT  atttypid::regtype AS data_type, CASE WHEN atttypid IN (21, 23, 20) THEN 0 WHEN atttypid IN (1700) THEN CASE WHEN atttypmod = -1 THEN null ELSE (atttypmod - 4) & 65535 END ELSE null END AS numeric_scale from pg_attribute where attrelid = (current_schema()||'.'||'%s')::regclass and attname = '%s' and attnum > 0 and not attisdropped" %(input_table.replace(" ",""),value_field.replace(" ","")))
    
    # Due to the nature of the PLyResult object returned by plpy.execute,
    # getting the columns properties is a little tricky
    
    columnProperties = [property for property in columnPropertiesQuery]
    
    columnType = columnProperties[0]["data_type"]
    columnScale = columnProperties[0]["numeric_scale"]
        
    # Because the percentage values range from 0 to 1,
    # we need to make room for more decimal numbers for 
    # precision not be lost
    
    if as_percentage:
        fieldType = "numeric(6,5)"
    else:
        if columnType == 'numeric' or agg_func =='avg':
            if columnScale == None:
                columnScale = 8
            maxFieldValue = plpy.execute("select %s(%s) as agg_funcres from %s" %(agg_func, value_field,input_table))
            magnitudeOrder = math.log10(maxFieldValue[0]["agg_funcres"])
            numericFieldWidth = magnitudeOrder + columnScale + 1
            fieldType = ("numeric(%i, %i)" %(numericFieldWidth, columnScale))
            
        elif columnType in ('bigint','smallint','integer','real','double precision'):
            fieldType = columnType
        else:
            return "Your pivot column %s is not of a numeric type.  Please check and try again." %pivot_field
       
    flds = ['"%s" %s'%(destColumn["column"],fieldType) for destColumn in destColumns2]
    fields = ','.join(flds)
    
    gfields=group_fields.replace(" ","").split(',')
    groupingColumns=list()
    idx=1
    for field in gfields:
        columnType = plpy.execute("SELECT format_type(atttypid, atttypmod) AS data_type from pg_attribute where attrelid = (current_schema()||'.'||'%s')::regclass and attname = '%s' and attnum > 0 and not attisdropped" %(input_table.replace(" ",""),field.replace(" ","")))
        tc=[columna["data_type"] for columna in columnType]
        
        # We need this to recover the grouped fields after crosstab execution 
        
        nc = ("split_part(trim(joinedcols,'()'),',',%i)::%s as %s" %(idx,tc[0],field))
               
        groupingColumns.append (nc)
        idx += 1
    groupingColumns=','.join(groupingColumns)

    # Construction of the queries to be passed to crosstab function

    if as_percentage:
        # If we want percentages instead of absolute values, we need to
        # create a new table with the percentages, and the queries
        # that are passed to crosstab funcion are silightly different
        # Notice the use of row(), that makes possible to have crosstabs
        # grouped by more than one single field, but to this to be done,
        # we have had to do a lot of tricks before (see the use of
        # split_part above)
        
        groupbyclause=group_fields + ',' + value_field

        plpy.execute("""create temporary table intertable as
            (select %s,%s,%s/%s(%s::numeric) over (partition by %s) as percentages from %s group by %s,%s,%s order by %s)""" 
            %(group_fields,pivot_field,value_field,agg_func,value_field,group_fields,input_table,group_fields,pivot_field,value_field,group_fields))
        firstQuery=("""select distinct row(%s),%s,%s(percentages) from intertable
            group by %s,%s order by 1,2"""
            %(group_fields,pivot_field,agg_func,group_fields,pivot_field)) 
        secondQuery = ("select distinct %s from intertable order by 1" %(pivot_field))
        
    else:
        # For absolute values, the queries passed to crosstab are simpler
    
        aggrfun=("%s(%s)" %(agg_func,value_field))
        groupbyclause=group_fields
        firstQuery=("select distinct row(%s),%s,%s from %s group by %s,%s order by 1,2" %(group_fields,pivot_field,aggrfun,input_table,groupbyclause,pivot_field))
        secondQuery = ("select distinct %s from %s order by 1" %(pivot_field,input_table))
    
    # We create the sentence for output fields for crosstab
    crossTabQuery = ("joinedcols varchar,%s"%(fields))
    
    # This trick makes possible to get rid of the null values in the pivoted tables.
    # TODO: Consider to make it optional, with a parameter for it.
    
    replaceZeros = ['coalesce("{0}",0) as "{0}"'.format(destColumn["columns"]) for destColumn in destColumns]
    replaze0 = ','.join(replaceZeros)
    
    # And now, we put everything together and execute the query
    finalQuery = ("create %s %s as (select %s,%s from crosstab('%s','%s') as newtable(%s))" %(table_type,output_table,groupingColumns,replaze0,firstQuery,secondQuery,crossTabQuery))  
    plpy.execute(finalQuery)
    
    # A little cleanup may be necessary
    if as_percentage:
        pass
        plpy.execute("drop table intertable")

    return "Your pivoted %s %s has been created." %(table_type, output_table.replace(" ",""))
  
$$;

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

/* pivotMyTable An improved crosstab function for PostgreSQL.

pivotMyTable is free software: you can redistribute it and/or modify

it under the terms of the GNU General Public License as published by

the Free Software Foundation, either version 3 of the License, or

(at your option) any later version.

pivotMyTable is distributed in the hope that it will be useful,

but WITHOUT ANY WARRANTY; without even the implied warranty of

MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

GNU General Public License for more details.

You should have received a copy of the GNU General Public License

along with pivotMyTable. If not, see <http://www.gnu.org/licenses/>.

FUNCTION DESCRIPTION

====================

pivotMyTablee is a PL/Python function for use in PostgreSQL servers. Its aim is

to get crosstab/pivoted tables in a more friendly way that PostgreSQL module

tablefunc does with its crosstab series functions and in fact it behaves in the

end as a proxy for tablefunc functions.

pivotMyTablee makes it possible in the same way that other available solutions,

automating the creation the queries that the tablefunc crosstab functions need to work.

Also, pivotMyTablee makes possible to directly get percentages,

as well as get rid of null values in the oputput tables.

FUNCTION PARAMETERS/OPTIONS

===========================

input_table (varchar): Name of the table to get data to pivot.

output_table(varchar): Name for the output table.

group_fields(varchar): Name of the column(s) for categorizing the data. Unlike the

native tablefunc module functions crosstab, it accepts multiple

columns.

pivot_field(varchar): Column to be pivoted, must be a single column.

value_field(varchar): Data to aggregate for the pivot field. It is expected for

value_field to be a numeric (int, float, etc) field.

agg_func(varchar): Aggregate function to apply to the data from value_field. It

must be specified without brackets.

It's user responsability to check for function/data type

compatibility and resulting data. So far it has been tested

with sum.

as_percentage(bool): This option enables the output as percentage of each output

columns over its overall sum BY ROW. It defaults to FALSE, so to enable percentage calculation set it to TRUE.

sort_order(varchar): This option enables sorting the resulting pivoted columns.

You can specify 'asc', 'desc'. It defaults to 'no_sorting'

drop_ex_tbl (boolean): pivotMyTablee checks if the output table specified with the

parameter output_table exists before further processing.

Setting this option to TRUE makes possible to automatically

drop the existing table. By default, drop_ex_tbl is set to

FALSE so the function exists with a warning if output_table

already exists in the database.

as_view (boolean): This option enables the output to a view instead of to a table.

To enable view output, set it to TRUE. Defaults to FALSE.

CREATE OR REPLACE FUNCTION pivotmytable (input_table varchar,

output_table varchar,

group_fields varchar,

pivot_field varchar,

value_field varchar,

agg_func varchar,

as_percentage bool default false,

sort_order varchar default 'no_sorting',

drop_ex_tbl boolean default false,

as_view boolean default false)

RETURNS varchar

LANGUAGE plpythonu

AS $$

"""Function pivotMyTable"""

import math

# Time to check if input_table,group_fields, pivot_field and value_field parameters are OK

tablesQuery = plpy.execute("select relname table_name from pg_catalog.pg_stat_all_tables")

tableList = [tablename["table_name"] for tablename in tablesQuery]

inputTableColumnsQuery = plpy.execute("select attname column_name from pg_attribute where attrelid = (current_schema()||'.'||'%s')::regclass and attnum > 0 AND NOT attisdropped" %input_table.replace(" ",""))

inputTableColumnsList = [col["column_name"] for col in inputTableColumnsQuery]

if input_table.replace(" ","") not in tableList:

return "Your input table %s does not exist. Please check and try again." %input_table

for parameter in group_fields.split(','):

if parameter.replace(" ", "") not in inputTableColumnsList:

return ("The column %s specified in group_fields does not exist in the table %s. Please check and try again." %(parameter,input_table))

if pivot_field.replace(" ","") not in inputTableColumnsList:

return ("The column %s specified in group_field does not exist in the table %s. Please check and try again." %(pivot_field,input_table))

if value_field.replace(" ","") not in inputTableColumnsList:

return ("The column %s specified in value_field does not exist in the table %s. Please check and try again." %(value_field,input_table))

# Checking for percentage and aggregate function compatibility

if agg_func.replace(" ","") not in ('sum','count','avg'):

return "Aggregate function %s is not compatible/tested. Aborting" %agg_func

if as_percentage and (agg_func.replace(" ","")!='sum'):

return "Percentage calculation and aggregate function %s are not compatible. Aborting" %agg_func

# Checks if the output table name already exists in the database

if as_view:

table_type = 'view'

else:

table_type = 'table'

if (output_table.replace(" ","") in tableList):

if (drop_ex_tbl):

plpy.execute("drop %s %s" %(table_type, output_table))

else:

return ("The %s %s already exists in the database. Please check and try again." %(table_type, output_table))

# Management of sort order parameter

if sort_order.replace(" ","") == 'asc':

ordering="(select %s, %s(%s) as ordervalue from %s group by %s order by 2 asc) as t2"%(pivot_field,agg_func,value_field,input_table,pivot_field)

elif sort_order.replace(" ","") == 'desc':

ordering="(select %s, %s(%s) as ordervalue from %s group by %s order by 2 desc) as t2"%(pivot_field,agg_func,value_field,input_table,pivot_field)

elif sort_order.replace(" ","") == 'no_sorting':

ordering="(select distinct %s from %s order by 1) as t2"%(pivot_field,input_table)

else:

return "Wrong sort parameter specification. It must be 'asc', 'desc' or no. Please check and try again."

# GATHERING OF THE PIVOTED COLUMN NAMES AND THEIR DATA TYPES.

# First we get the names of the destination fields, which are the values present in

# the pivot_field column

destColumns = plpy.execute("select %s as columns from %s " %(pivot_field, ordering))

# We need this for the final field splitting

destColumns2 = plpy.execute("select distinct %s as column from %s order by 1" %(pivot_field,input_table))

# Now we get the column type. Because all the output pivoted columns

# will have the same data type than input columns, its definition is

# simple in the case of integer/double columns, but for numeric columns

# we need to get both the precision (number of decimals) and the maximum

# value of its aggregate, to prevent overflows when aggregating the data

# First we make the query to pg_attribute to get columns details.

columnPropertiesQuery = plpy.execute("SELECT atttypid::regtype AS data_type, CASE WHEN atttypid IN (21, 23, 20) THEN 0 WHEN atttypid IN (1700) THEN CASE WHEN atttypmod = -1 THEN null ELSE (atttypmod - 4) & 65535 END ELSE null END AS numeric_scale from pg_attribute where attrelid = (current_schema()||'.'||'%s')::regclass and attname = '%s' and attnum > 0 and not attisdropped" %(input_table.replace(" ",""),value_field.replace(" ","")))

# Due to the nature of the PLyResult object returned by plpy.execute,

# getting the columns properties is a little tricky

columnProperties = [property for property in columnPropertiesQuery]

columnType = columnProperties[0]["data_type"]

columnScale = columnProperties[0]["numeric_scale"]

# Because the percentage values range from 0 to 1,

# we need to make room for more decimal numbers for

# precision not be lost

if as_percentage:

fieldType = "numeric(6,5)"

else:

if columnType == 'numeric' or agg_func =='avg':

if columnScale == None:

columnScale = 8

maxFieldValue = plpy.execute("select %s(%s) as agg_funcres from %s" %(agg_func, value_field,input_table))

magnitudeOrder = math.log10(maxFieldValue[0]["agg_funcres"])

numericFieldWidth = magnitudeOrder + columnScale + 1

fieldType = ("numeric(%i, %i)" %(numericFieldWidth, columnScale))

elif columnType in ('bigint','smallint','integer','real','double precision'):

fieldType = columnType

else:

return "Your pivot column %s is not of a numeric type. Please check and try again." %pivot_field

flds = ['"%s" %s'%(destColumn["column"],fieldType) for destColumn in destColumns2]

fields = ','.join(flds)

gfields=group_fields.replace(" ","").split(',')

groupingColumns=list()

idx=1

for field in gfields:

columnType = plpy.execute("SELECT format_type(atttypid, atttypmod) AS data_type from pg_attribute where attrelid = (current_schema()||'.'||'%s')::regclass and attname = '%s' and attnum > 0 and not attisdropped" %(input_table.replace(" ",""),field.replace(" ","")))

tc=[columna["data_type"] for columna in columnType]

# We need this to recover the grouped fields after crosstab execution

nc = ("split_part(trim(joinedcols,'()'),',',%i)::%s as %s" %(idx,tc[0],field))

groupingColumns.append (nc)

idx += 1

groupingColumns=','.join(groupingColumns)

# Construction of the queries to be passed to crosstab function

if as_percentage:

# If we want percentages instead of absolute values, we need to

# create a new table with the percentages, and the queries

# that are passed to crosstab funcion are silightly different

# Notice the use of row(), that makes possible to have crosstabs

# grouped by more than one single field, but to this to be done,

# we have had to do a lot of tricks before (see the use of

# split_part above)

groupbyclause=group_fields + ',' + value_field

plpy.execute("""create temporary table intertable as

(select %s,%s,%s/%s(%s::numeric) over (partition by %s) as percentages from %s group by %s,%s,%s order by %s)"""

%(group_fields,pivot_field,value_field,agg_func,value_field,group_fields,input_table,group_fields,pivot_field,value_field,group_fields))

firstQuery=("""select distinct row(%s),%s,%s(percentages) from intertable

group by %s,%s order by 1,2"""

%(group_fields,pivot_field,agg_func,group_fields,pivot_field))

secondQuery = ("select distinct %s from intertable order by 1" %(pivot_field))

else:

# For absolute values, the queries passed to crosstab are simpler

aggrfun=("%s(%s)" %(agg_func,value_field))

groupbyclause=group_fields

firstQuery=("select distinct row(%s),%s,%s from %s group by %s,%s order by 1,2" %(group_fields,pivot_field,aggrfun,input_table,groupbyclause,pivot_field))

secondQuery = ("select distinct %s from %s order by 1" %(pivot_field,input_table))

# We create the sentence for output fields for crosstab

crossTabQuery = ("joinedcols varchar,%s"%(fields))

# This trick makes possible to get rid of the null values in the pivoted tables.

# TODO: Consider to make it optional, with a parameter for it.

replaceZeros = ['coalesce("{0}",0) as "{0}"'.format(destColumn["columns"]) for destColumn in destColumns]

replaze0 = ','.join(replaceZeros)

# And now, we put everything together and execute the query

finalQuery = ("create %s %s as (select %s,%s from crosstab('%s','%s') as newtable(%s))" %(table_type,output_table,groupingColumns,replaze0,firstQuery,secondQuery,crossTabQuery))

plpy.execute(finalQuery)

# A little cleanup may be necessary

if as_percentage:

pass

plpy.execute("drop table intertable")

return "Your pivoted %s %s has been created." %(table_type, output_table.replace(" ",""))

$$;

postgres_fdw-PostgreSQL外部数据封装器的使用

我的PostgreSQL数据库在Windows上，MADlib不支持，于是使用Docker部署了个Linux版本的PostgreSQL，并且安装了MADlib插件，利用postgres_fdw就可以方便分析Windows下的数据库进行分析了。 postgres_fdw模块提供外部数据封装器的功能，PostgreSQL通过它可以访问存储在外部的 PostgreSQL服务器上的数据。本模块提供的功能不但涵盖老版本中dblink模块实现的功能，而且postgres_fdw提供更加透明和符合标准的语法来访问远程表，并在许多情况下提供更好的性能。使用postgres_fdw模块做远程访问的准备: 使用CREATE EXTENSION语句安装postgres_fdw.

CREATE EXTENSION if not exists  postgres_fdw;

1	CREATE EXTENSION if not exists postgres_fdw;

使用CREATE SERVER语句,为每个需要连接的远程数据库创建一个外部服务器对象。指定除了user和password 以外的连接信息作为服务器对象的选项。

CREATE SERVER foreign_server_99
        FOREIGN DATA WRAPPER postgres_fdw
        OPTIONS (host '192.168.0.99', port '5432', dbname 'ml');

CREATE SERVER foreign_server_99

FOREIGN DATA WRAPPER postgres_fdw

OPTIONS (host '192.168.0.99', port '5432', dbname 'ml');

使用CREATE USER MAPPING语句，为每个需要通过外部服务器访问的数据库创建用户映射。指定远程的和密码作为映射用户的user …

PostgreSQL-利用递归的方法获得一个数组的所有子数组

create or replace function subarrays(inarray anyarray)
returns table (inarr anyarray)
as $$
WITH RECURSIVE r AS (
 SELECT inarray ar1
 UNION ALL
 SELECT array_remove(ar1 ,sub)
 FROM (
 SELECT ar1, unnest(ar1) sub FROM r
 ) x
 WHERE array_length(ar1, 1) > 1
)
SELECT distinct ar1 FROM r ;
 $$language sql;

create or replace function subarrays(inarray anyarray)

returns table (inarr anyarray)

as $$

WITH RECURSIVE r AS (

SELECT inarray ar1

UNION ALL

SELECT array_remove(ar1 ,sub)

FROM (

SELECT ar1, unnest(ar1) sub FROM r

) x

WHERE array_length(ar1, 1) > 1

)

SELECT distinct ar1 FROM r ;

$$language sql;

例子，

select * from subarrays('{a,b,c,d}'::text[]);
-----------------
inarr
{d}
{b,d}
{a,b,d}
{a,d}
{c}
{b,c}
{a,b,c}
{a,c}
{b}
{c,d}
{a,b,c,d}
{a,c,d}
{a,b}
{a}
{b,c,d}

select * from subarrays('{a,b,c,d}'::text[]);

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

inarr

{d}

{b,d}

{a,b,d}

{a,d}

{c}

{b,c}

{a,b,c}

{a,c}

{b}

{c,d}

{a,b,c,d}

{a,c,d}

{a,b}

{a}

{b,c,d}

实际测试下来，存在性能问题，不知道是否仍然有优化空间，另外一个实现

create or replace function subarrays(inarray anyarray, len int)
returns table (inarr anyarray)
as $$
with  a as (select unnest(inarray) aa)
	select array_remove(inarray, aa) from a 
	union select subarrays(array_remove(inarray, aa), len-1) from a where len>2;
$$LANGUAGE SQL;

select  subarrays(array['a','b','c','d'], 4) ;

create or replace function subarrays(inarray anyarray, len int)

returns table (inarr anyarray)

as $$

with a as (select unnest(inarray) aa)

select array_remove(inarray, aa) from a

union select subarrays(array_remove(inarray, aa), len-1) from a where len>2;

$$LANGUAGE SQL;

select subarrays(array['a','b','c','d'], 4) ;

回头比较下两个性能差异（目前看来两个都满足我做关联规则时实现多推多的情况）。 P.S. PL/Python 版本，

CREATE or replace FUNCTION subsets1(itemset text[])
  RETURNS table(a text[])
AS $$
	from itertools import chain, combinations
	#import numpy as np
	#return array
	#return np.array(list(chain(*[combinations(itemset, i + 1) for i, a in enumerate(itemset)])))
	#return list/table
	return list(chain(*[combinations(itemset, i + 1) for i, a in enumerate(itemset)]))
$$ LANGUAGE plpythonu;

CREATE or replace FUNCTION subsets1(itemset text[])

RETURNS table(a text[])

AS $$

from itertools import chain, combinations

#import numpy as np

#return array

#return np.array(list(chain(*[combinations(itemset, i + 1) for i, a in enumerate(itemset)])))

#return list/table

return list(chain(*[combinations(itemset, i + 1) for i, a in enumerate(itemset)]))

$$ LANGUAGE plpythonu;

PostgreSQL-关联规则的纯SQL实现

该SQL是基于Apriori算法的一个尝试(代码比较粗糙），并且只实现了一推一的情况，对照MadLib的结果做了验证，没有问题。

CREATE OR REPLACE FUNCTION assoc.rules()
 RETURNS void
 LANGUAGE plpgsql
AS $function$
  begin
/*
create  view assoc.trans as
select * from trans where purchase_date between '2015-08-01' and '2016-07-31';
*/

 create  TEMP table  assoc_customer_brands ON commit drop as 
 select distinct on(customer_id, brand) customer_id, brand from assoc.trans;
 
 

create TEMP  TABLE  assoc_customer_brands_0 ON commit drop as 
SELECT a.customer_id,a.brand _from, b.brand _to
from assoc_customer_brands a,assoc_customer_brands b
where a.customer_id=b.customer_id and a.brand<>b.brand; 


create TEMP TABLE  assoc_customer_brands_1 ON commit drop as 
select _from, _to, count(*) as count
from assoc_customer_brands_0
  GROUP BY 1,2;
  

CREATE TEMP  TABLE  assoc_customer_brands_2 ON commit drop as 
SELECT brand, count(*) cnt
from assoc_customer_brands
GROUP BY 1;

DROP TABLE IF EXISTS assoc_customer_brands_3;
CREATE TEMP  table assoc_customer_brands_3 ON commit drop as 
  SELECT count(DISTINCT customer_id) as total
  from assoc_customer_brands;
  
-- EXP_CONF=rhs_count/total;
-- LIFT= CONF/EXP_CONF
-- CONF=COUNT/lhs_count
-- SUPPORT= COUNT/total
DROP TABLE IF EXISTS assoc.rules;
CREATE table assoc.rules as
SELECT
    r.cnt::DOUBLE PRECISION/total as exp_conf,
    a.count::DOUBLE PRECISION/l.cnt as conf ,
    a.count::DOUBLE PRECISION/total as support,
    (a.count::DOUBLE PRECISION*total)/(l.cnt*r.cnt)  lift,
   total ,
    a.count,
    a._from||'==>'||a._to as rule,
    a._from _lhand,
    a._to _rhand ,
    a._from item1 ,
    '====================>'::text item2 ,
    a._to item3 ,
    NULL::text item4
from assoc_customer_brands_1 a
,assoc_customer_brands_2 l
,assoc_customer_brands_2 r
,assoc_customer_brands_3 t
where a._from=l.brand
and a._to=r.brand
;
end;
$function$

CREATE OR REPLACE FUNCTION assoc.rules()

RETURNS void

LANGUAGE plpgsql

AS $function$

begin

create view assoc.trans as

select * from trans where purchase_date between '2015-08-01' and '2016-07-31';

create TEMP table assoc_customer_brands ON commit drop as

select distinct on(customer_id, brand) customer_id, brand from assoc.trans;

create TEMP TABLE assoc_customer_brands_0 ON commit drop as

SELECT a.customer_id,a.brand _from, b.brand _to

from assoc_customer_brands a,assoc_customer_brands b

where a.customer_id=b.customer_id and a.brand<>b.brand;

create TEMP TABLE assoc_customer_brands_1 ON commit drop as

select _from, _to, count(*) as count

from assoc_customer_brands_0

GROUP BY 1,2;

CREATE TEMP TABLE assoc_customer_brands_2 ON commit drop as

SELECT brand, count(*) cnt

from assoc_customer_brands

GROUP BY 1;

DROP TABLE IF EXISTS assoc_customer_brands_3;

CREATE TEMP table assoc_customer_brands_3 ON commit drop as

SELECT count(DISTINCT customer_id) as total

from assoc_customer_brands;

-- EXP_CONF=rhs_count/total;

-- LIFT= CONF/EXP_CONF

-- CONF=COUNT/lhs_count

-- SUPPORT= COUNT/total

DROP TABLE IF EXISTS assoc.rules;

CREATE table assoc.rules as

SELECT

r.cnt::DOUBLE PRECISION/total as exp_conf,

a.count::DOUBLE PRECISION/l.cnt as conf ,

a.count::DOUBLE PRECISION/total as support,

(a.count::DOUBLE PRECISION*total)/(l.cnt*r.cnt) lift,

total ,

a.count,

a._from||'==>'||a._to as rule,

a._from _lhand,

a._to _rhand ,

a._from item1 ,

'====================>'::text item2 ,

a._to item3 ,

NULL::text item4

from assoc_customer_brands_1 a

,assoc_customer_brands_2 l

,assoc_customer_brands_2 r

,assoc_customer_brands_3 t

where a._from=l.brand

and a._to=r.brand

;

end;

$function$

关于transaction数据的准备，customer_id 可以是order_id(购物篮）, 也可以基于customer_id, brand可以是品牌，产品，品类中的一种，或者品类和品牌的组合（如果需要的话，比如category||’~’||brand）。结果，可以根据需要对confidence， support和lift进行筛选。 Todo,后续可能会依据PostgreSQL的Array来把所有可能的子集都查出来，然后生成完整的关联规则。

csv2table让PostgreSQL下导入csv不再那么麻烦

PostgreSQL下使用Copy命令导入csv，速度很快，但你得先分析CSV，创建表，有了csv2table，自动创建表并导入数据，一气呵成。

usage: csv2table [-h] --file FILE [--copy] [--backslash]
                 [--delimiter DELIMITER] [--quote QUOTE] [--table TABLE]
                 [--schema SCHEMA] [--create-schema] [--integer] [--numeric]
                 [--timestamp] [--date] [--tz] [--mogrify] [--lower] [--drop]
                 [--truncate] [--skip-parsing SKIP_PARSING] [--big-integer]
                 [--no-create] [--fix-duplicates] [--transaction]
                 [--temporary] [--file-column FILE_COLUMN]
                 [--redshift REDSHIFT] [--redshift-bucket REDSHIFT_BUCKET]
                 [--redshift-upload] [--gzip] [--missing-headers] [--cine]
                 [--keep-going N] [--primary-key PRIMARY_KEY]
                 [--serial-column SERIAL_COLUMN] [--array]

optional arguments:
  -h, --help            show this help message and exit
  --file FILE, -f FILE  csv file
  --copy, -y            issue a copy statement, for import
  --backslash, -s       when issuing copy, use \copy
  --delimiter DELIMITER, -d DELIMITER
                        csv delimiter
  --quote QUOTE, -q QUOTE
                        csv quote
  --table TABLE, -t TABLE
                        table name
  --schema SCHEMA       schema name
  --create-schema       create the named schema
  --integer, -i         integer columns
  --numeric, -n         numeric parsing
  --timestamp, -p       timestamp parsing
  --date                date parsing
  --tz                  use timestamptz instead of timestamp
  --mogrify, -m         clean names
  --lower, -l           lowercase names
  --drop, -x            drop table first
  --truncate            truncate table first
  --skip-parsing SKIP_PARSING
                        skip type parsing for cols
  --big-integer, -b     use bigint instead of int
  --no-create, -z       don't issue create table
  --fix-duplicates      handle duplicate column names by appending an
                        increasing suffix
  --transaction, -1     run in a transaction
  --temporary           create a temporary table
  --file-column FILE_COLUMN
                        add a column to the created table which will contain
                        the name of the file the data was loaded from
  --redshift REDSHIFT   generate a copy command for redshift. takes an
                        optional argument, a file to look for s3
                        credentials/bucket in. To try and read from the
                        environment, specify 
  --redshift-bucket REDSHIFT_BUCKET
                        when generating a copy command for redshift, using
                        this option specifies the bucket to store the file in,
                        if requested. This will override any bucket found by
                        the --redshift option.
  --redshift-upload     upload the specified file to redshift
  --gzip                the file is in gzip format
  --missing-headers     file is missing headers, make up column names
  --cine                create if not exists when creating a table
  --keep-going N        read the N rows instead of the first two rows, trading
                        performance for better type detection
  --primary-key PRIMARY_KEY
                        column(s) to make part of the primary key
  --serial-column SERIAL_COLUMN
                        add serial column of the given name
  --array               attempt to detect and create array columns

usage: csv2table [-h] --file FILE [--copy] [--backslash]

[--delimiter DELIMITER] [--quote QUOTE] [--table TABLE]

[--schema SCHEMA] [--create-schema] [--integer] [--numeric]

[--timestamp] [--date] [--tz] [--mogrify] [--lower] [--drop]

[--truncate] [--skip-parsing SKIP_PARSING] [--big-integer]

[--no-create] [--fix-duplicates] [--transaction]

[--temporary] [--file-column FILE_COLUMN]

[--redshift REDSHIFT] [--redshift-bucket REDSHIFT_BUCKET]

[--redshift-upload] [--gzip] [--missing-headers] [--cine]

[--keep-going N] [--primary-key PRIMARY_KEY]

[--serial-column SERIAL_COLUMN] [--array]

optional arguments:

-h, --help show this help message and exit

--file FILE, -f FILE csv file

--copy, -y issue a copy statement, for import

--backslash, -s when issuing copy, use \copy

--delimiter DELIMITER, -d DELIMITER

csv delimiter

--quote QUOTE, -q QUOTE

csv quote

--table TABLE, -t TABLE

table name

--schema SCHEMA schema name

--create-schema create the named schema

--integer, -i integer columns

--numeric, -n numeric parsing

--timestamp, -p timestamp parsing

--date date parsing

--tz use timestamptz instead of timestamp

--mogrify, -m clean names

--lower, -l lowercase names

--drop, -x drop table first

--truncate truncate table first

--skip-parsing SKIP_PARSING

skip type parsing for cols

--big-integer, -b use bigint instead of int

--no-create, -z don't issue create table

--fix-duplicates handle duplicate column names by appending an

increasing suffix

--transaction, -1 run in a transaction

--temporary create a temporary table

--file-column FILE_COLUMN

add a column to the created table which will contain

the name of the file the data was loaded from

--redshift REDSHIFT generate a copy command for redshift. takes an

optional argument, a file to look for s3

credentials/bucket in. To try and read from the

environment, specify

--redshift-bucket REDSHIFT_BUCKET

when generating a copy command for redshift, using

this option specifies the bucket to store the file in,

if requested. This will override any bucket found by

the --redshift option.

--redshift-upload upload the specified file to redshift

--gzip the file is in gzip format

--missing-headers file is missing headers, make up column names

--cine create if not exists when creating a table

--keep-going N read the N rows instead of the first two rows, trading

performance for better type detection

--primary-key PRIMARY_KEY

column(s) to make part of the primary key

--serial-column SERIAL_COLUMN

add serial column of the given name

--array attempt to detect and create array columns

P.S. 其实对于大的CSV文件，有时我会用

head -50 file.csv&gt;sample.csv

1	head -50 file.csv>sample.csv

这样的命令来取一部分进行分析并用R读取并来导入（自动创建表），然后再用copy来快速导入。

Currently browsing category

以Server的模式运行pgAdmin4

PostgreSQL下查找与去除重复数据

PostgreSQL与交叉表查询

postgres_fdw-PostgreSQL外部数据封装器的使用

PostgreSQL-利用递归的方法获得一个数组的所有子数组

PostgreSQL-关联规则的纯SQL实现

csv2table让PostgreSQL下导入csv不再那么麻烦