【Python】阿里云 DataWorks ODPS 钉钉联动实战配合全搞定一、背景引言二、整体流程与核心模块三、龙虾 Python代码实现全流程1. 本地文件处理2. 终端响应云连接 ODPS upload 文件3. 云端处理执行代码SQL or Pyhton4. 结果通知4.1 系统导出4.2. 通知到位打工人的觉悟5. 全流程all in五、总结一、背景引言今天聊一个很实用的场景当电脑不在身边如何用小龙虾配合完成本地云端的数据任务流。背景说明今天和朋友A聚餐他临时收到领导信息急要根据最新的一份市场信息出最新报告报告有固定模板但是电脑不在身边领导要的急要么放弃聚餐回公司 – 太远太累要么…小伙伴们是否也遇到同样的情况 遇到休息时间打扰的领导怎么办万一有急事呢又不好耽误…别急小龙虾来帮你二、整体流程与核心模块咱们来梳理分析下核心问题和模块链路如下根据链路涉及的核心工具和服务模块工具/服务说明文件处理pandas读取文件、如Excel/CSV文件清洗转换等数据upload/计算pyodps-ALiYun SDK连接 ODPS文件upload、ETL、download等终端通知requests sign算法机器人推送 指定通知文件操作glob、os系统os 文件排查 传输等三、龙虾 Python代码实现全流程1. 本地文件处理我们以常见的excel为例注意最好先转先转 CSV因为 ODPS 的 Tunnel 上传接口对 CSV 支持最好避免遇到编码和格式问题。用pandas读 Excel 再导出 UTF-8 with BOM确保中文不乱码importpandasaspdimportglobimportosdefexcel_to_csv(input_dir,output_dir): 利用把手机文件传给指定目录读取目录下指定Excel文件这里加遍历解决多个文件的情况格式转化UTF-8 BOM # 匹配 xlsx 和 xls 两种格式解决office和wps的各种问题excel_filesglob.glob(os.path.join(input_dir,*.xlsx))\ glob.glob(os.path.join(input_dir,*.xls))ifnotexcel_files:print(f❌ 四叔未找到目标文件:{input_dir})return[]csv_files[]forexcel_fileinexcel_files:try:dfpd.read_excel(excel_file)# 转 CSVutf-8-sig UTF-8 with BOM防止 Windows 打开乱码csv_pathos.path.join(output_dir,os.path.splitext(os.path.basename(excel_file))[0].csv)df.to_csv(csv_path,indexFalse,encodingutf-8-sig)csv_files.append(csv_path)print(f✅ 四叔转换完成:{os.path.basename(excel_file)}-{os.path.basename(csv_path)})exceptExceptionase:print(f❌ 四叔转换失败:{str(e)})returncsv_files 小技巧Excel 文件可能有合并单元格、多 sheet 等情况pandas默认只读第一个 sheet。如果有多 sheet 需求可以遍历pd.ExcelFile.sheet_names逐个处理。大家可以注意下细心不会错。2. 终端响应云连接 ODPS upload 文件连接阿里云 ODPS 需要三样东西AccessKey ID、AccessKey Secret和project name这部分在【Python】Python解决阿里云DataWorks导出数据1万条限制的问题 中已经详细讲过不再赘述。fromodpsimportODPS ACCESS_IDAccessKeyIdACCESS_KEYAccessKeySecretPROJECT_NAMEproject nameENDPOINThttp://service.cn-shanghai.maxcompute.aliyun.com/api# 建立连接odpsODPS(ACCESS_ID,ACCESS_KEY,PROJECT_NAME,endpointENDPOINT)上传文件之前先对文件进行处理再逐个文件写入 ODPS 表并加上字段映射。这里用的是ODPS表字段 → Excel列名的反向映射保证上传的数据字段名和目标表对齐# 字段映射Excel列名 - ODPS表字段名 这里也可以自定义一个find_column 关键词遍历对比FIELD_MAPPING{字段1:xxx,字段2:xxx,字段3:xxx,...:...}defupload_to_odps(odps,csv_files,table_name,field_mapping): 先 TRUNCATE 表确保信息干净上传 CSV 数据到 ODPS 具体的大家根据情况定义即可 try:# 清空目标表防止历史数据残留print(f 清空表ing{table_name}...)odps.execute_sql(fTRUNCATE TABLE{table_name})print(f✅ 四叔表已清空)# 逐个 CSV 上传forcsv_fileincsv_files:dfpd.read_csv(csv_file,encodingutf-8-sig)# 反转映射reverse_mapping{v:kfork,vinfield_mapping.items()}df_renameddf.rename(columnsreverse_mapping)# 只保留目标表定义的字段防止多余列报错columns_to_keep[colforcolinfield_mapping.keys()ifcolindf_renamed.columns]df_finaldf_renamed[columns_to_keep]print(f 正在上传:{os.path.basename(csv_file)})odps.write_table(table_name,df_final)print(f✅ 上传成功)returnTrueexceptExceptionase:print(f❌ 四叔ODPS 操作失败:{str(e)})returnFalse⚠️ 注意ODPS 上传对字段类型和顺序敏感字段映射一定要提前确认好不然容易报Column xxx not found或类型不匹配的错误。3. 云端处理执行代码SQL or Pyhtonupload目标文件后在 ODPS 里可以执行代码了。代码逻辑根据业务需求来定并获取执行状态。SQL_SCRIPT XXXXXXX 这里放你的代码 以SQL为例 ; defexecute_sql_script(odps,sql_script): 执行 SQL 脚本完成后获取状态避免时间等待 try:instanceodps.execute_sql(sql_script)instance.wait_for_success()# 阻塞等待 SQL 完成print(f✅ 四叔SQL 执行成功)returnTrueexceptExceptionase:print(f❌ 四叔SQL 执行失败:{str(e)})returnFalse 划重点wait_for_success()建议加作为判断代码是否执行的标准避免队列wait or faile4. 结果通知4.1 系统导出代码执行完后根据结果导出。因为遇到过特殊情况这里顺带做了一个字段清洗处理掉 HTML 标签等非法字符避免 Excel 打开报格式错误。这部分可以看【Python】Python解决阿里云DataWorks导出数据1万条限制的问题 。4.2. 通知到位打工人的觉悟作为打工人完成任务更何况是加班完成的必须通知到群啊以钉钉为例最后一步推送钉钉群机器人通知。钉钉自定义机器人支持签名验证机制需要把当前时间戳 密钥做 HMAC-SHA256 签名具体内容之前在Python Mac 钉钉机器人推送 中也给大家讲过啦。 安全性提醒这里还是提醒下AccessKey 和钉钉 Webhook 地址属于敏感信息不要硬编码在代码里。建议放到环境变量或.env文件中代码里用os.getenv()读取防止泄露。5. 全流程all in把上面所欲步骤串联起来就是一个完整的**【Python】阿里云 DataWorks ODPS 钉钉联动实战 全流程**。最终只需要一条指令给到它就能帮你完成以上所有工作。附案例截图 小tips由于钉钉没法个人直接通过api传输文件所以最后完你老板后 需要借助将文件传到终端。defmain():# 链接云odpsodpsODPS(ACCESS_ID,ACCESS_KEY,PROJECT_NAME,endpointENDPOINT)# ① 文件处理csv_filesexcel_to_csv(INPUT_DIR,OUTPUT_DIR)# ②③ upload文件upload_to_odps(odps,csv_files,UPLOAD_TABLE_NAME,FIELD_MAPPING)# ④ 执行代码任务execute_sql_script(odps,SQL_SCRIPT)# ⑤ 导出结果export_from_odps(odps,EXPORT_TABLE_NAME,EXPORT_OUTPUT_DIR,EXPORT_FILENAME)# ⑥ 系统通知send_dingtalk_notification(DINGTALK_WEBHOOK,DINGTALK_SECRET,DINGTALK_AT_MOBILE)print(✅ 四叔全部流程执行完成)if__name____main__:main()五、总结今天咱们把整个链路拆开讲了朋友A看完操作直夸666不带电脑也不怕领导下班打扰了当然AI自动化的前提是积累才是王道。平时多积累多学习才能在关键时候派上用临时抱佛脚AI也不是万能的。挖掘场景。多发现业务和生活的场景需求训练成skills能达到事半功倍的效果。以后的AI一定是skills的时代。