1
2
3
4
5
6
7
8

--对包含;的列进行拆分，然后转为行。注意：UNION ALL 前后的列顺序必须一致
SELECT
    --与 UNION ALL 关键字 后面的列顺序必须一致
	phone,
	UNNEST(regexp_split_to_array(introduce,	';')) introduce,
	name
FROM
	read_csv('testfile.csv', delim=',', header=True, columns={'phone':'BIGINT','introduce':'VARCHAR','name':'VARCHAR'})

1
2
3
4
5
6
7

SELECT
	phone,
	UNNEST(str_split_regex(introduce,	'[;|]')) introduce,
	name
	
FROM
	read_csv('testfile.csv', delim=',', header=True, columns={'phone':'BIGINT','introduce':'VARCHAR','name':'VARCHAR'})

1

COPY (SELECT * FROM tbl) TO 'output.csv' (HEADER, DELIMITER ',');

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

--对包含;的列进行拆分，然后转为行。注意：UNION ALL 前后的列顺序必须一致
COPY (
	SELECT
		phone,
		UNNEST(regexp_split_to_array(introduce,	'[;|]')) introduce,
		name
	FROM
		read_csv('testfile.csv',delim = ',',header = True,columns = {'phone':'BIGINT','introduce':'VARCHAR','name':'VARCHAR' })
	WHERE
		regexp_matches(introduce,
	'[;|]')
	UNION ALL
	SELECT
		*
	FROM
		read_csv('testfile.csv',delim = ',',header = True,columns = {'phone':'BIGINT','introduce':'VARCHAR','name':'VARCHAR' })
	WHERE
		not regexp_matches(introduce,'[;|]')
)
  TO 'output.csv' (HEADER, DELIMITER ',');

1

COPY (SELECT DISTINCT A,B,C,D FROM '待去重CSV文件路径') TO 'output.csv' (HEADER, DELIMITER ',');

1

COPY (SELECT A,B,C,D,ANY_VALUE(E),ANY_VALUE(F) FROM '待去重CSV文件路径' GROUP BY A,B,C,D) TO 'output.csv' (HEADER, DELIMITER ',');

1

COPY (SELECT A,B,C,D,MAX(E) 最晚日期时间,MIN(e) 最早日期时间 FROM '待去重CSV文件路径' GROUP BY A,B,C,D) TO 'output.csv' (HEADER, DELIMITER ',');

1

SELECT column_name	FROM information_schema.columns	WHERE table_name = '表名';

1

SELECT ARRAY_AGG(column_name)	FROM information_schema.columns	WHERE table_name = '表名';

1

SELECT row(column_name,data_type) as name_type FROM information_schema.columns WHERE table_name = '表名';

1

SELECT * FROM information_schema.columns WHERE table_name = '表名';

1

SELECT * FROM information_schema.schemata;

1

SELECT table_name FROM information_schema.tables WHERE table_catalog = 'test';

1

PRAGMA show_tables;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

-- 设置内存大小限制
SET memory_limit='10GB';
-- configure the system to use 1 thread 待测试作用
SET threads TO 1;

-- 查看所有可用设置，同时可在网页中查看所有可用设置：https://duckdb.org/docs/sql/configuration#configuration-reference
SELECT * FROM duckdb_settings();
-- 查看特定的设置值
SELECT current_setting('access_mode');
-- 将内存限制重置为默认，默认好像是程序自动的。
RESET memory_limit;

1
2
3
4

-- 设置内存大小限制
PRAGMA memory_limit='1GB';
-- 设置并行查询数
PRAGMA threads=4;

1

select * from duckdb_extensions();

1
2

INSTALL spatial;
LOAD spatial;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

import duckdb

def find_all_son_phones(phone): 
    path = [phone]
    def find_son_phones(phone):
        # phone对应的id
        conn.execute(f"select id from datas where phone='{phone}';")
        current_id = conn.fetchone()[0]
        # 父ID等于上面ID的Phone,即儿子的Phone
        conn.execute(f"select phone from datas where pid='{current_id}';")
        son_phone_array = conn.fetchone()
        if son_phone_array:  # 有儿子的情况
            son_phone = son_phone_array[0]
            path.append(son_phone)
            find_son_phones(son_phone)
    find_son_phones(phone)
    return path

def add_path(phone):
    """找出特定号码的子孙号码形成路径后保存到path列"""
    tree_list = find_all_son_phones(phone)
    tree_list = [str(i) for i in tree_list]
    if len(tree_list)>1:
        for i,e in enumerate(tree_list):
            conn.execute(f"update datas set path = '{'->'.join(tree_list)}',level={i} where phone ={e};")

def gen_phone_trees():
	# 全表生成数关系
    root_phones = conn.execute("select phone from datas where id NOTNULL AND pid ISNULL;").fetchall()
    root_phones = [j for i in root_phones for j in i]
    for phone in root_phones:
        add_path(phone)

conn = duckdb.connect('test.duckdb')
conn.execute("CREATE TABLE IF NOT EXISTS datas AS SELECT * FROM 'C:\\Users\\xxxxxx\\test.csv';")
conn.execute("ALTER TABLE datas ADD COLUMN IF NOT EXISTS path VARCHAR;")
conn.execute("ALTER TABLE datas ADD COLUMN IF NOT EXISTS level INT;")
# 创建索引，数据量大的时候测试是否有价值【待测试】。
conn.execute("CREATE INDEX p_id_pid_idx ON datas (phone, id,pid);")

gen_phone_trees()
conn.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

import duckdb
import time

count= 0
def find_related_ids(current_id,phone): 
    path_ids = [current_id]
    id_phone_dict = {current_id:phone}

    def find_son_ids(current_id):
        # 父ID等于上面ID的Phone,即儿子的ID
        conn.execute(f"select id, phone from datas where pid='{current_id}';")
        son_ids = conn.fetchall()
        for son_id_tuple in son_ids:
            son_id, son_phone = son_id_tuple
            path_ids.append(son_id)
            id_phone_dict[son_id] = son_phone
            find_son_ids(son_id)
    
    def find_parent_ids(current_id):
        # id等于上面PID的Phone,即父亲的ID
        conn.execute(f"select pid, phone from datas where id='{current_id}';")
        parent_ids = conn.fetchall()
        for pid_tuple in parent_ids:
            parent_id, parent_phone = pid_tuple
            path_ids.insert(0, parent_id)  # 在列表前方插入父亲ID，保持父辈在前，子辈在后的顺序
            id_phone_dict[parent_id] = parent_phone
            find_parent_ids(parent_id)

    find_son_ids(current_id)
    # find_parent_ids(current_id)

    return path_ids, id_phone_dict

def add_path(init_phone):
    global count
    """找出特定号码的上下级路径并形成路径后保存到path列"""
    # phone对应的所有id
    conn.execute(f"select DISTINCT id from datas where phone='{init_phone}';")
    current_ids = [i[0] for i in conn.fetchall()]
    print('该号码共对应ID数：',len(current_ids))
    for current_id in current_ids:
        tree_id_list, id_phone_dict = find_related_ids(current_id,init_phone)
        # tree_id_list = list(filter(None,tree_id_list))
        tree_id_list = list(set(tree_id_list))
        # 更新path和level
        if len(tree_id_list)>1:
            count+=len(tree_id_list)
            print(len(tree_id_list),'\ntree_id_list:\t',tree_id_list)
            print('  id_phone_dict:\t',id_phone_dict)
            path_of_phone = '->'.join([str(id_phone_dict[id]) for id in tree_id_list])
            print('    path_of_phone:\t',path_of_phone,end='\n\n')
            for i, id in enumerate(tree_id_list):
                phone = id_phone_dict[id]
                conn.execute(f"update datas set path = '{path_of_phone}',level={i} where phone='{phone}' and id='{id}';")


with duckdb.connect('test.duckdb') as conn:
    ceshi = ['']
    ceshi_str = ','.join([f"'{item}'" for item in ceshi])
    # 有记录的
    conn.execute(f'SELECT DISTINCT phone FROM datas WHERE phone IN ({ceshi_str});')
    phones = [phone[0] for phone in conn.fetchall()]
    for phone in phones:
        print('当前查找号码：',phone,end=',')
        start = time.time()
        add_path(phone)
        print('耗时：',time.time()-start)
    # add_path("97891706516")
print(count)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import pandas as pd

df = pd.read_csv(r"test.csv", encoding='utf-8', dtype='str',usecols=["name","phone","id","pid"])

def find_all_son_phones(df, phone): 
    path = [phone]
    def find_son_phones(df, phone):
        # phone对应的id
        current_id = df[df['phone'] == phone]['id'].values[0]
        # 父ID等于上面ID的Phone,即儿子的Phone
        son_phone_array = df[df['pid'] == current_id]['phone'].values
        if son_phone_array.any():  # 有儿子的情况
            son_phone = son_phone_array[0]
            path.append(son_phone)
            find_son_phones(df,son_phone)
    find_son_phones(df, phone)
    return path


def add_path(df,phone):
    """找出特定号码的子孙号码形成路径后保存到path列"""
    if 'path' not in df:
        df['path'] = None
    if 'level' not in df:
        df['level']= None
    tree_list = find_all_son_phones(df, phone)
    if len(tree_list)>1:
        for i,e in enumerate(tree_list):
            df.loc[df['phone']==e,'path'] =  '->'.join(tree_list)
            df.loc[df['phone']==e,'level'] = str(i)

# 为所有父ID为空而ID不为空的号码生成子孙路径关系
def gen_phone_trees(df): 
    root_phones = df[(df['id']!='')&(df['pid'].isnull())]['phone'].values
    for phone in root_phones:
        add_path(df,phone)
gen_phone_trees(df)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

import pandas as pd

df = pd.read_csv(r"test.csv", encoding='utf-8', dtype='str',usecols=["name","phone","pphone"])

def find_all_son_phones(df, phone): 
    path = [phone]
    def find_son_phones(df, phone):
        # phone对应的id
        # 父ID等于上面ID的Phone,即儿子的Phone
        son_phone_array = df[df['pphone'] == phone]['phone'].values
        if son_phone_array.any():  # 有儿子的情况
            son_phone = son_phone_array[0]
            path.append(son_phone)
            find_son_phones(df,son_phone)
    find_son_phones(df, phone)
    return path


def add_path(df,phone):
    """找出特定号码的子孙号码形成路径后保存到path列"""
    if 'path' not in df:
        df['path'] = None
    if 'level' not in df:
        df['level']= None
    tree_list = find_all_son_phones(df, phone)
    if len(tree_list)>1:
        for i,e in enumerate(tree_list):
            df.loc[df['phone']==e,'path'] =  '->'.join(tree_list)
            df.loc[df['phone']==e,'level'] = str(i)


# 为所有父ID为空而ID不为空的号码生成子孙路径关系
def gen_phone_trees(df): 
    root_phones = df[(df['phone']!='')&(df['pphone'].isnull())]['phone'].values
    for phone in root_phones:
        add_path(df,phone)
gen_phone_trees(df)

# 找出特定号码的子孙树关系
# add_path(df,'15834961213')

df[~df['path'].isnull()]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

import duckdb
# import functools

conn = duckdb.connect('zfrz.duckdb')
conn.execute("CREATE TABLE IF NOT EXISTS datas AS SELECT * FROM 'C:\\Users\\xxx\\编程\\test.csv';")
conn.execute("ALTER TABLE datas ADD COLUMN IF NOT EXISTS path VARCHAR;")
conn.execute("ALTER TABLE datas ADD COLUMN IF NOT EXISTS level INT;")
# conn.execute("CREATE INDEX p_id_pid_idx ON datas (phone, id,pid);")

def find_all_son_phones(phone): 
    path = [phone]
    def find_son_phones(phone):
        conn.execute(f"select phone from datas where pphone='{phone}';")
        # 父ID等于上面ID的Phone,即儿子的Phone
        son_phone_array = conn.fetchone()
        if son_phone_array:  # 有儿子的情况
            son_phone = son_phone_array[0]
            path.append(son_phone)
            find_son_phones(son_phone)
    find_son_phones(phone)
    return path


def add_path(phone):
    """找出特定号码的子孙号码形成路径后保存到path列"""
    tree_list = find_all_son_phones(phone)
    tree_list = [str(i) for i in tree_list]
    if len(tree_list)>1:
        for i,e in enumerate(tree_list):
            conn.execute(f"update datas set path = '{'->'.join(tree_list)}',level={i} where phone ={e};")


def gen_phone_trees():
    root_phones = conn.execute("select phone from datas where pphone ISNULL;").fetchall()
    root_phones = [j for i in root_phones for j in i]
    for phone in root_phones:
        add_path(phone)

## add_path(15834961213)
gen_phone_trees()

conn.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

WITH RECURSIVE datas_cte AS (
  SELECT id, pid,   ARRAY[id] as path,   0 as level,  id as rootid  FROM datas  WHERE pid IS NULL
  UNION ALL
  SELECT d.id,   d.pid,  
  c.path || d.id,  
  c.level + 1,  c.rootid  FROM datas d
  JOIN datas_cte c ON d.pid = c.id
)
UPDATE datas
SET path = datas_cte.path, 
    level = datas_cte.level, 
    rootid = datas_cte.rootid
FROM datas_cte
WHERE datas.id = datas_cte.id;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

-- 创建一个临时表，将每个id对应的所有电话号码用#连接起来
CREATE TEMP TABLE temp_id_phone AS
SELECT id, STRING_AGG(phone, '#') as phone
FROM datas
GROUP BY id;

-- 找出每个 rootid 组里面 level 最大的记录
WITH max_level AS (
  SELECT rootid, MAX(level) as level
  FROM datas
  GROUP BY rootid
),
max_level_datas AS (
  SELECT d.*
  FROM datas d
  JOIN max_level m ON d.rootid = m.rootid AND d.level = m.level
),
-- 更新这些记录的 path，将 id 替换为对应的电话号码
updated_path AS (
  SELECT mld.id, mld.pid,
         ARRAY_TO_STRING(ARRAY(
           SELECT t.phone::TEXT
           FROM unnest(mld.path::INTEGER[]) p(id)
           JOIN temp_id_phone t ON p.id = t.id
         ), '->') as path
  FROM max_level_datas mld
)
UPDATE datas d
SET path = up.path
FROM updated_path up
WHERE d.id = up.id AND d.pid = up.pid;

-- 删除临时表
DROP TABLE temp_id_phone;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

UPDATE datas AS d1
SET path = d2.path
FROM (
    SELECT rootid, path
    FROM datas
    WHERE level = (
        SELECT MAX(level)
        FROM datas AS d3
        WHERE d3.rootid = datas.rootid
    )
) AS d2
WHERE d1.rootid = d2.rootid;

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
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

--CREATE TABLE datas AS SELECT * FROM 'C:\Users\xxx\Desktop\tree.csv';
--ALTER table datas ADD COLUMN path varchar DEFAULT null;
--ALTER table datas ADD COLUMN rootid varchar DEFAULT null;
--ALTER table datas ADD COLUMN level int DEFAULT null;
---------------------------------------------------------------------------
-- 创建一个递归查询来获取每一行的根节点、路径和级别
WITH RECURSIVE datas_cte AS (
  SELECT id, 
         pid, 
         ARRAY[id] as path, 
         0 as level,
         id as rootid
  FROM datas
  WHERE pid IS NULL
  UNION ALL
  SELECT d.id, 
         d.pid, 
         c.path || ARRAY[d.id], 
         c.level + 1,
         c.rootid
  FROM datas d
  JOIN datas_cte c ON d.pid = c.id
)
UPDATE datas
SET path = datas_cte.path, 
    level = datas_cte.level, 
    rootid = datas_cte.rootid
FROM datas_cte
WHERE datas.id = datas_cte.id;
---------------------------------------------------------------------------

UPDATE datas SET path=null,level=null,rootid=null;

---------------------------------------------------------------------------
-- 找出rootid个数大于1的所有记录，即树形节点大于1的记录
SELECT d.*
FROM datas d
WHERE d.rootid IN (
  SELECT rootid
  FROM datas
  GROUP BY rootid
  HAVING COUNT(*) > 1
);

---------------------------------------------------------------------------
-- 创建一个中间表，将每个id对应的所有电话号码用#连接起来
CREATE TEMP TABLE temp_id_phone AS
SELECT id, STRING_AGG(phone, '#') as phone
FROM datas
GROUP BY id;

-- 更新路径，将id替换为对应的电话号码
WITH RECURSIVE datas_cte AS (
  SELECT d.id, 
         d.pid, 
         ARRAY[t.phone::TEXT] as path, 
         0 as level,
         d.id as rootid
  FROM datas d
  JOIN temp_id_phone t ON d.id = t.id
  WHERE pid IS NULL
  UNION ALL
  SELECT d.id, 
         d.pid, 
         c.path || ARRAY[t.phone::TEXT], 
         c.level + 1,
         c.rootid
  FROM datas d
  JOIN datas_cte c ON d.pid = c.id
  JOIN temp_id_phone t ON d.id = t.id
)
UPDATE datas
SET path = ARRAY_TO_STRING(datas_cte.path, '->'), 
    level = datas_cte.level, 
    rootid = datas_cte.rootid
FROM datas_cte
WHERE datas.id = datas_cte.id;

-- 删除临时表
DROP TABLE temp_id_phone;
---------------------------------------------------------------------------


---------------------------------------------------------------------------
-- 将level值最大的path列数组中的id替换成phone
-- 创建一个临时表，将每个id对应的所有电话号码用#连接起来
CREATE TEMP TABLE temp_id_phone AS
SELECT id, STRING_AGG(phone, '#') as phone
FROM datas
GROUP BY id;

-- 找出每个 rootid 组里面 level 最大的记录
WITH max_level AS (
  SELECT rootid, MAX(level) as level
  FROM datas
  GROUP BY rootid
),
max_level_datas AS (
  SELECT d.*
  FROM datas d
  JOIN max_level m ON d.rootid = m.rootid AND d.level = m.level
),
-- 更新这些记录的 path，将 id 替换为对应的电话号码
updated_path AS (
  SELECT mld.id, mld.pid,
         ARRAY_TO_STRING(ARRAY(
           SELECT t.phone::TEXT
           FROM unnest(mld.path::INTEGER[]) p(id)
           JOIN temp_id_phone t ON p.id = t.id
         ), '->') as path
  FROM max_level_datas mld
)
UPDATE datas d
SET path = up.path
FROM updated_path up
WHERE d.id = up.id AND d.pid = up.pid;

-- 删除临时表
DROP TABLE temp_id_phone;
---------------------------------------------------------------------------


---------------------------------------------------------------------------
--将rootid相同的所有行的path设置为：rootid相同的对应level列值最大的对应的path的值。
UPDATE datas AS d1
SET path = d2.path
FROM (
    SELECT rootid, path
    FROM datas
    WHERE level = (
        SELECT MAX(level)
        FROM datas AS d3
        WHERE d3.rootid = datas.rootid
    )
) AS d2
WHERE d1.rootid = d2.rootid;
---------------------------------------------------------------------------