### 案例分享：金蝶云星空数据集成到MySQL
在本案例中，我们将探讨如何通过轻易云数据集成平台，将金蝶云星空系统中的调拨申请单信息（MOM-T02接收的调拨申请单-表头-新），高效、安全地集成到MySQL数据库中。重点包括接口调用、数据转换以及异常处理等关键技术环节。
首先，通过调用金蝶云星空API `executeBillQuery` 获取调拨申请单的相关数据。这一过程涉及分页查询和限流控制，以确保大规模数据获取过程中服务稳定性及效率。我们设计了定时任务来可靠地抓取所需的数据，保证系统能够按预设周期自动执行，无需人工干预。
其次，针对从金蝶云获取的数据，我们需要进行自定义转换逻辑，使其符合MySQL数据库表结构要求。在这一步骤中，可视化的数据流设计工具发挥了重要作用，使得整个流程更加直观且便于管理。
在完成初步的数据处理后，通过调用MySQL写入API `execute` 实现批量数据写入。由于高吞吐量需求，我们采取了一系列优化措施，包括分片插入与并行处理，来提升写入速度。此外，还设置了详细的监控和告警机制，以实时跟踪任务状态，并在出现异常时及时响应。
最后，为确保整个集成过程不漏单、不丢失任何关键信息，实现精确的数据同步和一致性，我们特别关注异常检测和错误重试机制。在每一次操作失败后，均会触发自动重试功能，同时记录日志以便问题定位与分析。这些措施不仅提高了系统的健壮性，也为运维提供了便利支持。
这种从获取到存储全流程无缝对接解决方案，不仅提升了企业资源利用率，还极大降低了手动操作带来的风险与工作负担，更好地支撑企业核心业务的发展需求。
### 调用源系统金蝶云星空接口executeBillQuery获取并加工数据
在数据集成生命周期的第一步，我们需要从源系统金蝶云星空中调用`executeBillQuery`接口来获取调拨申请单的表头数据，并对其进行初步加工。这一步骤至关重要，因为它为后续的数据清洗、转换和写入奠定了基础。
#### 接口调用配置
我们使用POST方法调用金蝶云星空的`executeBillQuery`接口，以下是具体的元数据配置：
```json
{
"api": "executeBillQuery",
"effect": "QUERY",
"method": "POST",
"number": "FBillNo",
"id": "FID",
"name": "FNumber",
"idCheck": true,
"request": [
{"field":"FID","label":"表头实体主键","type":"string","value":"FID"},
{"field":"FBillNo","label":"单据编号","type":"string","value":"FBillNo"},
{"field":"FDate","label":"日期","type":"string","value":"FDate"},
{"field":"FDocumentStatus","label":"单据状态","type":"string","value":"FDocumentStatus"},
{"field":"FAPPORGID","label":"申请组织","type":"string","value":"FAPPORGID.FNumber"},
{"field":"FCloseStatus","label":"关闭状态","type":"string","value":"FCloseStatus"},
{"field":"FCANCELSTATUS","label":"取消状态","type":"string","value":"FCANCELSTATUS"}
],
"otherRequest": [
{"field": "Limit", "label": "最大行数", "type": "string", "describe": "金蝶的查询分页参数", "value": "1000"},
{"field": "StartRow", "label": "开始行索引", "type": "string", "describe": "金蝶的查询分页参数", "value": "{PAGINATION_START_ROW}"},
{"field": "TopRowCount", "label": "返回总行数", "type": "int", "describe": "金蝶的查询分页参数"},
{"field": "FilterString",
"label":
"过滤条件",
"type":
"string",
"describe":
"示例写法 FSupplierId.FNumber = 'VEN00010' and FApproveDate>=",
"value":
"FDocumentStatus='C' and FOwnerId.FNumber='T02' and fqty>FAllTransOutBaseQty and FAPPROVEDATE>='{{LAST_SYNC_TIME|datetime}}'"
},
{"field":
"FieldKeys",
"label":
"需查询的字段key集合",
"type":
"array",
"describe":
"金蝶分录主键ID格式：FPOOrderEntry_FEntryId,其它格式 FPurchaseOrgId.FNumber",
"parser":{
"name":
"ArrayToString",
"params":
","
}
},
{"field":
"FormId",
"label":
"业务对象表单Id",
"type":
"string",
"describe":
"必须填写金蝶的表单ID如:PUR_PurchaseOrder",
"value":
"STK_TRANSFERAPPLY"
}
],
“autoFillResponse”: true
}
```
#### 请求参数详解
1. **基本请求字段**：
- `FID`: 表头实体主键，用于唯一标识每条记录。
- `FBillNo`: 单据编号，标识具体的调拨申请单。
- `FDate`: 日期，记录单据生成或修改的时间。
- `FDocumentStatus`: 单据状态，用于筛选特定状态下的单据。
- `FAPPORGID.FNumber`: 申请组织，标识发起调拨申请的组织。
- `FCloseStatus`: 关闭状态，标识单据是否已关闭。
- `FCANCELSTATUS`: 取消状态，标识单据是否已取消。
2. **其他请求字段**：
- `Limit`: 最大行数，限制每次查询返回的数据量，这里设置为1000。
- `StartRow`: 开始行索引，用于分页查询。
- `TopRowCount`: 返回总行数，用于统计总记录数。
- `FilterString`: 过滤条件，这里设置为筛选已审核（`C`）且未完全出库（`fqty > FAllTransOutBaseQty`）且审批日期在上次同步时间之后的记录。
- `FieldKeys`: 查询字段集合，将需要返回的字段以逗号分隔字符串形式传递。
- `FormId`: 表单ID，这里指定为调拨申请单（`STK_TRANSFERAPPLY`）。
#### 数据获取与初步加工
通过上述配置，我们可以向金蝶云星空发送请求并获取调拨申请单表头数据。接下来，我们需要对这些数据进行初步加工，以便后续处理。
1. **数据清洗**：
确保每条记录都有完整的数据，例如检查必填字段是否为空，对日期格式进行标准化等。
2. **数据转换**：
将原始数据转换为目标系统所需格式，例如将日期格式从“YYYY-MM-DD”转换为“DD/MM/YYYY”，或者将组织编码转换为目标系统中的对应编码。
3. **数据验证**：
对获取的数据进行验证，确保其符合业务逻辑和规则。例如，检查单据状态是否正确，确保数量字段大于零等。
通过上述步骤，我们可以高效地从金蝶云星空中获取并初步加工调拨申请单表头数据，为后续的数据处理打下坚实基础。
### 使用轻易云数据集成平台进行ETL转换并写入MySQLAPI接口
在数据集成生命周期的第二步中，我们需要将已经集成的源平台数据进行ETL（Extract, Transform, Load）转换，并最终写入目标平台。本文将详细探讨如何利用轻易云数据集成平台，将调拨申请单的表头数据转换为MySQLAPI接口所能接收的格式，并写入目标数据库。
#### 元数据配置解析
我们首先来看一下元数据配置，它定义了如何将源数据映射到目标API接口：
```json
{
"api": "execute",
"effect": "EXECUTE",
"method": "POST",
"idCheck": true,
"request": [
{
"field": "main_params",
"label": "main_params",
"type": "object",
"describe": "111",
"children": [
{"field": "ID", "label": "实体主键", "type": "string", "value":"{FID}"},
{"field": "BILLNO", "label": "单据编号", "type": "string", "value":"{FBillNo}"},
{"field": "DOCUMENTSTATUS", "label": "单据状态", "type": "string", "value":"{FDocumentStatus}"},
{"field": "REQUESTDATE", "label": 申请日期, type: string, value: {FDate}},
{"field":"APPORG","label":"申请组织","type":"string","value":"{FAPPORGID}"},
{"field":"CANCELSTATUS","label":"取消状态","type":"string","value":"{FCANCELSTATUS}"},
{"field":"CLOSESTATUS","label":"关闭状态","type":"string","value":"{FCloseStatus}"}
]
}
],
otherRequest: [
{
field: main_sql,
label: main_sql,
type: string,
describe: 111,
value: INSERT INTO ty_aps.wms_transfer_apply (id,billno,documentstatus,requestdate,apporg,CANCELSTATUS,CLOSESTATUS) VALUES (:ID,:BILLNO,:DOCUMENTSTATUS,:REQUESTDATE,:APPORG,:CANCELSTATUS,:CLOSESTATUS)
}
]
}
```
#### 数据请求与清洗
在这一阶段，我们需要从源系统中提取调拨申请单的表头数据。这些数据通常包括实体主键、单据编号、单据状态、申请日期、申请组织、取消状态和关闭状态等字段。提取的数据将通过轻易云的数据请求功能进行清洗，以确保其符合目标API接口的要求。
#### 数据转换
接下来，我们需要将清洗后的数据进行转换，以匹配MySQLAPI接口所需的格式。根据元数据配置，字段映射如下：
- `ID` 映射到 `{FID}`
- `BILLNO` 映射到 `{FBillNo}`
- `DOCUMENTSTATUS` 映射到 `{FDocumentStatus}`
- `REQUESTDATE` 映射到 `{FDate}`
- `APPORG` 映射到 `{FAPPORGID}`
- `CANCELSTATUS` 映射到 `{FCANCELSTATUS}`
- `CLOSESTATUS` 映射到 `{FCloseStatus}`
这些字段将被封装在一个名为 `main_params` 的对象中，并通过POST方法发送给目标API。
#### 数据写入
最后一步是将转换后的数据写入目标数据库。根据元数据配置中的 `otherRequest` 部分，生成的SQL语句如下：
```sql
INSERT INTO ty_aps.wms_transfer_apply (id,billno,documentstatus,requestdate,apporg,CANCELSTATUS,CLOSESTATUS)
VALUES (:ID,:BILLNO,:DOCUMENTSTATUS,:REQUESTDATE,:APPORG,:CANCELSTATUS,:CLOSESTATUS)
```
该SQL语句会被执行，插入相应的数据到MySQL数据库中的 `ty_aps.wms_transfer_apply` 表。
#### API接口调用示例
以下是一个示例代码片段，展示如何调用上述配置的API接口：
```python
import requests
import json
url = 'http://your-mysql-api-endpoint/execute'
headers = {'Content-Type': 'application/json'}
data = {
'main_params': {
'ID': '12345',
'BILLNO': 'ABCD1234',
'DOCUMENTSTATUS': 'Approved',
'REQUESTDATE': '2023-10-01',
'APPORG': 'Org001',
'CANCELSTATUS': 'None',
'CLOSESTATUS': 'Open'
},
'main_sql': """
INSERT INTO ty_aps.wms_transfer_apply
(id,billno,documentstatus,requestdate,apporg,CANCELSTATUS,CLOSESTATUS)
VALUES (:ID,:BILLNO,:DOCUMENTSTATUS,:REQUESTDATE,:APPORG,:CANCELSTATUS,:CLOSESTATUS)
"""
}
response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
print('Data successfully written to MySQL.')
else:
print('Failed to write data:', response.text)
```
这个示例展示了如何通过HTTP POST请求，将封装好的JSON对象发送给MySQLAPI接口，从而实现数据的插入操作。
通过以上步骤，我们成功地完成了从源平台提取调拨申请单表头数据，进行ETL转换，并最终写入目标MySQL数据库的全过程。这一过程充分利用了轻易云数据集成平台的强大功能，实现了不同系统间的数据无缝对接。
