### 采购入库同步-江油市旌泰：汤臣倍健营销云数据集成到SQL Server 在企业运行过程中，实时、准确的数据交换是保障各业务系统顺利对接与协同运作的关键。本文将分享一个实际案例——如何通过轻易云数据集成平台，将汤臣倍健营销云中的采购入库单数据无缝同步到SQL Server中，方案名称为“采购入库同步-江油市旌泰”。 #### 集成背景与挑战 为了实现这一目标，我们需要解决以下几个技术难点： 1. **全量捕获与增量更新**：确保从汤臣倍健营销云获取的每一条采购单数据都能准确无误地写入SQL Server。 2. **高效大批量处理**：应对大量数据快速写入到SQL Server的问题，同时确保性能和稳定性。 3. **API接口调用频率控制**：处理分页及限流问题，以防止API请求过多导致被限制或阻断。 #### 数据获取流程设计 首先，我们要从汤臣倍健营销云接口`/erp/api/order/query/purInWarehsOrder`取得具体订单信息。采用定时任务机制，通过合理设置时间间隔，定期抓取新生成的订单数据进行后续处理。在这一步骤中，还必须设计好错误重试机制，以应对网络异常或其他临时失败情况。 #### 数据转换与格式适配 由于源系统（汤臣倍健营销云）和目标系统（SQL Server）之间的数据格式存在差异，需要提前设定好字段映射关系及类型转换逻辑。一旦抓取到原始的JSON格式订单数据，就需立即进行解析并转化为适合插入操作的结构体。 #### 高效写入及事务保证 对于已经完成初步清洗和整合的数据，将其动态批量提交至`insert`接口，实现高效且安全的数据存储。此过程中还包含事务管理策略，以确保在任何意外情况下均可回滚到正确状态，从而不影响整体数据库的一致性和完整性。 以上简要介绍了我们在这一实际项目中的技术思路和操作步骤。而接下来会详细分解各个模块如何具体实施，并结合代码示例展示整个实施过程中的关键节点及注意事项。  ### 调用源系统汤臣倍健营销云接口/erp/api/order/query/purInWarehsOrder获取并加工数据 在数据集成的生命周期中，调用源系统接口是至关重要的一步。本文将详细探讨如何通过轻易云数据集成平台调用汤臣倍健营销云的API接口`/erp/api/order/query/purInWarehsOrder`，并对获取的数据进行初步加工。 #### 接口调用配置 首先，我们需要配置API接口的请求参数。根据提供的元数据配置，可以看到该接口使用POST方法，并且需要传递多个参数来过滤和分页查询采购入库订单。 ```json { "api": "/erp/api/order/query/purInWarehsOrder", "method": "POST", "number": "number", "id": "id", "pagination": { "pageSize": 30 }, "idCheck": true, "request": [ {"field":"tenantId","label":"经销商id","type":"string","describe":"经销商id(必填,营销云id)如：34cc4109705e4c058b7b3b0352e57d31","value":"xxx"}, {"field":"yxyNumber","label":"营销云销售订单号","type":"string","describe":"如：YD1215710122031701，传此参数时，其他时间状态等条件无效"}, {"field":"number","label":"系统订单号","type":"string","describe":"如：XOUT0000000293，传此参数时,其他时间状态等条件无效"}, {"field":"status","label":"订单状态","type":"string","describe":"0:未审核 1：已审核（已出库）","value":"1"}, {"field":"beginTime","label":"开始时间","type":"string","describe":"timeType为空或者0，基于创建时间查询，timeType为1，基于更新时间查询，格式：0000-00-00或0000-00-00 00:00:00，如果不传单号此字段必填","value":"{{LAST_SYNC_TIME|datetime}}"}, {"field":"endTime","label":"结束时间","type":"string","describe":"timeType为空或者0，基于创建时间查询，timeType为1，基于更新时间查询，格式：0000-00-00或0000-00-00 00:00:00，如果不传单号此字段必填","value":"{{CURRENT_TIME|datetime}}"}, {"field":"pageNo","label":"页码","type":"string","describe":"默认1","value":"1"}, {"field":"pageSize","label":"每页条数","type":"string","describe":"默认30","value":"30"}, {"field":"timeType","label":"时间段标志","type":"string","describe":"查询时间段标识，0:创建时间（默认），1：最后更新时间","value":"1"} ], "condition_bk": [ [{"field": "itemList.materialNumber", "logic": "notlike", "value": "F.A"}] ], "condition": [ [{"field": "itemList.materialNumber", "logic": "notlike", "value": "F.A"}] ] } ``` #### 请求参数详解 1. **tenantId**: 经销商ID，这是一个必填项，用于标识具体的经销商。 2. **yxyNumber**: 营销云销售订单号，当传递该参数时，其它条件将被忽略。 3. **number**: 系统订单号，同样，当传递该参数时，其它条件将被忽略。 4. **status**: 订单状态，这里我们设置为已审核（值为1）。 5. **beginTime**和**endTime**: 时间范围，用于筛选在特定时间段内创建或更新的订单。 6. **pageNo**和**pageSize**: 分页参数，用于控制每次请求返回的数据量。 7. **timeType**: 时间段标志，用于指定是基于创建时间还是更新时间进行筛选。 #### 数据请求与清洗 在完成API请求配置后，我们可以通过轻易云平台发起请求，并获取返回的数据。由于返回的数据可能包含多种信息，我们需要对其进行初步清洗，以便后续处理。 例如，我们可以通过以下步骤进行数据清洗： 1. **过滤无效数据**：根据`condition`中的条件过滤掉不符合要求的数据。例如，这里我们过滤掉物料编号包含"F.A"的记录。 2. **字段映射与转换**：将返回的数据字段映射到目标系统所需的字段格式，并进行必要的类型转换。 3. **去重处理**：如果存在重复记录，需要进行去重处理，以确保数据的一致性。 #### 示例代码 以下是一个示例代码片段，用于展示如何通过轻易云平台调用API并进行初步数据清洗： ```python import requests import json # 配置API请求参数 url = 'https://api.example.com/erp/api/order/query/purInWarehsOrder' headers = {'Content-Type': 'application/json'} payload = { 'tenantId': '34cc4109705e4c058b7b3b0352e57d31', 'status': '1', 'beginTime': '2023-01-01', 'endTime': '2023-12-31', 'pageNo': '1', 'pageSize': '30', 'timeType': '1' } # 发起POST请求 response = requests.post(url, headers=headers, data=json.dumps(payload)) # 检查响应状态码 if response.status_code == 200: data = response.json() # 初步数据清洗 cleaned_data = [] for item in data['items']: if not item['materialNumber'].startswith('F.A'): cleaned_data.append({ 'order_id': item['id'], 'order_number': item['number'], # 更多字段映射... }) # 输出清洗后的数据 print(cleaned_data) else: print(f"Error: {response.status_code}") ``` 通过上述步骤，我们可以高效地调用汤臣倍健营销云的API接口，并对获取的数据进行初步加工，为后续的数据转换与写入奠定基础。  ### 轻易云数据集成平台中的ETL转换与写入SQL Server API接口 在数据集成生命周期的第二步中，重点是将已经集成的源平台数据进行ETL（Extract, Transform, Load）转换，并最终写入目标平台SQL Server。本文将详细探讨如何通过轻易云数据集成平台实现这一过程，特别是如何配置元数据以满足SQL Server API接口的要求。 #### 数据请求与清洗 首先，我们需要从源平台获取原始数据，并进行初步的清洗和整理。这一步骤包括解析源数据格式、过滤无效记录以及标准化字段值等操作。虽然这部分内容不在本文的重点范围内，但它为后续的ETL转换奠定了基础。 #### 数据转换与写入 接下来，我们进入关键的ETL转换阶段。根据提供的元数据配置，我们需要将原始数据转换为目标平台SQL Server API接口所能接受的格式，并通过API接口将其写入数据库。 以下是元数据配置中的关键部分及其解释： ```json { "api": "insert", "method": "POST", "idCheck": true, "request": [ { "label": "主表参数", "field": "main_params", "type": "object", "children": [ {"parent":"main_params","label":"单号编号","field":"djbh","type":"string","value":"{number}"}, {"parent":"main_params","label":"采购入库传CGC 销售退回传XHH","field":"djlx","type":"string","value":"CGC"}, {"parent":"main_params","label":"日期","field":"rq","type":"string","value":"{{opertime|date}}"}, {"parent":"main_params","label":"时间","field":"ontime","type":"string","value":"{{auditTime|time}}"}, {"parent":"main_params","label":"单位内码","field":"wldwid","type":"string","value":"WLD00000927"}, {"parent":"main_params","label":"含税金额","field":"hsje","type":"string","value":"{{itemList.taxlastmoney}}"}, {"parent":"main_params","label":"收货人","field":"shouhr","type":"string","value":"{recvContact}"}, {"parent":"main_params","label":"地址","field":"shhdz","type":"string","value":"{recvAddr}"}, {"parent":"main_params","label":"联系电话","field":"lxdh","type":"string","value":"{recvTel}"}, {"parent":"main_params","label":"备注","field":"beizhu","type":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","valueType":"","type":"","beizhu":{"orderNo}：订单备注：{remark}"},{"parent:"主表参数"：营销云单号"："webdjbh"："营销云单号"："webdjbh"："webdjbh"："webdjbh"："webdjbh"："webdjbh"："webdjbh"："webdjbh"："{yxyNumber}"} ] }, { "label": "扩展表参数", "field": "extend_params_1", "type": "array", "value": "itemList", "children": [ {"parent": "extend_params_1", "label": "单号", "field": "djbh", "type": "string", "value": "{number}"}, {"parent": "extend_params_1", "label": "序号", "field": "dj_sn", "type": "", value: "{oSn}"}, {"parent: extend_params_1, label: 商品内码, field: spid, type: string, value: _findCollection find spid from d76b64f9-f0e0-3436-a2d9-14c5579faa1b where spbh2={extMaterialNo}}, { parent: extend_params_1, label: 仓库编号, field: ckid, type: string, value: {{itemList.depotNo}}}, { parent: extend_params_1, label: 批号, field: pihao, type: string, value: {{itemList._Flot}}}, { parent: extend_params_1, label: 效期, field: sxrq, type: string, value: {{itemList._Fexp}}}, { parent: extend_params_1, label: 生产日期, field: baozhiqi, type:, value:, {{itemList._Fmfg}}}, { parent:, extend_, params_, 1,, label:, 数量,, 数量大于0,, field:, shl,, type:, string,, value:, {{itemList.opernumber}}}, { parent:, extend_, params_, 1,, label:, 含税价,, field:, hshj,, type:, string,, value:_function {{itemList.taxlastmoney}}/{{itemList.opernumber}}}, { parent:, extend_, params_, 1,, label:, 含税金额,, field:, hsje,, type:, string,, value:, {{itemList.taxlastmoney}}}, { parent:, extend_, params_, 1,, label:, 相关单号，如果退回，采购退 等原来的单号， field:xgdjbh,type:string,value:{yxyNumber}}, parent:"extend_params_1", label:"相关序号 如果退回，采购退 等对应明细的那个序号"， field:"recnum"， type:"string"， value:"{oSn}"} ,{"parent:"extend_:params_:1,"label:"组织ID,"field:"hzid,"type:"string,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"valuestring,"values:{orgId}, { parent:"extend_:params_:1,label:"仓库名称"， field:"ckname"， type:"string"， value:{tenantName}} ] } ], otherRequest:[{"label主SQL语句"， field main_sql， type string， values INSERT INTO gxkphz ( djbh,djlx,rq,onTime,wldwid,hseje,shehr,shehdz,lxdh,bizhu,weebdh ) values ( : djbh,: djlx,: rq,: ontime,: wldwid,: hsje,: shouhr,: shhdz,: lxdh,: beizhu,: webdjbh)}, { label扩展SQL语句1， field extend_sql_1， typestring， valuesINSERT INTO gxkpmx ( djbh,dj_sn.spid.ckid.pihao.sxrq.baozhiqi.shl.hshj.hsje.xgdjbh.recnum.hzid.ckname ) values ( : djbh,: dj_sn,: spid,: ckid,: pihao.: sxrq.: baozhiqi.: shl.: hshj.: hsje.: xgdjbh.: recnum.: hzid.: ckname)}] } ``` ##### 主表参数配置 主表参数包含了采购入库信息的基本字段，如单号编号、日期、时间、单位内码、含税金额等。这些字段需要映射到目标数据库中的相应字段。例如： - `djbh` 映射到 SQL Server 中的 `gxkphz` 表的 `djbh` 字段。 - `djlx` 固定值为 `CGC`。 - `rq` 和 `ontime` 分别映射到日期和时间字段，并通过模板函数进行格式化处理。 - `hsje` 从 `itemList.taxlastmoney` 获取并映射到 SQL Server 中的相应字段。 ##### 扩展表参数配置 扩展表参数则包含了详细商品信息，如商品内码、仓库编号、批号、数量等。这些字段需要逐条插入到目标数据库中的扩展表中。例如： - `spid` 使用 `_findCollection find spid from d76b64f9-f0e0-3436-a2d9-14c5579faa1b where spbh2={extMaterialNo}` 查找对应商品内码。 - `ckid`, `pihao`, `sxrq`, 等字段从 `itemList` 中提取并映射到 SQL Server 中相应字段。 ##### SQL 插入语句 最后，通过配置好的 SQL 插入语句，将上述转换后的数据写入目标数据库： ```sql INSERT INTO gxkphz (djbh,djlx,rq,ontime,wldwid,hsje,shouhr,shhdz,lxdh,beizhu,webdjbh) VALUES (:djbh,:djlx,:rq,:ontime,:wldwid,:hsje,:shouhr,:shhdz,:lxdh,:beizhu,:webdjbh); INSERT INTO gxkpmx (djbh,dj_sn,spid,ckid,pihao,sxrq,baozhiqi,shl,hshj,hsje,xgdjbh, recnum,hzid.ckname) VALUES (: dj bh , : dj sn , : sp id , : ck id , : pi hao , : sx rq , : bao zhi qi , : sh l , : hs hj , : hs je , : xg dj bh , : re c num , : hz id . c kn ame ); ``` 通过以上步骤和配置，我们能够高效地将源平台的数据进行ETL转换，并通过API接口写入到目标平台SQL Server中。这不仅提高了数据处理效率，还确保了数据的一致性和准确性。