从Python scrip更新SQL表的多行

Question

我有一个庞大的表（超过100B条记录），我在其中添加了一个空列。我分析另一个字段（string）中的字符串如果所需的字符串可用，则从该字段中提取一个整数，并希望在新列中为具有该字符串的所有行更新它。你知道吗

目前，在数据被解析并保存在dataframe中之后，我对它进行迭代，用干净的数据更新Redshift表。每次迭代大约需要1秒，这太长了。你知道吗

我当前的代码示例：

conn = psycopg2.connect(connection_details)
cur = conn.cursor()
clean_df = raw_data.apply(clean_field_to_parse)
for ind, row in clean_df.iterrows():
  update_query = build_update_query(row.id, row.clean_integer1, row.clean_integer2)
  cur.execute(update_query)

其中update_query是生成更新查询的函数：

def update_query(id, int1, int2):
  query = """
  update tab_tab
  set 
  clean_int_1 = {}::int,
  clean_int_2 = {}::int,
  updated_date = GETDATE()
  where id = {}
  ;
  """
  return query.format(int1, int2, id)

其中，干净的df的结构如下：

id . field_to_parse . clean_int_1 . clean_int_2
1  . {'int_1':'2+1'}.      3      .    np.nan
2  . {'int_2':'7-0'}.     np.nan  .      7

有没有办法批量更新特定的表字段，这样就不需要一次执行一个查询了？你知道吗

我正在解析字符串并从Python运行update语句。数据库存储在红移上。你知道吗

Answer 1

如前所述，考虑纯SQL，通过将Pandas数据帧作为一个staging表推送到Postgres，然后在两个表中运行一个UPDATE，避免遍历数十亿行。使用SQLAlchemy，您可以使用^{}创建数据帧的表副本。甚至可以添加连接字段的索引，id，并在末尾删除非常大的临时表。你知道吗

from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2://myuser:mypwd!@myhost/mydatabase")

# PUSH TO POSTGRES (SAME NAME AS DF)
clean_df.to_sql(name="clean_df", con=engine, if_exists="replace", index=False)

# SQL UPDATE (USING TRANSACTION)
with engine.begin() as conn:     

    sql = "CREATE INDEX idx_clean_df_id ON clean_df(id)"
    conn.execute(sql)

    sql = """UPDATE tab_tab t
             SET t.clean_int_1 = c.int1,
                 t.clean_int_2 = c.int2,
                 t.updated_date = GETDATE()
             FROM clean_df c
             WHERE c.id = t.id
          """
    conn.execute(sql)

    sql = "DROP TABLE IF EXISTS clean_df"
    conn.execute(sql)

engine.dispose()