SZ-SC采购订单审核流-钉钉工作通知
在系统集成的实际应用中，构建高效、稳定的数据流至关重要。本文将聚焦于一个具体案例——如何将MySQL中的数据集成到钉钉，实现SZ-SC采购订单审核流程，并通过轻易云数据集成平台来搭建并优化这一流程。
首先，我们需要从MySQL数据库中定时获取最新的采购订单数据。这一过程利用了MySQL提供的API接口SELECT语句进行数据查询操作。在实现这一功能时，需要特别关注分页和限流问题，以确保不会因大规模查询造成系统负荷过重或超时错误。
其次，为了满足业务需求，将获取到的采购订单信息写入到钉钉，实现实时工单通知。这个环节通过调用钉钉提供的API接口 topapi/message/corpconversation/asyncsend_v2 来完成大量数据快速写入操作。在此过程中，我们设计了一套可靠的数据转换逻辑，处理了MySQL与钉钉之间可能存在的数据格式差异。此外，还实现对接异常处理与错误重试机制，保证消息推送的稳定性。
整个任务链条由集中监控及告警系统加以保障，通过实时跟踪每个步骤的状态和性能指标，及时发现和处理潜在问题。同时，自定义数据转换逻辑使得我们可以灵活应对多变的业务需求，而可视化的数据流设计工具进一步简化了管理难度，使得配置更加直观、友好。
以上总结为以下几个关键技术点：
- 定时抓取MySQL接口数据： 利用
SELECT语句按时间间隔读取最新记录。 - 批量集成至钉钉： 调用
topapi/message/corpconversation/asyncsend_v2进行消息批量发送。 - 分页与限流控制： 确保在高吞吐量场景下，不漏单且保持响应效率。
- 异常处理及重试机制： 针对网络波动或服务不稳定情况设置自动重试策略。
- 实时监控与日志记录： 跟踪每一步骤执行情况，提高透明度和可维护性。
上述方法呈现出如何运用先进的平台特性来构筑一个健壮、高效的数据集成方案，从而大幅提升企业内部协同效率。接下来，将详细阐述各个步骤实施细节及代码实例。
调用MySQL接口获取并加工数据
在轻易云数据集成平台中，调用源系统MySQL接口select获取并加工数据是数据处理生命周期的第一步。本文将详细探讨如何通过配置元数据来实现这一过程。
元数据配置解析
元数据配置是整个数据集成过程的核心，它定义了如何从源系统中提取和加工数据。以下是我们将要使用的元数据配置：
{
"api": "select",
"effect": "QUERY",
"method": "POST",
"number": "user_id",
"id": "user_id",
"request": [
{
"field": "main_params",
"label": "主参数",
"type": "object",
"describe": "对应其它请求字段内SQL语句的主参数，必须一一对应。",
"value": "1",
"children": [
{
"field": "limit",
"label": "限制结果集返回的行数",
"type": "int",
"describe": "必要的参数！LIMIT 子句用于限制查询结果返回的行数。它指定了查询应该返回的最大行数。例如，LIMIT 10 表示查询结果只包含最多 10 行数据。这对于分页查询非常有用，可以在每次查询中返回一定数量的结果。",
"value": "{PAGINATION_PAGE_SIZE}"
},
{
"field": "offset",
"label": "偏移量",
"type": "int",
"describe": "OFFSET 子句用于指定查询结果的起始位置或偏移量。它表示查询应该从结果集的哪一行开始返回数据。例如，OFFSET 20 表示查询应该从结果集的第 21 行开始返回数据。结合 LIMIT 子句使用时，OFFSET 指定了查询结果的起始行数。",
"value": "{PAGINATION_START_ROW}"
}
]
}
],
...
}主SQL语句解析
主SQL语句是整个数据提取和加工过程中的关键部分。以下是元数据中的主SQL语句：
select user1.user_id,
user1.user_count,
user2.job_number,
case user3.userid
when '0119001462201065096' then '0119001462201065096,2635645303840252'
else user3.userid
end as userid,
user2.real_name,
now() as id
from (
SELECT user_id, COUNT(*) AS user_count
FROM (
SELECT SUBSTRING_INDEX(SUBSTRING_INDEX(afp.audit_user, ',', numbers.n), ',', -1) AS user_id
FROM mbs_pur_record a
LEFT JOIN basic_supplier_info b ON b.supplier_uuid = a.supplier_uuid
LEFT JOIN sys_user c ON c.user_id = a.create_by
LEFT JOIN act_to_form_process afp ON afp.form_uuid = a.pur_audit_uuid
AND afp.company_code = a.company_code
JOIN (
SELECT a.N + b.N * 10 + 1 AS n
FROM (
SELECT 0 AS N
UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5
UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9
) a
CROSS JOIN (
SELECT 0 AS N
UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5
UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9
) b
) numbers ON numbers.n <= LENGTH(afp.audit_user) - LENGTH(REPLACE(afp.audit_user, ',', '')) + 1
WHERE a.pur_status IN ('1', '11')
AND a.company_code = 'TYZN'
AND a.pur_type = '2'
) AS users
GROUP BY user_id
ORDER BY user_id) user1
left join sys_user user2 on user1.user_id=user2.user_id
left join basic_dingding_userid user3 on user2.job_number=user3.workid
limit :limit offset :offset;参数绑定与优化
为了确保主SQL语句中的动态字段（如:limit和:offset）与请求参数正确对应，我们需要进行参数绑定。这不仅提高了查询语句的可读性和维护性，还能确保其准确性和安全性。
具体步骤如下：
-
替换动态字段：将主SQL语句中的动态字段
:limit和:offset替换为占位符（例如?）。limit ? offset ? -
绑定请求参数：在执行查询之前，将请求参数值与占位符进行绑定。
{ ... { field: 'main_params', value: { limit: PAGINATION_PAGE_SIZE, offset: PAGINATION_START_ROW, } } ... }
通过这种方式，我们能够确保动态字段与请求参数的一一对应，从而保证查询操作的准确性。
数据请求与清洗
在完成上述配置后，我们可以通过POST方法发送请求以获取并清洗数据。以下是一个示例请求：
{
main_params: {
limit: PAGINATION_PAGE_SIZE,
offset: PAGINATION_START_ROW,
},
main_sql: `select ... limit ? offset ?`
}该请求将根据配置好的元数据，从MySQL数据库中提取符合条件的数据，并进行初步清洗和加工。
小结
通过以上步骤，我们成功地调用了MySQL接口select获取并加工了所需的数据。这一步骤不仅包括对主SQL语句及其动态字段进行优化，还涉及到对请求参数进行精确绑定，以确保数据提取过程的准确性和高效性。在轻易云平台中，这些操作都可以通过全透明可视化界面进行，大大提升了业务透明度和效率。
轻易云数据集成平台：ETL转换与钉钉API接口集成案例
在轻易云数据集成平台的生命周期中，数据转换与写入是关键的一步。本文将深入探讨如何将已经集成的源平台数据进行ETL转换，转为目标平台钉钉API接口所能够接收的格式，并最终写入目标平台。
钉钉API接口配置
在本案例中，我们需要将采购订单审核流的数据通过钉钉工作通知发送给相关用户。为了实现这一目标，我们使用了钉钉的topapi/message/corpconversation/asyncsend_v2 API接口。以下是该接口的元数据配置：
{
"api": "topapi/message/corpconversation/asyncsend_v2",
"effect": "EXECUTE",
"method": "POST",
"idCheck": true,
"request": [
{
"field": "userid_list",
"label": "userid_list",
"type": "string",
"describe": "111",
"value": "{userid}"
},
{
"field": "to_all_user",
"label": "to_all_user",
"type": "string",
"describe": "111",
"value": "false"
},
{
"field": "msg",
"label": "msg",
"type": "object",
"describe": "111",
"value": "test",
"children": [
{
"field": "msgtype",
"label": "msgtype",
"type": "string",
"value": "markdown"
},
{
"field": "markdown",
"label": "markdown",
"type": "object",
...
}
]
},
{
...
}
]
}数据转换过程
-
提取用户ID列表：从源系统中提取需要接收通知的用户ID列表，并将其映射到
userid_list字段。{ ... { “field”: “userid_list”, “label”: “userid_list”, “type”: “string”, “describe”: “111”, “value”: “{userid}” }, ... } -
设置全员通知标志：根据需求设置是否发送给所有用户。在本案例中，设置为
false。{ ... { “field”: “to_all_user”, “label”: “to_all_user”, “type”: “string”, “describe”: “111”, “value”: “false” }, ... } -
构建消息内容：消息内容采用Markdown格式，需要嵌入动态数据，如待审批的采购订单数量和具体订单ID。使用轻易云提供的函数进行字符串拼接。
{ ... { “field”: “msg”, “label”： ”msg“， ”type“: ”object“， ”describe“: ”111“， ”value“: ”test“， ”children“: [ { ”field“: ”msgtype“， ”label“: ”msgtype“， ”type“: ”string“， ”value“: ”markdown“ }, { ”field“: ”markdown“， ”label“: ”markdown“， ”type“: ”object“， ... } ] }, ... } -
设置代理ID：指定发送消息的应用代理ID。
{ ... { “field”：”agent_id”， ”label”：”agent_id”， ”type”：”string”， ”describe”：”111”， ”value”：”2811489571” } ... }
数据写入过程
完成数据转换后，将其通过POST请求发送到钉钉API接口。以下是请求示例：
POST /topapi/message/corpconversation/asyncsend_v2 HTTP/1.1
Host: oapi.dingtalk.com
Content-Type: application/json
{
„userid_list„：“user123,user456„，
„to_all_user„：“false„，
„msg„：{
„msgtype„：“markdown„，
„markdown„：{
„title„：“采购审核流„，
„text„：“# 四化MES系统 \n ## 采购审核流提醒 \n ### 你有3张采购订单等待审批 \n PO12345 \n ### [<查看详情>](http://192.168.110.232:9000/examine/outsourcAudit)”
}
}，
„agent_id„：“2811489571”
}通过上述步骤，我们成功地将源平台的数据进行ETL转换，并通过钉钉API接口发送工作通知，实现了不同系统间的数据无缝对接。这一过程不仅提高了业务流程的透明度和效率，也确保了信息传递的准确性和及时性。
