Spark覆盖PostgreSQL表后触发器有效性及行不全问题排查

在外部Spark程序覆盖PostgreSQL中的某张表后，希望通过触发器利用该表的数据在另一张表执行upsert操作并做部分修改。请问此场景下触发器是否可用？若可用，应使用UPDATE还是INSERT触发器？

补充信息 #

PySpark写入表代码 #

df.write.format("jdbc")
        .option("truncate","true")
        .option("driver", "org.postgresql.Driver")
        .option("url", postgres_host)
        .option("user", postgres_user)
        .option("password", postgres_password)
        .option("dbtable", table_name)
        .mode("overwrite")
        .save()

触发器及函数示例 #

CREATE
    OR REPLACE FUNCTION test_trigger() 
    RETURNS TRIGGER AS $func$ BEGIN IF NOT EXISTS (
    SELECT
    FROM
        information_schema.tables
    WHERE
        table_schema = 'public'
        AND table_name = 'test_table'
) THEN EXECUTE 'CREATE TABLE IF NOT EXISTS test_table as
            SELECT *,ST_GeomFromText(ST_AsText(col), 4326) as geom 
            FROM  spark_table';
END IF;
RETURN NULL;
END;

$func$ LANGUAGE plpgsql;

CREATE TRIGGER insert_or_update_parcel_1
  AFTER INSERT OR UPDATE
  ON spark_table
  FOR EACH STATEMENT
  EXECUTE PROCEDURE create_table_trigger();

测试情况：触发器可生效但仅读取被覆盖表中的一行数据，而非全部。

1. 触发器可用性：此场景下触发器完全可用，但需匹配Spark写入PostgreSQL的行为逻辑。
   你使用mode("overwrite") + option("truncate","true")时，Spark的执行流程是先TRUNCATE TABLE清空原表，再批量插入新数据。此时只有INSERT触发器会触发，因为所有新数据都是插入操作，没有更新现有行的动作，UPDATE触发器不会被触发。

2. 当前触发器的核心问题：

   • 你的触发器是语句级触发器（FOR EACH STATEMENT），且逻辑仅在test_table不存在时创建表。Spark批量插入是分批次执行的，每批次会触发一次语句级触发器，但你的逻辑只在第一次触发时执行CREATE TABLE ... AS SELECT，此时spark_table中只有第一批次的少量数据（可能仅一行），后续批次触发时因表已存在不会执行任何操作，最终导致test_table仅包含部分数据。

3. 修正后的实现方案：

   • 提前创建好test_table（结构与spark_table一致，额外添加geom列），避免在触发器中动态建表。
   • 根据数据量选择触发器类型：
     • 行级触发器（适合中小数据量）：逐行处理upsert，逻辑简单直观

       CREATE OR REPLACE FUNCTION test_trigger() 
       RETURNS TRIGGER AS $func$
       BEGIN
           -- 假设id是主键，根据实际主键调整匹配条件
           INSERT INTO test_table (col1, col2, col, geom)
           VALUES (NEW.col1, NEW.col2, NEW.col, ST_GeomFromText(ST_AsText(NEW.col), 4326))
           ON CONFLICT (id) DO UPDATE 
           SET col1 = EXCLUDED.col1, col2 = EXCLUDED.col2, col = EXCLUDED.col, geom = EXCLUDED.geom;
           RETURN NULL;
       END;
       $func$ LANGUAGE plpgsql;
       
       CREATE TRIGGER spark_table_insert_trigger
         AFTER INSERT ON spark_table
         FOR EACH ROW
         EXECUTE PROCEDURE test_trigger();
       

     • 语句级触发器（适合大数据量）：批量处理upsert，减少触发器执行次数，提升性能

       CREATE OR REPLACE FUNCTION test_trigger_batch() 
       RETURNS TRIGGER AS $func$
       BEGIN
           INSERT INTO test_table (col1, col2, col, geom)
           SELECT col1, col2, col, ST_GeomFromText(ST_AsText(col), 4326)
           FROM spark_table
           ON CONFLICT (id) DO UPDATE 
           SET col1 = EXCLUDED.col1, col2 = EXCLUDED.col2, col = EXCLUDED.col, geom = EXCLUDED.geom;
           RETURN NULL;
       END;
       $func$ LANGUAGE plpgsql;
       
       CREATE TRIGGER spark_table_insert_batch_trigger
         AFTER INSERT ON spark_table
         FOR EACH STATEMENT
         EXECUTE PROCEDURE test_trigger_batch();
       

4. 额外注意点：

   • Spark的TRUNCATE操作不会触发触发器，如果需要在表被清空时执行逻辑，需单独添加AFTER TRUNCATE触发器。
   • 使用语句级批量upsert时，若Spark分批次插入，需确保重复执行upsert不会引发数据异常（依赖PostgreSQL的ON CONFLICT逻辑处理重复数据）。

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