SESAM中REST Sink的实体删除操作处理最佳实践问询

场景说明 #

传入的Vendor实体通过post-vendor操作发送至REST Sink，在BigQuery中创建供应商；当该供应商从源数据集被删除时，对应实体的_deleted字段值为true，需要触发delete-vendor操作完成BigQuery侧的删除。

现有系统操作配置 #

"operations": {
  "delete-vendor": {
    "method": "DELETE",
    "rate_limiting_delay": 60,
    "rate_limiting_retries": 3,
    "url": "rest/vendor/{{_id}}"
  },
  "post-vendor": {
    "method": "POST",
    "payload-type": "json",
    "url": "rest/vendor"
  }
}

尝试但无效的Pipe配置 #

{
  "_id": "vendor-rest",
  "type": "pipe",
  "source": {
    "type": "dataset",
    "dataset": "global-vendor"
  },
  "transform": {
    "type": "dtl",
    "rules": {
      "default": [
        ["if",
          ["eq", "_S._deleted", true],
          ["add", "operation", "delete-vendor"],
          ["add", "operation", "post-vendor"]
        ],
        ["add", "payload",
          ["apply", "payload", "_S."]
        ]
      ],
      "payload": [
        ["copy", "*",
          ["list", "_id", "_deleted", "_ts", "_updated", "_previous", "_hash"]
        ]
      ]
    }
  }
}

最佳实践与修正方案 #

问题分析 #

当前配置存在两个核心问题：

1. 操作逻辑冲突：post-vendor会在所有分支执行，包括_deleted=true的场景，导致删除时仍尝试创建实体，引发无效操作。
2. DELETE请求冗余payload：REST DELETE接口通常无需携带完整payload，当前配置强制添加payload可能导致接口报错或忽略操作。

修正后的Pipe配置 #

{
  "_id": "vendor-rest",
  "type": "pipe",
  "source": {
    "type": "dataset",
    "dataset": "global-vendor"
  },
  "transform": {
    "type": "dtl",
    "rules": {
      "default": [
        ["if",
          ["eq", "_S._deleted", true],
          // 删除场景仅标记删除操作，无需payload
          ["add", "operation", "delete-vendor"],
          // 非删除场景绑定操作与payload生成
          ["do",
            ["add", "operation", "post-vendor"],
            ["add", "payload", ["apply", "payload", "_S."]]
          ]
        ]
      ],
      "payload": [
        // 仅复制业务字段，排除系统元数据
        ["copy", "*", ["not", ["in", "_key", ["_id", "_deleted", "_ts", "_updated", "_previous", "_hash"]]]]
      ]
    }
  }
}

关键优化点 #

• 分支逻辑隔离：用do语句将非删除场景的操作和payload生成绑定，确保删除动作唯一触发。
• 精简DELETE请求：删除场景下不生成payload，符合REST接口设计规范。
• payload过滤：剔除系统元数据，仅传递业务字段给POST接口，避免冗余数据干扰。

额外建议 #

• 确认REST Sink接口支持DELETE /rest/vendor/{_id}路由，且能正确解析路径中的_id参数。
• 若无需批量处理，可在Pipe中添加"batch": false，保障单个实体操作的原子性。

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