在 DataFrame 和 SQL Server 表之间加入，而不使用 #temp 表

Question

这是我所拥有的：

• Pandas 数据框（名为 df）

  • 2 列：c1 和 c2

• SQL 中的表（名为 sql_table）

  • 3 列：c3、c4 和 c5

• SQL 中的目标表（名为 target_table）

  • 5 个空列：c1、c2、c3、c4 和 c5

• c1（来自 df）可以与 c3（来自 sql_table）连接

• c2（来自 df）可以与 c4（来自 sql_table）连接

df 中的所有记录都必须显示（暗示外连接）

   INSERT INTO target_table (c1, c2, c3, c4, c5)

   SELECT c1, c2, c3, c4, 
          CASE WHEN t1.c5 IS NOT NULL THEN t1.c5
               WHEN t2.c5 IS NOT NULL THEN t2.c5
               ELSE NULL END AS c5
    FROM dataframe as df
    LEFT JOIN sql_table AS t1 ON df.c1 = t1.c3 
    LEFT JOIN sql_table AS t2 ON df.c2 = t2.c4;

我知道这个查询（和语法）不正确，但这是一般的想法。
目前，我在 SQL 中有 3 个表：

• temp_target _table
• sql_table
• 目标表

到目前为止，我正在将 df（从 Python）导入 temp_target_table（通过 sqlalchemy）并手动执行连接（在 SQL 中）并将这些结果插入到 target_table 中。我想避免使用 temp_target_table。

Answer 1

这是我为解决我的问题所做的：

query = '''
        INSERT INTO target_table (c1, c2, c3, c4, c5)

        SELECT c1, c2, c3, c4, 
               CASE WHEN t1.c5 IS NOT NULL THEN t1.c5
                    WHEN t2.c5 IS NOT NULL THEN t2.c5
                    ELSE NULL END AS c5
        FROM (VALUES (?, ?)) as df (c1, c2)
        LEFT JOIN sql_table AS t1 ON df.c1 = t1.c3 
        LEFT JOIN sql_table AS t2 ON df.c2 = t2.c4;
        '''
for row in dataframe.itertuples():
    cursor.execute(query, row.c1, row.c2)

connection.commit()

Answer 2

SQL Server 2016 版及更高版本（包括 Azure SQL）包含一个 OPENJSON 函数，允许我们以 JSON 形式传递类似表格的数据。将它与 pandas 的 .to_json() 方法和 CTE 结合起来，我们可以做这样的事情：

import pandas as pd
from sqlalchemy import create_engine, text

engine = create_engine(
    "mssql+pyodbc://scott:tiger^5HHH@192.168.0.199/test"
    "?driver=ODBC+Driver+17+for+SQL+Server"
)

# server-side table
with engine.begin() as conn:
    conn.exec_driver_sql("DROP TABLE IF EXISTS sql_table")
    conn.exec_driver_sql(
        "CREATE TABLE sql_table (user_id varchar(50), employee_no varchar(10))"
    )
    conn.exec_driver_sql(
        "INSERT INTO sql_table (user_id, employee_no) VALUES ('gord', '12345')"
    )

# dataframe "table" to join
df = pd.DataFrame([("gord", "gord@example.com")], columns=["user_id", "email"])

# note: requires SQL Server 2016 or later (including Azure SQL)
sql = """\
WITH df_table AS (
    SELECT user_id, email
    FROM OPENJSON(:df_data)
    WITH (
        user_id varchar(50) '$.user_id',
        email varchar(255) '$.email'
    )
)
SELECT st.user_id, st.employee_no, dt.email
FROM sql_table st INNER JOIN df_table dt ON st.user_id = dt.user_id
"""

with engine.begin() as conn:
    results = conn.execute(
        text(sql), {"df_data": df.to_json(orient="records")}
    ).fetchall()
    print(results)
    # [('gord', '12345', 'gord@example.com')]