MySQL数据集成到金蝶云星空的技术案例分享
在企业日常运营中，如何高效地完成数据从MySQL数据库到金蝶云星空的集成，是一个至关重要的问题。本文将围绕具体方案“SR新建调拨单内部供应商-深圳天一”展开，详细探讨这一过程中的关键技术点。
首先，为了确保大量MySQL数据能够快速写入金蝶云星空，我们利用其API接口batchSave进行了高吞吐量的数据写入配置。这种方法极大提升了数据处理的时效性，有助于企业及时掌握业务动态。在此过程中，我们还特别留意了分页和限流问题，通过自定义逻辑和合理参数设置来优化性能，确保不会因为瞬间的大规模数据传输而导致系统性能下降。
为了实现定时可靠的数据抓取，我们使用了MySQL提供的API接口select。这个接口不仅帮助我们实时获取最新的数据，同时也支持批量处理，使得从源头上保证了数据完整性。此外，在提取过程中涉及到的一系列监控机制，如异常检测与告警系统，也被部署以便随时应对潜在风险，从而保障整个流程始终稳定运行。
针对两端系统之间存在的数据格式差异问题，我们通过自定义数据转换逻辑进行有效解决。这一步骤包括编写专用映射规则，以适应特定的业务需求。同时，在实际操作中，还需仔细处理各种可能出现的不匹配情况，以避免数据丢失或错误引导等现象。
最后，为强化整体管理体验，此次项目特别采用可视化的数据流设计工具，对每个环节进行直观展示和操控。这不但使复杂流程变得更加易于理解与管理，也为未来可能需要的调整或扩展提供便利条件。此外，通过集中监控及告警系统，我们可以实时跟踪数据集成任务状态，并及时响应潜在异常，从而进一步增强任务执行效率与准确度。
调用源系统MySQL接口select获取并加工数据
在轻易云数据集成平台的生命周期中，调用源系统接口获取数据是至关重要的一步。本文将深入探讨如何通过MySQL接口select语句获取并加工数据，具体以“SR新建调拨单内部供应商-深圳天一”集成方案为例。
元数据配置解析
元数据配置是实现数据集成的关键，它定义了如何从源系统中提取数据。以下是本次集成方案的元数据配置：
{
"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。”
}
]
}
],
“otherRequest”: [
{
“field”: “main_sql”,
“label”: “主查询语句”,
“type”: “string”,
“describe”: “使用 :created_at 格式与主参数字段进行对应”,
“value”: “select CONCAT(d.confrim_no,'_',CAST(c.id AS CHAR)) as 单据编号,\nc.create_time as 日期,\nb.mode_no as 计划跟踪号,\nb.part_no as 物料编号,\nc.confirm_numb as 数量,\nb.business_no as 采购单号,\nb.ser_code as 条码,\nb.supplier_uuid as 供应商,\nc.id as sourceid,\nm.delivery_org as 供应组织\n#e.supplier_name\nfrom wms_instock_confirm_task_detail c\nleft join wms_instock_confirm_main_task_detail a on c.connect_uuid=a.uuid\nleft join wms_instock_purchase_task_detail b on b.uuid=a.connect_uuid\nleft join wms_instock_confirm_task d on d.uuid=c.confirm_detail_task_uuid\nleft join basic_supplier_info e on e.supplier_uuid=b.supplier_uuid\nleft join oms_order_bom f on f.bom_no=b.mode_no and f.part_no=b.part_no \nleft join mbs_order_plan_bom l on b.mode_no=l.bom_no\nleft join mbs_order_bom m on m.bom_uuid=l.bom_uuid\nwhere a.company_code='TYZN' #and a.status='10' #and a.confirm_status='20' #and b.task_type in ('2','3')\nand ((a.matterial_type='3' and b.next_process_code is null) or a.matterial_type='4') and f.material_source='2'\nand c.create_time>(select config_value from sys_config where config_id=340)\nand e.is_inner=0\nand c.is_success5<>'1'\nlimit :limit offset :offset”
}
],
“autoFillResponse”: true,
“omissionRemedy”: {
“crontab”: “1 1 1 1 1”,
“takeOverRequest”: []
}
}主查询语句分析
主查询语句是从MySQL数据库中提取所需数据的核心部分。以下是本次集成方案中的主查询语句：
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 oms_order_bom f on f.bom_no=b.mode_no and f.part_no=b.part_no
left join mbs_order_plan_bom l on b.mode_no=l.bom_no
left join mbs_order_bom m on m.bom_uuid=l.bom_uuid
where a.company_code='TYZN'
and ((a.matterial_type='3' and b.next_process_code is null) or a.matterial_type='4')
and f.material_source='2'
and c.create_time>(select config_value from sys_config where config_id=340)
and e.is_inner=0
and c.is_success5<>'1'
limit :limit offset :offset该查询语句包含多个表连接和条件过滤，确保提取的数据准确且符合业务需求。
动态参数应用
在元数据配置中，main_params定义了两个动态参数：limit和offset。这些参数用于控制查询结果的记录数和偏移量。
limit: 用于限制返回的记录数。在本次配置中，默认值为100。offset: 用于指定查询开始的位置，默认值为0。
这些参数在实际调用时会被替换到主查询语句中的相应位置，如下所示：
...
limit :limit offset :offset实际调用示例
假设我们需要获取前200条记录，并从第50条记录开始，可以通过如下方式设置动态参数：
{
...
main_params: {
limit: 200,
offset: 50
}
...
}然后执行以下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 oms_order_bom f on f.bom_no=b.mode_no and f.part_no=b.part_no
left join mbs_order_plan_bom l on b.mode_no=l.bom_no
left join mbs_order_bom m on m.bom_uuid=l.bom_uuid
where a.company_code='TYZN'
and ((a.matterial_type='3' and b.next_process_code is null) or a.matterial_type='4')
and f.material_source='2'
and c.create_time>(select config_value from sys_config where config_id=340)
and e.is_inner=0
and c.is_success5<>'1'
limit :200 offset :50;通过上述步骤，我们成功地调用了MySQL接口并获取了所需的数据。这一步骤不仅确保了数据的准确性，还为后续的数据清洗和转换奠定了基础。
总结来说，通过合理配置元数据和动态参数，我们可以高效地从MySQL数据库中提取所需的数据，为后续的数据处理提供坚实保障。
使用轻易云数据集成平台将数据转换并写入金蝶云星空API接口
在数据集成的生命周期中，ETL（Extract, Transform, Load）转换是一个关键步骤。本文将重点探讨如何使用轻易云数据集成平台，将已经集成的源平台数据进行ETL转换，并转为金蝶云星空API接口所能够接收的格式，最终写入目标平台。
元数据配置解析
在本案例中，我们需要将源平台的数据转换为金蝶云星空API接口所需的格式。以下是元数据配置的详细解析：
-
基本信息字段
FBillNo（单据编号）：直接从源数据中获取，类型为字符串。FBillTypeID（单据类型）：固定值“ZJDB01_SYS”，通过ConvertObjectParser解析为FNumber。FBizType（业务类型）：固定值“NORMAL”。FTransferDirect（调拨方向）：固定值“GENERAL”。FTransferBizType（调拨类型）：根据供应商和供应组织动态生成，使用函数判断。
-
组织信息字段
FSaleOrgId（销售组织）：通过ConvertObjectParser解析为FNumber。FSettleOrgId（结算组织）：根据供应商动态生成，通过函数判断。FStockOutOrgId（调出库存组织）：根据供应商动态生成，通过函数判断。FOwnerTypeOutIdHead（调出货主类型）：固定值“BD_OwnerOrg”。FOwnerOutIdHead（调出货主）：根据供应商动态生成，通过函数判断。
-
库存信息字段
FStockOrgId（调入库存组织）：根据供应组织动态生成，通过函数判断。FSETTLECURRID（结算币别）：固定值“PRE001”。FDate（日期）：直接从源数据中获取，类型为字符串。FNote（备注）：多行文本。
-
明细信息字段
明细信息包含多个子字段，每个子字段都需要进行相应的转换和解析：
FMaterialId（物料编码）：通过ConvertObjectParser解析为FNumber，从源数据中获取物料编号。FCMKBarCode（零售条形码）：文本类型，从源数据中获取。FQty（调拨数量）：数量类型，从源数据中获取。FSrcStockId（调出仓库）：根据供应商动态生成，通过函数判断。FDestStockId（调入仓库）：根据供应组织动态生成，通过函数判断。
-
其他请求参数
除了上述字段外，还需要配置一些其他请求参数：
FormId: 固定值“STK_TransferDirect”，表示业务对象表单ID。IsAutoSubmitAndAudit: 布尔值，表示是否自动提交并审核，设置为true。IsVerifyBaseDataField: 布尔值，表示是否验证基础资料有效性，设置为true。
实际操作步骤
-
定义请求结构
根据元数据配置定义请求结构，并确保所有字段都按照金蝶云星空API接口要求进行转换。例如：
{ "FormId": "STK_TransferDirect", "IsAutoSubmitAndAudit": true, "IsVerifyBaseDataField": true, "Operation": "Save", "InterationFlags": "STK_InvCheckResult", "Model": { "FBillNo": "{{单据编号}}", "FBillTypeID": {"FNumber": "ZJDB01_SYS"}, ... } } -
实现字段映射与转换
利用轻易云提供的解析器和函数，根据元数据配置中的规则，对每个字段进行映射与转换。例如：
{ "FSaleOrgId": {"FNumber": "_function case '{{供应商}}' when '804842' then 'T04' else '' end"} } -
发送请求至金蝶云星空API接口
将构建好的请求结构通过POST方法发送至金蝶云星空的batchSave API接口，并处理返回结果。例如：
import requests url = 'https://api.kingdee.com/batchSave' headers = {'Content-Type': 'application/json'} response = requests.post(url, json=request_data, headers=headers) if response.status_code == 200: print("Data successfully saved to Kingdee Cloud") else: print("Failed to save data:", response.text)
结论
通过以上步骤，我们可以成功地将已经集成的源平台数据进行ETL转换，并转为金蝶云星空API接口所能够接收的格式，最终写入目标平台。这不仅提高了数据处理的效率，也确保了不同系统之间的数据无缝对接。
