pandas - Pandas 或 psycopg2 中的 Postgres 9.5 upsert 命令？

Question

我看到的大多数示例都是人们使用 ON CONFLICT DO UPDATE 语法将单行插入到数据库中。

有没有人有任何使用 SQLAlchemy 或 pandas.to_sql 的例子？

我 99% 的插入使用 psycopg2 COPY 命令(所以我保存了 csv 或 stringio，然后批量插入)，另外 1% 是 pd.to_sql。我检查新行或新维度的所有逻辑都是在 Python 中完成的。

def find_new_rows(existing, current, id_col):
        current[id_col] = current[id_col].astype(int)
        x = existing[['datetime', id_col, 'key1']]
        y = current[['datetime', id_col, 'key2']]
        final = pd.merge(y, x, how='left', on=['datetime', id_col])
        final = final[~(final['key2'] == final['key1'])]
        final = final.drop(['key1'], axis=1)
        current = pd.merge(current, final, how='left', on=['datetime', id_col])
        current = current.loc[current['key2_y'] == 1]
        current.drop(['key2_x', 'key2_y'], axis=1, inplace=True)
        return current

有人可以向我展示一个使用新的 PostgreSQL 语法通过 pyscopg2 进行 upsert 的示例吗？一个常见的用例是检查维度更改(每天 50k - 100k 行之间，我将其与现有值进行比较)，这是 CONFLICT DO NOTHING 仅添加新行。

另一个用例是我有随时间变化的事实数据。我只取最近的值(我目前使用 View 来选择不同的)，但如果可能的话，最好使用 UPSERT。

Answer 1

仅供引用，这是我目前使用的解决方案。

它似乎适合我的目的。我不得不添加一行来用 None 替换空(NaT)时间戳，因为我在将每一行加载到数据库中时遇到错误。

def create_update_query(table):
    """This function creates an upsert query which replaces existing data based on primary key conflicts"""
    columns = ', '.join([f'{col}' for col in DATABASE_COLUMNS])
    constraint = ', '.join([f'{col}' for col in PRIMARY_KEY])
    placeholder = ', '.join([f'%({col})s' for col in DATABASE_COLUMNS])
    updates = ', '.join([f'{col} = EXCLUDED.{col}' for col in DATABASE_COLUMNS])
    query = f"""INSERT INTO {table} ({columns}) 
                VALUES ({placeholder}) 
                ON CONFLICT ({constraint}) 
                DO UPDATE SET {updates};"""
    query.split()
    query = ' '.join(query.split())
    return query


def load_updates(df, table, connection):
    conn = connection.get_conn()
    cursor = conn.cursor()
    df1 = df.where((pd.notnull(df)), None)
    insert_values = df1.to_dict(orient='records')
    for row in insert_values:
        cursor.execute(create_update_query(table=table), row)
        conn.commit()
    row_count = len(insert_values)
    logging.info(f'Inserted {row_count} rows.')
    cursor.close()
    del cursor
    conn.close()