### MySQL数据集成至金蝶云星空案例分享：SR新建调拨单外部供应商-全-包-深圳天一
在企业日常运营中，高效的数据管理和流畅的系统对接是业务成功的关键。本文将直接切入一个真实案例，展示如何通过轻易云数据集成平台，将MySQL数据库中的数据集成到金蝶云星空系统。这次我们专注于 "SR新建调拨单外部供应商-全-包-深圳天一" 方案。
#### API接口实现与高效数据处理
首先，我们利用MySQL提供的数据获取API `select`来提取所需的数据。这个方法支持高度定制化的查询逻辑，能够精确地抽取符合特定需求的数据记录。在实际操作过程中，需要考虑分页和限流问题，以确保不会因大批量请求导致系统性能下降或超时错误。
```sql
SELECT * FROM orders WHERE supplier='外部供应商' AND status='new';
```
这个查询语句帮助我们从MySQL数据库中准确定位需要转移的新建调拨单记录。
#### 数据转换与质量监控
为了适应金蝶云星空的数据结构，我们使用自定义的数据转换逻辑进行预处理。这一步至关重要，因为不同系统之间可能存在格式上的差异。例如，某些字段类型需要进行单位换算或者格式调整，以便能顺利写入目标系统。
在完成初步准备之后，我们利用轻易云平台提供的集中监控和告警功能，对数据转换过程进行实时监测，并在发现异常情况时及时采取纠正措施。保障每条记录都能准确无误地传输到最终目的地，从而避免漏单或重复问题。
#### 写入金蝶云星空
下一步是将清洗后的数据批量写入金蝶云星空，通过其支持的大吞吐量API `batchSave`实现快速且高效的数据导入。同时，为了进一步提升可靠性，可以设置重试机制及异常处理策略，确保即使遇到瞬时网络抖动等意外情况也不影响整体任务进度。
```json
{
"api": "batchSave",
"data": [
{
"order_id": 12345,
"supplier_name": "深圳天一",
...
},
...
]
}
```
以上示例展示了JSON格式下的一组待写入数据信息，有助于理解实际执行中的具体参数配置和操作细节。
接下来部分将深入探讨这一方案各个步骤背后的实现细节、注意事项以及优化建议。在此基础上，希望为其他类似需求场景提供有价值的参考指引。
### 调用MySQL接口获取并加工数据的技术实现
在轻易云数据集成平台中，调用源系统MySQL接口`select`获取并加工数据是数据处理生命周期的第一步。本文将深入探讨如何通过元数据配置来实现这一过程。
#### 元数据配置解析
元数据配置是实现数据集成的核心。以下是我们使用的元数据配置：
```json
{
"api": "select",
"effect": "QUERY",
"method": "SQL",
"number": "单据编号",
"id": "单据编号",
"name": "name",
"request": [
{
"field": "main_params",
"label": "主参数",
"type": "object",
"describe": "对应主查询语句内的动态参数对象",
"children": [
{
"field": "limit",
"label": "返回的记录数",
"type": "int",
"describe": "你可以使用 LIMIT 属性来设定返回的记录数。",
"value": 100
},
{
"field": "offset",
"label": "数据偏移量",
"type": "int",
"describe": "你可以通过OFFSET指定SELECT语句开始查询的数据偏移量。默认情况下偏移量为0。"
}
]
}
],
...
}
```
#### 主查询语句
主查询语句是整个数据获取过程的核心部分。以下是配置中的主查询语句：
```sql
select CONCAT(d.confrim_no,'_',CAST(c.id AS CHAR)) as 单据编号,
c.create_time as 日期,
b.mode_no as 计划跟踪号,
b.part_no as 物料编号,
c.confirm_numb as 数量,
b.business_no as 采购单号,
b.ser_code as 条码,
b.supplier_uuid as 供应商,
c.id as sourceid,
m.delivery_org as 供应组织
from wms_instock_confirm_task_detail c
left join wms_instock_confirm_main_task_detail a on c.connect_uuid=a.uuid
left join wms_instock_purchase_task_detail b on b.uuid=a.connect_uuid
left join wms_instock_confirm_task d on d.uuid=c.confirm_detail_task_uuid
left join basic_supplier_info e on e.supplier_uuid=b.supplier_uuid
left join mbs_order_plan_bom l on b.mode_no=l.bom_no and l.company_code = a.company_code
left join mbs_order_bom m on m.bom_uuid=l.bom_uuid
where a.company_code='TYZN'
and b.task_type in ('2') and b.out_type='2'
and ((a.matterial_type='3' and b.next_process_code is null) or a.matterial_type='4')
and c.create_time>(select config_value from sys_config where config_id=340)
and e.is_inner=1
and m.delivery_org='T01.01'
and c.is_success5<>'1' and c.is_success4='1'
limit :limit offset :offset;
```
#### 动态参数对象
在上述SQL查询中，`:limit`和`:offset`是动态参数，它们通过元数据配置中的`main_params`字段进行传递。
```json
{
...
"request":[{"field":"main_params","label":"主参数","type":"object","describe":"对应主查询语句内的动态参数对象","children":[{"field":"limit","label":"返回的记录数","type":"int","describe":"你可以使用 LIMIT 属性来设定返回的记录数。","value":"100"},{"field":"offset","label":"数据偏移量","type":"int","describe":"你可以通过OFFSET指定SELECT语句开始查询的数据偏移量。默认情况下偏移量为0。"}]}],
...
}
```
#### 数据请求与清洗
在调用MySQL接口时，首先需要构建请求对象，并将其传递给API接口进行调用。以下是一个示例代码片段：
```python
import mysql.connector
def fetch_data(limit, offset):
query = """
select CONCAT(d.confrim_no,'_',CAST(c.id AS CHAR)) as 单据编号,
c.create_time as 日期,
b.mode_no as 计划跟踪号,
b.part_no as 物料编号,
c.confirm_numb as 数量,
b.business_no as 采购单号,
b.ser_code as 条码,
b.supplier_uuid as 供应商,
c.id as sourceid,
m.delivery_org as 供应组织
from wms_instock_confirm_task_detail c
left join wms_instock_confirm_main_task_detail a on c.connect_uuid=a.uuid
left join wms_instock_purchase_task_detail b on b.uuid=a.connect_uuid
left join wms_instock_confirm_task d on d.uuid=c.confirm_detail_task_uuid
left join basic_supplier_info e on e.supplier_uuid=b.supplier_uuid
left join mbs_order_plan_bom l on b.mode_no=l.bom_no and l.company_code = a.company_code
left join mbs_order_bom m on m.bom_uuid=l.bom_uuid
where a.company_code='TYZN'
and b.task_type in ('2') and b.out_type='2'
and ((a.matterial_type='3' and b.next_process_code is null) or a.matterial_type='4')
and c.create_time>(select config_value from sys_config where config_id=340)
and e.is_inner=1
and m.delivery_org='T01.01'
and c.is_success5<>'1' and c.is_success4='1'
limit %s offset %s;
"""
connection = mysql.connector.connect(
host="your_host",
user="your_user",
password="your_password",
database="your_database"
)
cursor = connection.cursor()
cursor.execute(query, (limit, offset))
result = cursor.fetchall()
cursor.close()
connection.close()
return result
# Example usage:
data = fetch_data(100, 0)
print(data)
```
#### 数据转换与写入
在获取到原始数据后，需要对其进行必要的清洗和转换，然后写入目标系统。这一步通常包括格式转换、字段映射等操作。
```python
def transform_and_write(data):
transformed_data = []
for record in data:
transformed_record = {
'document_number': record[0],
'date': record[1],
'tracking_number': record[2],
'material_number': record[3],
'quantity': record[4],
'purchase_order_number': record[5],
'barcode': record[6],
'supplier': record[7],
'source_id': record[8],
'delivery_organization': record[9]
}
transformed_data.append(transformed_record)
# Write to target system (e.g., another database, API endpoint)
# This part depends on the specific requirements of the target system.
return transformed_data
# Example usage:
transformed_data = transform_and_write(data)
print(transformed_data)
```
以上内容展示了如何利用轻易云平台提供的元数据配置，通过调用MySQL接口获取并加工数据。这一过程不仅确保了高效的数据集成，还为后续的数据处理奠定了坚实基础。
### 使用轻易云数据集成平台实现金蝶云星空API接口的数据转换与写入
在数据集成生命周期的第二阶段，我们需要将已经集成的源平台数据进行ETL（Extract, Transform, Load）转换，转为目标平台——金蝶云星空API接口所能够接收的格式，并最终写入目标平台。以下将详细介绍如何通过轻易云数据集成平台配置元数据，实现这一过程。
#### API接口配置
我们使用的API接口为`batchSave`，通过POST方法提交数据，并设置了多个字段以确保请求的正确性和完整性。以下是主要字段及其配置：
1. **单据编号 (FBillNo)**
- 类型：string
- 描述：单据编号
- 值：`{{单据编号}}`
2. **单据类型 (FBillTypeID)**
- 类型：string
- 描述：单据类型
- 解析器：ConvertObjectParser，参数为FNumber
- 值：`ZJDB01_SYS`
3. **业务类型 (FBizType)**
- 类型：string
- 描述：下拉列表
- 值：`NORMAL`
4. **调拨方向 (FTransferDirect)**
- 类型：string
- 描述：下拉列表
- 值：`GENERAL`
5. **调拨类型 (FTransferBizType)**
- 类型：string
- 描述：下拉列表
- 值：`OverOrgTransfer`
6. **销售组织 (FSaleOrgId)**
- 类型：string
- 描述：组织
- 解析器：ConvertObjectParser，参数为FNumber
7. **结算组织 (FSettleOrgId)**
- 类型：string
- 描述：组织
- 解析器：ConvertObjectParser，参数为FNumber
8. **调出库存组织 (FStockOutOrgId)**
- 类型：string
- 描述：组织
- 解析器：ConvertObjectParser，参数为FNumber
- 值：
```sql
_function
case '{{供应组织}}'
when 'T01.01' then 'T01.06'
when 'T04' then 'T04'
else ''
end
```
9. **调出货主类型 (FOwnerTypeOutIdHead)**
- 类型: string
- 描述: 多类别基础资料列表
- 值: `BD_OwnerOrg`
10. **调出货主 (FOwnerOutIdHead)**
- 类型: string
- 描述: 多类别基础资料
- 解析器: ConvertObjectParser，参数为FNumber
```sql
_function
case '{{供应组织}}'
when 'T01.01' then 'T01.06'
when 'T04' then 'T04'
else ''
end
```
11. **日期 (FDate)**
- 类型: string
- 描述: 日期
```sql
{{日期}}
```
12. **明细信息 (FBillEntry)**
明细信息部分是一个数组，其中包含多个子字段，例如物料编码、零售条形码、调拨数量等。这些字段也需要进行相应的转换和映射。
- **物料编码 (FMaterialId)**:
```sql
{{物料编号}}
```
- **调拨数量 (FQty)**:
```sql
{{数量}}
```
- **调出仓库 (FSrcStockId)**:
```sql
_function
case '{{供应组织}}'
when 'T01.01' then 'T0106CK091'
when 'T04' then 'T04CK008'
else ''
end
```
- **单位 (FUnitID)**:
```sql
_findCollection find FBaseUnitId_FNumber from 9f845ce5-cf31-3ae6-ab67-21f2d881489b where FNumber={{物料编号}}
```
#### 操作配置
除了上述字段，我们还需要配置一些操作选项：
1. **业务对象表单Id (FormId)**
```sql
STK_TransferDirect`
```
2. **提交并审核 (IsAutoSubmitAndAudit)**
```sql
true`
```
3. **验证基础资料有效性 (IsVerifyBaseDataField)**
```sql
true`
```
4. **执行的操作 (Operation)**
```sql Save```
5.**允许负库存(InterationFlags)
```sql STK_InvCheckResult```
#### 数据转换与写入流程
1、**提取数据**：
从源系统中提取原始数据，包括单据编号、日期、物料编码等。
2、**数据清洗**：
根据业务需求，对提取的数据进行清洗和预处理，例如去除无效记录、标准化字段格式等。
3、**数据转换**：
使用轻易云提供的元数据配置，将清洗后的数据转换为金蝶云星空API所需的格式。特别注意使用解析器（如ConvertObjectParser）对特定字段进行转换。
4、**数据写入**：
通过调用金蝶云星空API接口，将转换后的数据批量写入目标系统。确保每个请求都包含必要的字段和操作选项，以保证数据的一致性和完整性。
#### 注意事项
1、确保所有必填字段都已正确映射，并且值符合金蝶云星空API的要求。
2、使用解析器（如ConvertObjectParser）对复杂字段进行正确转换。
3、在实际操作中，可以通过日志和监控功能实时跟踪数据流动和处理状态，以便及时发现并解决问题。
通过以上步骤，我们可以高效地将源平台的数据转换并写入到金蝶云星空系统中，实现不同系统间的数据无缝对接。
