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接口所需的格式。以下是元数据配置的详细解析：

1. 基本信息字段

   • FBillNo（单据编号）：直接从源数据中获取，类型为字符串。
   • FBillTypeID（单据类型）：固定值“ZJDB01_SYS”，通过ConvertObjectParser解析为FNumber。
   • FBizType（业务类型）：固定值“NORMAL”。
   • FTransferDirect（调拨方向）：固定值“GENERAL”。
   • FTransferBizType（调拨类型）：根据供应商和供应组织动态生成，使用函数判断。

2. 组织信息字段

   • FSaleOrgId（销售组织）：通过ConvertObjectParser解析为FNumber。
   • FSettleOrgId（结算组织）：根据供应商动态生成，通过函数判断。
   • FStockOutOrgId（调出库存组织）：根据供应商动态生成，通过函数判断。
   • FOwnerTypeOutIdHead（调出货主类型）：固定值“BD_OwnerOrg”。
   • FOwnerOutIdHead（调出货主）：根据供应商动态生成，通过函数判断。

3. 库存信息字段

   • FStockOrgId（调入库存组织）：根据供应组织动态生成，通过函数判断。
   • FSETTLECURRID（结算币别）：固定值“PRE001”。
   • FDate（日期）：直接从源数据中获取，类型为字符串。
   • FNote（备注）：多行文本。

4. 明细信息字段

   明细信息包含多个子字段，每个子字段都需要进行相应的转换和解析：

   • FMaterialId（物料编码）：通过ConvertObjectParser解析为FNumber，从源数据中获取物料编号。
   • FCMKBarCode（零售条形码）：文本类型，从源数据中获取。
   • FQty（调拨数量）：数量类型，从源数据中获取。
   • FSrcStockId（调出仓库）：根据供应商动态生成，通过函数判断。
   • FDestStockId（调入仓库）：根据供应组织动态生成，通过函数判断。

5. 其他请求参数

   除了上述字段外，还需要配置一些其他请求参数：

   • FormId: 固定值“STK_TransferDirect”，表示业务对象表单ID。
   • IsAutoSubmitAndAudit: 布尔值，表示是否自动提交并审核，设置为true。
   • IsVerifyBaseDataField: 布尔值，表示是否验证基础资料有效性，设置为true。

实际操作步骤

1. 定义请求结构

   根据元数据配置定义请求结构，并确保所有字段都按照金蝶云星空API接口要求进行转换。例如：

   {
       "FormId": "STK_TransferDirect",
       "IsAutoSubmitAndAudit": true,
       "IsVerifyBaseDataField": true,
       "Operation": "Save",
       "InterationFlags": "STK_InvCheckResult",
       "Model": {
           "FBillNo": "{{单据编号}}",
           "FBillTypeID": {"FNumber": "ZJDB01_SYS"},
           ...
       }
   }

2. 实现字段映射与转换

   利用轻易云提供的解析器和函数，根据元数据配置中的规则，对每个字段进行映射与转换。例如：

   {
       "FSaleOrgId": {"FNumber": "_function case '{{供应商}}' when '804842' then 'T04' else '' end"}
   }

3. 发送请求至金蝶云星空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接口所能够接收的格式，最终写入目标平台。这不仅提高了数据处理的效率，也确保了不同系统之间的数据无缝对接。