在PowerBI中使用跨事务任务流实现自定义备注

前言

关于数据回写的一个应用，即让用户可以在PowerBI报表中自定义备注信息，这个功能在之前的文章中已经分享过实现方案了，但之前的实现方案需要借助第三方的视觉对象，实现起来较复杂，且自由度和用户体验上还是差了一点。

而在2025年05月的更新中，PowerBI新增了跨事务任务流（Translytical Tasks Flow）功能，可以使用按钮来调用Fabric的用户数据函数，从而可以实现数据回写等功能。

因此，本篇文章就基于此背景，使用官方原生的组件来重新实现之前介绍过的自定义备注功能，同时也是为了介绍新出的跨事务任务流的使用方法。

---

下面先来看一下最终效果，如下图所示：

实现该效果的简要步骤如下：

1、根据需求，设计存储回传数据的表的表结构，并在SQL Server中创建该表，为方便描述，将其称为回传数据表

2、创建Fabric用户数据函数，将接收到的自定义备注信息写入数据库

3、在PowerBI中使用按钮来调用Fabric用户数据函数并传递自定义备注信息

4、在PowerBI中使用DirectQuery模式连接到SQL Server中的回传数据表，实时显示最新的备注信息

注意，由于要实时显示最新的备注信息，所以需要使用DirectQuery模式来获取数据，因此存放回传数据表的数据源要支持DirectQuery模式才行。这里为了演示Fabric用户数据函数的更多可能性，所以使用了自己部署的SQL Server数据库。

如果没有数据库环境，则可以直接使用Fabric上的SQL Database服务，但要在租户设置中打开相应选项，如下图所示：

另外，PowerBI中的数据源或表并不是全部都要用到DirectQuery模式，除了关键的回传数据表外，其他的数据源或表都可以使用Import模式的。

最后，关于Fabric用户数据函数的详细介绍和使用可以参考我的另一篇文章：Fabric用户数据函数的介绍

回传数据表的表结构设计

回传数据表需要存储在矩阵或表格中添加的自定义备注信息，因此可以用矩阵的行标签即客户城市和产品类别为组合主键，然后再添加一个备注信息字段即可，但考虑到备注信息的追诉和跟踪，所以再添加更新时间以及更新用户这两个字段会更稳妥。

综合考虑后，回传数据表的表结构确定为如下：

然后在SQL Server数据库中按以上表结构创建表即可，使用到的SQL语句如下：

IF EXISTS (SELECT * FROM sys.all_objects WHERE object_id = OBJECT_ID(N'[dbo].[RemarkMessage]') AND type IN ('U'))
    DROP TABLE [dbo].[RemarkMessage]
GO

CREATE TABLE [dbo].[RemarkMessage] (
  [City] nvarchar(255) COLLATE Chinese_PRC_CI_AS  NULL,
  [ProductCategory] nvarchar(255) COLLATE Chinese_PRC_CI_AS  NULL,
  [Remark] nvarchar(255) COLLATE Chinese_PRC_CI_AS  NULL,
  [UpdateUser] nvarchar(255) COLLATE Chinese_PRC_CI_AS  NULL,
  [UpdatedTime] datetime  NULL
)

回传数据表的表结构决定了后续数据回写时的方式，也会影响到报表中如何显示最新回写的数据的逻辑，因此需要针对具体的应用场景进行分析与设计。

创建Fabric用户数据函数

首先，在PowerBI Server的工作区中创建一个用户数据函数，如下图所示：

然后，将以下代码粘贴到用户数据函数中：

import fabric.functions as fn
import logging
import pymssql

udf = fn.UserDataFunctions()

@udf.function()
def remarkMessage(city:str,productCategory:str,reamrkContent:str,updateUser:str,updateTime:str)->str:
    logging.info('开始执行')
    try:
        connect = pymssql.connect(host='xxxxx',port=1433,database='xxxxx',user='xxxxx',password='xxxxx')
        cursor = connect.cursor()
        sql = f'insert into RemarkMessage values (%s,%s,%s,%s,%s)'
        cursor.execute(sql,(city,productCategory,reamrkContent,updateUser,updateTime))
        connect.commit()
        logging.info('执行成功！')
        return "备注添加成功！"
    except Exception as e:
        logging.warning(f'出现意外错误,信息：{e}')
        raise fn.UserThrownError(f'出现意外错误,信息：{e}')
    finally: 
        cursor.close()
        connect.close()

如下图所示：

由于用到了第三方库pymssql来连接SQL Server，因此还需要在上面菜单栏的库管理中添加pymssql库，如下图所示：

最后，点击上方菜单栏的发布按钮进行发布，如下图所示：

到这里为止，用户数据函数就算是创建完了。另外说一下，由于是出于演示的目的，所以连接数据库时使用的凭证就简单的以明文形式在代码中存储了，这在生产环境中是非常不建议的。

如果是在生产环境中，可以将数据库的账号密码等凭证作为函数的参数，然后在PowerBI调用时再使用度量值去传参。甚至更进一步的，在度量值中还可以再从另一个数据源里获取凭证，或者事先对凭证进行编码，然后Fabric用户数据函数那边添加一个解码的模块等等。

在PowerBI中调用Fabric用户数据函数

在PowerBI报表里，可以使用跨事务任务流来调用Fabric用户数据函数，这个跨事务任务流的名字听起来好像很复杂，但它其实就是在平时经常用到的按钮里添加了一个Fabric用户数据函数的选项而已。

如上图所示的，在按钮的操作设置中可以指定类型为用户数据函数，然后配置工作区、具体要调用的函数、以及函数的参数等即可。

其中，函数的参数可以使用度量值或绑定到按钮切片器、列表切片器、以及文本切片器，除度量值外目前只支持这三个后面新出的切片器视觉对象，不支持绑定到旧的切片器视觉对象。将参数绑定到切片器时，只需要在下拉框里选择切片器的标题即可。

下面是用到的度量值的代码：

Params_City = SELECTEDVALUE('客户城市'[客户城市])

Params_Category = SELECTEDVALUE('产品表'[产品类别])

Params_User = USERNAME()

Params_UpdateTime = CONVERT(UTCNOW()+TIME(8,0,0),STRING)

将上面的度量值绑定到用户数据函数对应的参数即可，然后剩下的备注信息参数则是绑定到文本切片器。

在PowerBI中实时显示备注信息

首先，使用DirectQuery模式连接到SQL Server数据库上的回传数据表，并在模型关系中保持回传数据表的独立，即不与其他表建立关系。如下图所示：

然后创建一个备注度量值，并添加到表格或矩阵中，以从回传数据表中获取最新的备注信息，如下图所示：

备注度量值的表达式如下：

备注 = 
IF(ISFILTERED('产品表'[产品类别])&&ISFILTERED('客户城市'[客户城市]),
    CALCULATE(
        MAXX(
            TOPN(1,'RemarkMessage','RemarkMessage'[UpdatedTime]),
            'RemarkMessage'[Remark]
        ),
        TREATAS(VALUES('产品表'[产品类别]),'RemarkMessage'[ProductCategory]),
        TREATAS(VALUES('客户城市'[客户城市]),'RemarkMessage'[City])
    )
)

至此，自定义备注功能所需的操作就全部完成了。

总结

PowerBI有了跨事务任务流后，就可以借助Fabric用户数据函数来集成其他的软件生态服务，从而可以实现更多的可能。本篇文章所介绍的自定义备注功能仅仅只是冰山一角，更多的应用场景还有待发掘。