Azure Data Factory管道取消运行后如何删除已导入PSQL的数据？

方案1：Python取消操作联动PSQL清理（最直接高效） #

既然你已经通过Python触发和取消管道，直接在取消逻辑后追加PSQL数据清理是最简便的方式：

• 先维护好ADF管道中涉及的PSQL目标表清单，确保和复制活动的目标表一一对应
• 在调用取消管道运行的函数后，立即连接PSQL数据库，对目标表执行清理语句（TRUNCATE比DELETE性能更好，根据权限和业务需求选择）

示例Python代码片段：

import psycopg2
from azure.mgmt.datafactory import DataFactoryManagementClient

# 你已有的管道取消函数
def cancel_adf_pipeline(df_client, rg_name, df_name, run_id):
    df_client.pipeline_runs.cancel(rg_name, df_name, run_id)

# PSQL清理函数
def clean_psql_targets(psql_config, target_tables):
    conn = psycopg2.connect(
        host=psql_config["host"],
        database=psql_config["db"],
        user=psql_config["user"],
        password=psql_config["pwd"]
    )
    cursor = conn.cursor()
    try:
        for table in target_tables:
            # 用TRUNCATE需确保账号有对应权限，若用DELETE则替换为DELETE FROM {table};
            cursor.execute(f"TRUNCATE TABLE {table};")
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cursor.close()
        conn.close()

# 主执行逻辑
if __name__ == "__main__":
    # 初始化ADF客户端、参数等（你的现有代码）
    # ...
    # 先取消管道运行
    cancel_adf_pipeline(df_client, "your-rg", "your-df", "target-run-id")
    # 再清理PSQL数据
    psql_config = {"host": "psql-host", "db": "target-db", "user": "db-user", "pwd": "db-pwd"}
    target_tables = ["table_a", "table_b", "table_c", "table_d"]
    clean_psql_targets(psql_config, target_tables)

方案2：ADF管道内状态标记+独立清理管道（适合依赖ADF运行状态的场景） #

如果希望清理逻辑依托ADF执行，可以通过状态标记实现：

• 步骤1：在主管道开头添加Script活动，在PSQL中创建临时状态表（或复用现有状态表），记录本次管道Run ID及待复制的表清单，状态设为running
• 步骤2：每个复制活动完成后，用Script活动更新对应表的状态为copied
• 步骤3：Python取消管道后，触发一个独立的ADF清理管道：
  • 用Lookup活动读取状态表中关联当前Run ID且状态为running/copied的表
  • 用Foreach活动遍历这些表，调用Script活动执行PSQL清理语句

关键注意事项 #

• 权限：执行清理的PSQL账号需拥有目标表的DELETE或TRUNCATE权限
• 幂等性：状态表需绑定管道Run ID，避免误清理其他运行批次的数据
• 语句选择：TRUNCATE执行更快但会重置自增序列且无法回滚，DELETE支持条件删除但性能较差，按需选择

内容的提问来源于stack exchange，提问作者Jason Lo