数据收集 SOP（一页纸执行版）+ 场景化脚本（直接复制用）

数据收集SOP（一页纸执行版）+场景化脚本（直接复制用）一、一页纸SOP（8步闭环，精简可执行）步骤核心动作输出物关键控制点1.需求确认业务提需求→三方评审→定目标/范围/口径/约束《采集需求说明书》《字段清单》需求不明确、合规不通过不启动2.数据源盘点梳理内/外部源→探查结构/质量/权限→评估可用性《数据源盘点表》《可用性报告》无权限/不可用源提前反馈，避免无效投入3.采集方案设计选全量/增量/实时→定工具→设计链路/断点/脱敏《采集技术方案》《脱敏规则表》优先CDC/增量，避免全量压垮源库4.环境准备部署组件→开通权限/网络→连通测试→小批量试采《环境部署文档》《连通测试报告》只读权限最小化，传输加密，试采无问题再批量5.采集执行全量初始化→增量/实时同步→监控→异常处理《采集执行日志》《监控报表》断点必记录，失败自动重试，7×24h监控告警6.数据校验验收条数/字段/主键校验→5大质量核查→业务验收→整改《数据校验报告》《验收单》技术+业务双重验收，不合格不交付7.归档交付按周期归档→交付数据/字典/权限→文档同步《归档清单》《数据字典》数据+文档一起交付，权限按角色管控8.监控迭代日常监控→定期巡检→适配源库/业务变化→版本管理《监控日报》《巡检报告》采集规则随业务迭代，重大变更必评审二、核心模板（直接复制到表格/文档）模板1：数据采集需求说明书项目内容需求编号DQ-YYYYMMDD-XXX提报人/部门业务目标（如：用户流失预警建模、门店销量日报）数据用途□报表□建模□监控□合规□其他时间范围历史回溯：；更新频率：□实时□小时□天□周主体范围（如：近1年注册≥30天、消费≥1次用户）必选字段（字段名、类型、含义）口径规则（如：有效订单=支付成功且未退款；活跃用户=近7天登录≥1次）过滤规则（如：排除测试账号、金额≤0订单）延迟要求核心≤2h，非核心≤24h合规要求□含PII（需脱敏）□需授权□需审计日志验收标准（如：缺失率≤1%，重复率≤0.1%，延迟达标）评审签字业务：数据：合规：日期：模板2：数据源盘点表数据源ID来源类型系统/平台表/接口名总数据量日增量主键缺失率权限状态更新方式负责人可用性备注DS-001内部数据库CRMuser_info100万5000user_id0.5%已开通只读增量（时间戳）张XX可用含手机号，需脱敏DS-002日志APPbehavior_log5000万/天5000万log_id1%已开通实时（Filebeat）李XX可用模板3：数据校验报告校验项校验内容校验结果阈值是否合格备注基础校验总记录数源库：100000，采集：99998误差≤0.1%□是□否2条测试数据过滤字段完整性必选字段25个，齐全必选字段无缺失□是□否主键唯一性user_id无重复重复率≤0.1%□是□否质量校验缺失率核心字段0.3%，非核心2%核心≤1%，非核心≤5%□是□否异常值占比0.2%（年龄>100岁）≤0.5%□是□否标记为异常，不删除及时性平均延迟1.5h≤2h□是□否业务验收样本核对随机100条，业务确认无误-□是□否指标核对日活跃用户数与报表一致误差≤1%□是□否整改意见验收签字技术：业务：日期：模板4：数据字典字段ID字段名数据类型含义来源清洗规则脱敏规则更新频率备注F001user_idBIGINT用户唯一IDCRM去重，保留最新无实时主键F002user_phoneVARCHAR(11)用户手机号CRM去空格，校验11位138****1234实时PII数据F003login_timeDATETIME登录时间APP日志格式化为YYYY-MM-DDHH:MM:SS无实时F004order_amountDECIMAL(10,2)订单金额订单系统过滤≤0，修正异常无小时三、场景化采集脚本（直接复制运行）1.MySQL增量采集（DataX，用户数据）json{"job":{"content":[{"reader":{"name":"mysqlreader","parameter":{"username":"readonly_user","password":"xxxxxx","column":["user_id","user_phone","create_time","update_time"],"connection":[{"jdbcUrl":["jdbc:mysql://00:3306/crm"],"table":["user_info"]}],"where":"update_time>'${last_time}'","splitPk":"user_id"}},"writer":{"name":"mysqlwriter","parameter":{"username":"data_user","password":"xxxxxx","column":["user_id","user_phone_mask","create_time","update_time"],"connection":[{"jdbcUrl":"jdbc:mysql://00:3306/ods","table":["ods_crm_user_info"]}],"preSql":["insertoverwritetableods_crm_user_infopartition(dt='${dt}')select*fromtmp_table"],"postSql":["updatedata_collection_logsetlast_time='${current_time}'wherejob_id='mysql_crm_user'"]}},"transformer":[{"name":"MaskTransformer","parameter":{"fieldIndex":1,"maskType":"phone"}}]}],"setting":{"speed":{"channel":3},"errorLimit":{"record":0,"percentage":0.001}}}}说明：按update_time增量采集，断点记录在data_collection_log表，手机号自动脱敏，写入ODS分区表。2.Python脚本（API采集，第三方物流数据）python运行importrequestsimportpandasaspdfromdatetimeimportdatetime,timedeltaimportos#配置信息API_URL="/logistics/track"API_KEY="your_api_key"OUTPUT_PATH="/data/ods/logistics_track"LAST_RUN_FILE="/data/log/last_run_time.txt"#获取上次运行时间defget_last_run_time():ifos.path.exists(LAST_RUN_FILE):withopen(LAST_RUN_FILE,"r")asf:returnf.read().strip()return(datetime.now()-timedelta(days=1)).strftime("%Y-%m-%d%H:%M:%S")#保存本次运行时间defsave_current_run_time():current_time=datetime.now().strftime("%Y-%m-%d%H:%M:%S")withopen(LAST_RUN_FILE,"w")asf:f.write(current_time)returncurrent_time#采集数据defcollect_data(start_time):headers={"Authorization":f"Bearer{API_KEY}"}params={"start_time":start_time,"page_size":1000,"page_num":1}all_data=[]whileTrue:response=requests.get(API_URL,headers=headers,params=params)response.raise_for_status()data=response.json()ifnotdata.get("data"):breakall_data.extend(data["data"])params["page_num"]+=1returnall_data#数据清洗defclean_data(data):df=pd.DataFrame(data)#去重df=df.drop_duplicates(subset=["logistics_id"])#缺失值处理df["receiver_phone"]=df["receiver_phone"].fillna("未知")#格式转换df["create_time"]=pd.to_datetime(df["create_time"]).dt.strftime("%Y-%m-%d%H:%M:%S")#脱敏df["receiver_phone_mask"]=df["receiver_phone"].str.replace(r"(\d{3})\d{4}(\d{4})",r"\1****\2",regex=True)returndf#主函数if__name__=="__main__":last_time=get_last_run_time()raw_data=collect_data(last_time)ifraw_data:clean_df=clean_data(raw_data)#保存到ODS层（按天分区）dt=datetime.now().strftime("%Y%m%d")output_file=f"{OUTPUT_PATH}/logistics_track_{dt}.parquet"clean_df.to_parquet(output_file,index=False)#保存运行时间save_current_run_time()print(f"采集完成，共{len(clean_df)}条数据，保存至{output_file}")else:print("无新数据")说明：定时拉取第三方物流API数据，自动处理断点续传、去重、缺失值、脱敏，输出Parquet格式到ODS层。3.Filebeat配置（日志实时采集，APP行为日志）yamlfilebeat.inputs:-type:filestreampaths:-/var/log/app/behavior_*.logfields:log_type:app_behaviorfields_under_root:truejson.keys_under_root:truejson.add_error_key:trueoutput.kafka:hosts:["kafka1:9092","kafka2:9092"]topic:"app_behavior_log"partition.round_robin:reachable_only:fal