lblod/gelinkt-notuleren-consumer

gelinkt-notuleren-consumer

镜像概述和主要用途

gelinkt-notuleren-consumer 是一个可配置的消费者服务，用于基于生产者生成的差异文件从外部源同步数据。它定期检查新的差异文件并摄取其中的数据，根据授权规则将数据存储到适当的图中。生产者示例可参考 http://github.com/lblod/mandatendatabank-mandatarissen-producer%E3%80%82

核心功能和特性

• 可配置的数据类型：通过配置文件定义需要摄取的数据类型及关联路径
• 定期同步：支持通过环境变量设置自动同步间隔
• 手动触发：提供 API 端点手动触发同步任务
• 数据权限管理：根据授权规则将数据存储到适当的图中
• 任务状态跟踪：记录同步任务的状态（未开始、进行中、成功、失败）及时间戳

使用场景和适用范围

适用于需要从外部数据源定期同步数据的场景，尤其适合基于差异文件进行增量数据同步的项目，如***数据同步、组织信息管理系统等需要确保数据一致性的应用。

使用方法和配置说明

将服务添加到栈中

1. 在 docker-compose.yml 中添加服务：

   yaml
   consumer:
     image: lblod/gelinkt-notuleren-consumer
     environment:
       SERVICE_NAME: 'your-custom-consumer-identifier'  # 替换为所需的消费者标识符
       SYNC_BASE_URL: 'http://base-sync-url'  # 替换为生产者服务器的基础 URL
     volumes:
       - ./config/consumer/:/config/  # 替换为类型配置文件的路径
   

2. 配置 SERVICE_NAME：

   • 必须设置为唯一的消费者标识符，用于确保数据持久性。

3. 配置 SYNC_BASE_URL：

   • 设置为托管生产者服务器的应用 URL。

4. 创建类型配置文件： 在 config/consumer/ 目录下创建 types.js 文件，定义需要摄取的数据类型数组：

   json
   [
     {
       "type": "http://www.w3.org/ns/person#Person"
     },
     {
       "type": "http://www.w3.org/ns/person#Person",
       "pathToOrg": [
          "http://www.w3.org/ns/org#holds",
          "^http://www.w3.org/ns/org#hasPost"
       ]
     }
   ]
   

   每个类型对象包含：

   • "type"：需摄取类型的完整 URI（必填）
   • "pathToOrg"：可选，构造链接组织路径的 URI 数组，用于确定数据存储的组织图

自动化同步任务调度

通过添加 INGEST_INTERVAL 环境变量配置自动同步间隔（毫秒）：

yaml
consumer:
  image: lblod/gelinkt-notuleren-consumer
  environment:
    INGEST_INTERVAL: 60000  # 每分钟同步一次

API 接口

GET /

• 简单的 "hello-world" 端点，返回当前运行版本。

POST /schedule-ingestion

• 调度并执行同步任务。

配置参数

环境变量

必填项

• SERVICE_NAME：消费者标识符，用于确保持久性，在项目中需唯一。
• SYNC_BASE_URL：托管生产者 API 的栈的基础 URL（例如：[***]

可选项

• SYNC_FILES_PATH：获取差异文件名的端点相对路径，默认：/sync/files。
• DOWNLOAD_FILES_PATH：下载差异文件的端点相对路径，:id 会替换为文件 UUID，默认：/files/:id/download。
• INGEST_INTERVAL：自动同步间隔（毫秒），默认：-1（需通过 API 手动触发）。
• START_FROM_DELTA_TIMESTAMP：开始同步的时间戳（ISO 格式），默认：当前时间。
• PUBLIC_GRAPH：存储公共数据和同步任务的公共图，默认：http://mu.semte.ch/graphs/public。
• TMP_INGEST_GRAPH：临时存储插入变更集的图，默认：http://mu.semte.ch/graphs/tmp-ingest-gelinkt-notuleren-mandatarissen-consumer。

数据模型

使用的前缀

同步任务（Sync task）

• 类：ext:SyncTask
• 属性：

同步任务状态

任务状态会更新以反映进度，包含以下状态：

• http://lblod.data.gift/gelinkt-notuleren-mandatarissen-consumer-sync-task-statuses/not-started（未开始）
• http://lblod.data.gift/gelinkt-notuleren-mandatarissen-consumer-sync-task-statuses/ongoing（进行中）
• http://lblod.data.gift/gelinkt-notuleren-mandatarissen-consumer-sync-task-statuses/success（成功）
• http://lblod.data.gift/gelinkt-notuleren-mandatarissen-consumer-sync-task-statuses/failure（失败）

数据流程

服务定期调度同步任务，执行步骤如下：

1. 获取同步起始时间戳
2. 查询生产者服务获取该时间戳后的所有差异文件
3. 下载每个差异文件的内容
4. 按顺序处理每个差异文件

差异文件处理：

• 插入变更集：先摄取到临时图 TMP_INGEST_GRAPH（因确定目标图所需信息可能缺失）
• 删除变更集：在所有图（含临时图）中执行删除查询

每个差异文件处理结束后，执行查询将数据从临时图移动到基于 rdf:type 和授权规则的目标图（核心逻辑位于 ./lib/delta-file/moveTriplesFromTmpGraph()）。若一个文件处理失败，后续文件将被阻塞（需按顺序处理）。

核心假设：

1. 任何时刻，处于失败/进行中/成功状态的任务的最新 ext:deltaUntil 时间戳反映已完全成功消费的最新差异文件时间戳
2. 任何时刻最多运行一个同步任务

本实现是 https://github.com/lblod/gelinkt-notuleren-mandatarissen-consumer 的可配置分支。