python - 将新列从 Pandas 添加到 SQLite 表的工作流程

Question

设置

两个表:学校和学生。 SQLite 中的索引(或键)对于 students 表和 school 和 将是 id 和 time schools 表的 >time。我的数据集是关于一些不同的东西，但我认为学校学生的例子更容易理解。

import pandas as pd
import numpy as np
import sqlite3

df_students = pd.DataFrame(
{'id': list(range(0,4)) + list(range(0,4)),
'time': [0]*4 + [1]*4, 'school': ['A']*2 + ['B']*2 + ['A']*2 + ['B']*2,
'satisfaction': np.random.rand(8)} )
df_students.set_index(['id', 'time'], inplace=True)

        satisfaction    school
id  time        
0   0   0.863023    A
1   0   0.929337    A
2   0   0.705265    B
3   0   0.160457    B
0   1   0.208302    A
1   1   0.029397    A
2   1   0.266651    B
3   1   0.646079    B

df_schools = pd.DataFrame({'school': ['A']*2 + ['B']*2, 'time': [0]*2 + [1]*2, 'mean_scores': np.random.rand(4)})
df_schools.set_index(['school', 'time'], inplace=True)
df_schools


               mean_scores
school  time    
A       0     0.358154
A       0     0.142589
B       1     0.260951
B       1     0.683727

## Send to SQLite3

conn = sqlite3.connect('schools_students.sqlite')

df_students.to_sql('students', conn)
df_schools.to_sql('schools', conn)

我需要做什么？

我有一堆函数可以操作 pandas 数据框并创建新的列，然后应该将这些列插入 schools 或 students 表(取决于我正在构建的内容)。一个典型的函数按顺序执行:

1. 从两个 SQL 表中查询列
2. 使用pandas函数，例如groupby、自定义函数的apply、rolling_mean等(许多它们在 SQL 上不可用，或者难​​以编写)来构造一个新列。返回类型是 pd.Series 或 np.array
3. 将新列添加到适当的数据框(学校 或学生)

这些函数是在我有一个适合内存的小型数据库时编写的，因此它们是纯 pandas。

这是一个伪代码示例:

def example_f(satisfaction, mean_scores)
    """Silly function that divides mean satisfaction per school by mean score"""
    #here goes the pandas functions I already wrote
    mean_satisfaction = mean(satisfaction) 
    return mean_satisfaction/mean_scores

satisf_div_score = example_f(satisfaction, mean_scores)
# Here push satisf_div_score to `schools` table

因为我的数据集非常大，所以我无法在内存中调用这些函数。想象一下，学校位于不同的地区。原来我只有一个区，所以我知道这些函数可以分别处理每个区的数据。

我认为可行的工作流程是:

• 查询i地区的相关数据
• 对 i 区的数据应用函数并生成新列作为 np.array 或 pd.Series
• 在适当的表中插入此列(将为该列的 i 区填充数据
• 重复从 i = 1 到 K 的地区

虽然我的数据集在 SQLite 中(我希望它保持这种状态!)但如果好处很大，我愿意将它迁移到其他东西。

---

我知道有不同的合理答案，但很高兴听到一些已证明对您有用且简单的答案。谢谢!

Answer 1

有几种方法，您可以选择哪种更适合您的特定任务:

1. 将所有数据移动到“更大”的数据库。我个人更喜欢 PostgreSQL——它在大数据集上表现得很好。 幸运的是，pandas 支持 SQLAlchemy - 跨数据库 ORM，因此您可以对不同的数据库使用相同的查询。

2. 将数据分成 block 并分别计算任何 block 。我将使用 PostgreSQL 对其进行演示，但您可以使用任何数据库。

   from sqlalchemy import create_engine
   import psycopg2
   mydb = create_engine('postgresql://user@host.domain:5432/database')
   # lets select some groups of data into first dataframe, 
   # you may use school ids instead of my sections
   df=pd.read_sql_query('''SELECT sections, count(id) FROM table WHERE created_at <'2016-01-01' GROUP BY sections ORDER BY 2 DESC LIMIT 10''', con=mydb)
   print(df)  # don't worry about strange output - sections have type int[] and it's supported well!
   
      sections     count
   0  [121, 227]  104583
   1  [296, 227]   48905
   2  [121]        43599
   3  [302, 227]   29684 
   4  [298, 227]   26814
   5  [294, 227]   24071
   6  [297, 227]   23038
   7  [292, 227]   22019
   8  [282, 227]   20369
   9  [283, 227]   19908
   
   # Now we have some sections and we can select only data related to them
   for section in df['sections']:
      df2 = pd.read_sql_query('''SELECT sections, name, created_at, updated_at, status 
                                 FROM table 
                                 WHERE created_at <'2016-01-01'   
                                     AND sections=%(section)s 
                                 ORDER BY created_at''', 
                              con=mydb, params=dict(section=section))
       print(section, df2.std())
   
   [121, 227] status    0.478194
   dtype: float64
   [296, 227] status    0.544706
   dtype: float64
   [121] status    0.499901
   dtype: float64
   [302, 227] status    0.504573
   dtype: float64
   [298, 227] status    0.518472
   dtype: float64
   [294, 227] status    0.46254
   dtype: float64
   [297, 227] status    0.525619
   dtype: float64
   [292, 227] status    0.627244
   dtype: float64
   [282, 227] status    0.362891
   dtype: float64
   [283, 227] status    0.406112
   dtype: float64
   

   当然这个例子是合成的——计算文章的平均状态是非常荒谬的:)但它演示了如何拆分大量数据并将其分成几部分。

3. 使用特定的 PostgreSQL(或 Oracle 或 MS 或任何您喜欢的)进行统计。这是关于 Window Functions in PostgreSQL 的优秀文档.幸运的是，您可以在 DB 中执行一些计算并将预制数据移动到 DataFrame，如上所示。

更新:如何将信息加载回数据库。

还好DataFrame支持方法to_sql使这个过程变得简单:

from sqlalchemy import create_engine
mydb = create_engine('postgresql://user@host.domain:5432/database')
df2.to_sql('tablename', mydb, if_exists='append', chunksize=100)

您可以指定您需要的操作:if_exists='append' 将行添加到表中，如果您有很多行，您可以将它们拆分成 block ，以便数据库可以插入它们。