如何基于REST API事务数据更新Power BI自定义连接器数据集

需求背景 #

你已经搭建好返回事务型数据的REST API，JSON格式如下（修正了原格式的语法错误）：

[
    {
        "key": {"id": 1, "name": "示例记录"},
        "transaction": "added"
    },
    {
        "key": {"id": 2, "name": "已更新记录"},
        "transaction": "updated"
    }
]

原有自定义连接器仅直接拉取数据生成表，现在需要调整逻辑，实现：

• 读取现有全量数据集
• 空数据集时用新增记录初始化
• 对现有数据执行新增、更新、删除操作
• 返回最终更新后的数据集

原有连接器代码 #

shared GetTableFromApi = (query as text) => 
    let
        json = Json.Document(Web.Contents("some-url/" & query))
    in
        Table.FromRecords(json);

调整后的实现方案（M语言） #

Power Query M不支持传统循环逻辑，需通过表操作实现事务处理，核心是基于**masterEntries（全量数据集）**来应用各类事务：

shared GetTableFromApi = (query as text) => 
    let
        // 1. 从API获取事务数据并转换为表
        transactionsJson = Json.Document(Web.Contents("some-url/" & query)),
        transactionsTable = Table.FromRecords(transactionsJson),
        
        // 2. 获取现有全量数据集，无数据则初始化空表（需匹配你的实际字段结构）
        emptyMasterTable = Table.FromRecords({[id=null, name=null]}),
        masterEntries = try Table.FromRecords(YourExistingMasterData) otherwise emptyMasterTable,
        
        // 3. 拆分不同类型的事务记录
        addedRecords = Table.SelectRows(transactionsTable, each [transaction] = "added")[key],
        updatedRecords = Table.SelectRows(transactionsTable, each [transaction] = "updated")[key],
        deletedRecords = Table.SelectRows(transactionsTable, each [transaction] = "deleted")[key],
        
        // 4. 执行新增操作：将新增记录追加到主表
        afterAdd = Table.Combine({masterEntries, Table.FromRecords(addedRecords)}),
        
        // 5. 执行更新操作：通过主键（示例用id）匹配并替换旧记录
        mergedForUpdate = Table.NestedJoin(afterAdd, {"id"}, Table.FromRecords(updatedRecords), {"id"}, "Updated", JoinKind.LeftOuter),
        afterUpdate = Table.ReplaceValue(mergedForUpdate, each [Updated], each if [Updated] <> null then [Updated]{0} else _, Replacer.ReplaceValue, {"Updated"}),
        afterUpdateClean = Table.RemoveColumns(afterUpdate, {"Updated"}),
        
        // 6. 执行删除操作：过滤掉标记为删除的记录
        deletedIds = Table.FromRecords(deletedRecords)[id],
        afterDelete = Table.SelectRows(afterUpdateClean, each not List.Contains(deletedIds, [id])),
        
        // 7. 去重处理（可选，根据API逻辑调整，防止重复新增）
        finalTable = Table.Distinct(afterDelete, {"id"})
    in
        finalTable;

关键逻辑说明 #

1. masterEntries的作用：它是存储当前全量数据的核心表，每次更新都基于这个表来应用事务；首次运行时会用定义的空表初始化。
2. 事务拆分：通过Table.SelectRows将新增、更新、删除三类事务分开处理，避免混合操作导致的逻辑混乱。
3. 更新逻辑：通过主键（示例用id）关联主表与更新记录，用新记录覆盖旧记录，确保数据为最新版本。
4. 删除逻辑：提取待删除记录的主键，从主表中过滤掉对应记录。

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