如何检测Azure Table Storage新增实体并同步至分析系统

一、接收新实体上传的通知 #

Azure Table Storage本身不支持直接作为Azure Functions触发器，但可以通过Change Feed + Event Grid实现实时通知：

1. 开启Table的Change Feed功能
   • 在Azure门户进入目标存储账户，找到对应Table，在「Change Feed」选项卡中开启；
   • 或用Azure CLI执行：

     az storage table update --name <你的表名> --account-name <存储账户名> --enable-change-feed true
     

2. 通过Event Grid订阅Change Feed事件
   • 在Azure门户创建Event Grid订阅，目标选择Azure Function/Logic Apps等服务；
   • 当Table中有新实体创建（或更新/删除）时，Change Feed会捕获操作，Event Grid将事件推送到你的目标服务，在服务中编写逻辑同步数据到分析系统。

二、按时间戳过滤查询新实体 #

NodeJS Azure SDK完全支持通过OData过滤条件查询指定时间戳之后的实体，具体实现如下：

1. 使用@azure/data-tables SDK（推荐最新版本），初始化TableClient：

   const { TableClient, AzureNamedKeyCredential } = require("@azure/data-tables");
   
   const credential = new AzureNamedKeyCredential("<存储账户名>", "<存储账户密钥>");
   const tableClient = new TableClient("<存储账户URL>", "<你的表名>", credential);
   

2. 构造时间戳过滤条件，调用listEntities获取新实体：

   // 从分析系统获取上次同步的最新时间戳
   const lastSyncTimestamp = new Date("2024-05-20T12:00:00Z");
   // 构造OData过滤条件，注意时间格式必须是ISO 8601并包裹在datetime''中
   const filter = `Timestamp gt datetime'${lastSyncTimestamp.toISOString()}'`;
   
   async function syncNewEntities() {
     const entities = [];
     // 遍历所有分页结果
     for await (const entity of tableClient.listEntities({ filter })) {
       entities.push(entity);
     }
     // 将entities同步到分析系统
     console.log(`同步了${entities.length}个新实体`);
   }
   
   syncNewEntities().catch(console.error);
   

3. 定期执行该查询：可以通过Azure Functions的定时器触发器（比如每分钟/每5分钟执行一次），实现增量同步。

注意：Timestamp是Table Storage的系统属性，无需额外定义，查询时直接使用即可；如果需要更精确的同步，可结合Partition Key和Timestamp一起过滤，提升查询效率。

内容的提问来源于stack exchange，提问作者Marc O'Morain