### 金蝶云星空数据集成至轻易云集成平台技术案例分享：查询金蝶物料单位
在企业多系统协同作业的过程中，如何高效、准确地进行数据对接和整合是一个关键问题。在本篇技术案例中，我们将探讨如何实现金蝶云星空中的物料单位数据与轻易云集成平台的对接。
此次项目的核心任务是通过调用金蝶云星空API `executeBillQuery` 接口，获取所需的物料单位信息，并定时可靠地抓取并写入到轻易云集成平台。为确保整个过程的数据不丢失且具有高效性，我们充分利用了轻易云提供的大量数据快速写入能力以及其强大的集中监控系统。
首先，针对接口封装及调用部分，通过分析金蝶云星空 `executeBillQuery` API 的请求结构和响应格式，我们制定了一套标准化的数据映射规则。此规则允许我们自定义转换逻辑，以适应特定业务需求，同时处理跨系统之间的数据格式差异。这一步骤不仅能够保证数据准确传输，还有效提升了处理效率。
另外，为预防在大批量数据操作中的异常情况，我们特别设计了一套错误重试机制。一旦检测到网络延迟或API限流等问题，会自动重新发送请求。同时配备实时告警功能，当出现连续失败的重试行为时，它会即时通知相关人员，确保问题能被及时解决。
最后，在完成初步方案架构后，通过轻易云可视化的数据流设计工具，对整体流程进行了模拟运行和优化调整，从而保障最终上线后的平稳性能。通过这种方式，即便面对复杂多变的实际业务环境，也能最大程度上减少出错概率，提高整体稳定性。
接下来，将具体介绍我们的详细实施方案，包括如何配置各类接口参数、执行步骤，以及遇到的一些常见问题及其应对策略。
### 调用金蝶云星空接口executeBillQuery获取并加工数据
在轻易云数据集成平台中，调用源系统接口是数据集成生命周期的第一步。本文将详细探讨如何通过调用金蝶云星空的`executeBillQuery`接口来获取并加工数据。
#### 接口概述
金蝶云星空的`executeBillQuery`接口主要用于查询业务对象的数据。该接口采用POST请求方式，支持分页查询和多种过滤条件。以下是我们需要配置的元数据：
```json
{
"api": "executeBillQuery",
"method": "POST",
"number": "FNumber",
"id": "FMATERIALID",
"pagination": {
"pageSize": 100
},
"request": [
{"field":"FMATERIALID","label":"实体主键","type":"string","value":"FMATERIALID"},
{"field":"FNumber","label":"编码","type":"string","value":"FNumber"},
{"field":"FName","label":"名称","type":"string","value":"FName"},
{"field":"FSpecification","label":"规格型号","type":"string","value":"FSpecification"},
{"field":"FOldNumber","label":"旧物料编码","type":"string","value":"FOldNumber"},
{"field":"FBARCODE","label":"条码","type":"string","value":"FBARCODE"},
{"field":"FDescription","label":"描述","type":"string","value":"FDescription"},
{"field":"FMaterialGroup_FNumber","label":"物料分组","type":"string","value":"FMaterialGroup.FNumber"},
{"field":"FErpClsID","label":"物料属性","type":"string","value":"FErpClsID"},
{"field":"FDocumentStatus","label":"数据状态","type":"string","value":"FDocumentStatus"},
{"field":"FForbidStatus","label":"禁用状态","type":"string","value":"FForbidStatus"},
{"field": "FilterString", "label": "过滤条件", "type": "string", "describe": "示例写法 FSupplierId.FNumber = 'VEN00010' and FApproveDate>=", "value": "FUseOrgId.FNumber='100' and FApproveDate>='{{LAST_SYNC_TIME|datetime}}'"}
],
"otherRequest":[
{"field": "Limit", "label": "最大行数", "type": "string", "describe": "金蝶的查询分页参数", "value": "1000"},
{"field": "StartRow", "label": "开始行索引", "type": "string", "describe": ""},
{"field": "TopRowCount", "label": "", type: int, describe: ""},
{"field": FieldKeys, label: "", type: array, describe: ""}
]
}
```
#### 配置请求参数
在调用`executeBillQuery`接口时，需要配置请求参数以确保能够正确获取所需的数据。以下是关键参数的配置说明：
1. **FormId**：业务对象表单Id，必须填写金蝶的表单ID，如`BD_MATERIAL`。
2. **FieldKeys**：需查询的字段key集合，可以通过解析器将数组转换为字符串格式。
3. **FilterString**：过滤条件，用于筛选特定的数据，例如 `FSupplierId.FNumber = 'VEN00010' and FApproveDate>= '2023-01-01'`。
4. **Limit** 和 **StartRow**：分页参数，用于控制每次查询的数据量和起始位置。
#### 示例代码
以下是一个调用该接口的示例代码：
```python
import requests
import json
url = 'https://api.kingdee.com/executeBillQuery'
headers = {'Content-Type': 'application/json'}
payload = {
'FormId': 'BD_MATERIAL',
'FieldKeys': 'FMATERIALID,FNumber,FName,FSpecification,FOldNumber,FBARCODE,FDescription',
'FilterString': 'FUseOrgId.FNumber="100" and FApproveDate>="2023-01-01"',
'Limit': 100,
'StartRow': 0
}
response = requests.post(url, headers=headers, data=json.dumps(payload))
if response.status_code == 200:
data = response.json()
# 对返回的数据进行处理
else:
print(f"Error: {response.status_code}, {response.text}")
```
#### 数据处理与清洗
在获取到原始数据后，需要对其进行清洗和转换，以便后续的数据写入和分析。常见的数据清洗操作包括：
1. **去除重复数据**：确保每条记录唯一。
2. **格式化字段值**：例如日期格式转换、字符串去除空格等。
3. **字段映射**：将源系统的字段映射到目标系统的字段。
以下是一个简单的数据清洗示例：
```python
def clean_data(data):
cleaned_data = []
for record in data:
cleaned_record = {
'MaterialID': record['FMATERIALID'].strip(),
'Number': record['FNumber'].strip(),
'Name': record['FName'].strip(),
# 更多字段处理...
}
cleaned_data.append(cleaned_record)
return cleaned_data
cleaned_data = clean_data(response.json())
```
通过上述步骤，我们可以高效地从金蝶云星空获取所需数据，并进行必要的清洗和转换，为后续的数据集成打下坚实基础。
### 使用轻易云数据集成平台进行ETL转换并写入目标平台的技术案例
在数据集成生命周期的第二步，我们将已经从源平台（如金蝶）获取到的数据进行ETL转换，使其符合目标平台轻易云集成平台API接口所能接收的格式，最终写入目标平台。本文将详细探讨这一过程中的技术细节，特别是如何利用元数据配置实现这一目标。
#### 数据请求与清洗
首先，我们假设已经完成了从金蝶系统中获取物料单位数据的请求和清洗工作。此时，数据可能以JSON格式存储，包含多个字段，如`number`、`id`、`name`等。接下来，我们需要将这些数据转换为轻易云集成平台API能够接收的格式。
#### 数据转换与写入
根据提供的元数据配置，我们需要将清洗后的数据通过POST方法写入到轻易云集成平台。以下是元数据配置的具体内容：
```json
{
"api": "写入空操作",
"effect": "EXECUTE",
"method": "POST",
"number": "number",
"id": "id",
"name": "编码",
"idCheck": true
}
```
在这个配置中：
- `api`字段表示我们要调用的API接口名称。
- `effect`字段表示此次操作的效果类型，这里是执行操作（EXECUTE）。
- `method`字段指定了HTTP请求方法，这里使用POST。
- `number`、`id`和`name`字段分别映射了源数据中的对应字段。
- `idCheck`字段表示是否需要进行ID校验。
#### 实现步骤
1. **准备HTTP请求**：
我们需要构建一个HTTP POST请求，将处理后的数据发送到指定API接口。以下是Python代码示例：
```python
import requests
import json
# 假设已经获取并清洗后的数据
data = [
{"number": "001", "id": "1001", "name": "物料A"},
{"number": "002", "id": "1002", "name": "物料B"}
]
# API URL
api_url = 'https://api.qingyiyun.com/execute'
# 构建请求头
headers = {
'Content-Type': 'application/json'
}
# 遍历数据并发送POST请求
for item in data:
payload = {
'number': item['number'],
'id': item['id'],
'编码': item['name']
}
response = requests.post(api_url, headers=headers, data=json.dumps(payload))
if response.status_code == 200:
print(f"Data {item['number']} successfully written to target platform.")
else:
print(f"Failed to write data {item['number']}. Status code: {response.status_code}")
```
2. **处理响应**：
在发送POST请求后，我们需要检查响应状态码以确定操作是否成功。如果成功，则可以记录日志或执行后续操作；如果失败，则需要进行错误处理。
3. **ID校验**：
根据元数据配置中的`idCheck: true`，我们在发送每个POST请求前，可以先进行ID校验，以确保ID唯一性。这可以通过查询目标平台已有的数据来实现。如果发现重复ID，可以选择更新现有记录或跳过该条记录。
#### 技术细节讨论
1. **异步处理**：
为了提高效率，可以使用异步处理方式批量发送HTTP请求。例如，可以使用Python中的异步库（如aiohttp）来实现这一点。
2. **错误处理与重试机制**：
在实际应用中，网络问题或服务器故障可能导致部分请求失败。因此，需要设计一个重试机制，对失败的请求进行多次尝试，并记录失败原因以便后续分析。
3. **安全性考虑**：
在传输过程中，应确保数据加密传输（如使用HTTPS），并且对API接口进行身份验证，以防止未经授权的访问。
通过上述步骤和技术细节，我们可以高效地将从源平台获取的数据经过ETL转换后，写入到轻易云集成平台，实现不同系统间的数据无缝对接。这不仅提升了业务透明度和效率，也确保了数据的一致性和完整性。
