影刀RPA ETL流程自动化Airflow调度与任务编排什么情况用什么 → 怎么做 → 有什么坑作者林焱 | 飞行社出品什么情况用什么用RPA处理数据管道需要定时抽取、转换、加载数据手动管理依赖关系和失败重试效率低且容易出错。这套方案适合数据团队自动化ETL流程管理复杂的数据管道依赖自动化任务失败重试和告警核心工具影刀RPA Apache Airflow Celery任务队列怎么做第一步安装并配置Airflow# 安装Airflowpipinstallapache-airflow# 初始化数据库airflow db init# 启动调度器[video(video-AG3o9T9l-1783410739676)(type-csdn)(url-https://live.csdn.net/v/embed/525000)(image-https://v-blog.csdnimg.cn/asset/23da3fe1f67a47106d725406cfde9a97/cover/Cover0.jpg)(title-拼多多店群自动化上架方案)]airflow scheduler# 启动Web服务器airflow webserver-p8080fromairflowimportDAGfromairflow.operators.python_operatorimportPythonOperatorfromdatetimeimportdatetime,timedeltadefcreate_etl_dag():创建ETL流程DAGdefault_args{owner:data-team,depends_on_past:False,start_date:datetime(2026,6,1),retries:3,# 失败重试3次retry_delay:timedelta(minutes5),email_on_failure:True,email:[data-teamcompany.com]}dagDAG(etl_pipeline,default_argsdefault_args,description自动化ETL数据管道,schedule_interval0 2 * * *,# 每天凌晨2点catchupFalse)returndag# 使用示例etl_dagcreate_etl_dag()第二步定义ETL任务defextract_data(**context):抽取数据importpymysqlimportpandasaspd# 连接源数据库connpymysql.connect(hostsource-db.company.com,useretl_user,passwordsecret,databaseproduction)# 抽取数据query SELECT id, name, email, created_at FROM users WHERE created_at DATE_SUB(NOW(), INTERVAL 1 DAY) dfpd.read_sql(query,conn)conn.close()# 保存到临时文件output_path/tmp/extracted_data.csvdf.to_csv(output_path,indexFalse)# 推送文件路径到XComAirflow的任务间通信context[task_instance].xcom_push(keyextracted_file,valueoutput_path)print(f抽取完成{len(df)}条记录)returnoutput_pathdeftransform_data(**context):转换数据importpandasaspd# 从XCom获取抽取的文件路径ticontext[task_instance]input_pathti.xcom_pull(keyextracted_file)# 读取数据dfpd.read_csv(input_path)# 数据清洗和转换# 1. 去除重复记录dfdf.drop_duplicates(subset[id])# 2. 处理缺失值df[email]df[email].fillna(unknownexample.com)# 3. 添加衍生字段df[domain]df[email].apply(lambdax:x.split()[1]ifinxelseunknown)df[created_date]pd.to_datetime(df[created_at]).dt.date# 保存转换后的数据output_path/tmp/transformed_data.csvdf.to_csv(output_path,indexFalse)# 推送到XComti.xcom_push(keytransformed_file,valueoutput_path)print(f转换完成{len(df)}条记录)returnoutput_pathdefload_data(**context):加载数据到目标数据库importpandasaspdimportpymysql# 从XCom获取转换后的文件路径ticontext[task_instance]input_pathti.xcom_pull(keytransformed_file)# 读取数据dfpd.read_csv(input_path)# 连接目标数据库connpymysql.connect(hosttarget-db.company.com,useretl_user,passwordsecret,databasedata_warehouse)cursorconn.cursor()# 批量插入数据insert_query INSERT INTO users (id, name, email, domain, created_date) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE name VALUES(name), email VALUES(email), domain VALUES(domain) for_,rowindf.iterrows():cursor.execute(insert_query,(row[id],row[name],row[email],row[domain],row[created_date]))conn.commit()cursor.close()conn.close()print(f加载完成{len(df)}条记录)# 创建任务extract_taskPythonOperator(task_idextract_data,python_callableextract_data,dagetl_dag)transform_taskPythonOperator(task_idtransform_data,python_callabletransform_data,dagetl_dag)load_taskPythonOperator(task_idload_data,python_callableload_data,dagetl_dag)# 设置任务依赖extract_tasktransform_taskload_task第三步使用Celery分布式任务队列fromceleryimportCeleryimportpandasaspdimportpymysql# 创建Celery应用appCelery(etl_tasks,brokerredis://localhost:6379/0,backendredis://localhost:6379/0)app.task(bindTrue,max_retries3,default_retry_delay300)defprocess_data_chunk(self,chunk_path,target_table):处理数据块分布式执行try:# 读取数据块dfpd.read_csv(chunk_path)# 连接目标数据库connpymysql.connect(hosttarget-db.company.com,useretl_user,databasedata_warehouse)cursorconn.cursor()# 批量处理for_,rowindf.iterrows():# 数据验证逻辑ifpd.isna(row[email])ornotinstr(row[email]):continue# 跳过无效数据# 插入或更新query INSERT INTO target_table (id, name, email) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE name VALUES(name) cursor.execute(query,(row[id],row[name],row[email]))conn.commit()cursor.close()conn.close()returnf处理完成{len(df)}条记录exceptExceptionase:# 重试逻辑raiseself.retry(exce)# 并行处理多个数据块defdistribute_etl_work(file_path,chunk_size10000):分布式执行ETLimportpandasaspdimportos# 读取大文件并分块dfpd.read_csv(file_path)chunks[df[i:ichunk_size]foriinrange(0,len(df),chunk_size)]# 保存数据块到临时文件chunk_files[]fori,chunkinenumerate(chunks):chunk_pathf/tmp/chunk_{i}.csvchunk.to_csv(chunk_path,indexFalse)chunk_files.append(chunk_path)# 分发任务到Celery workerstasks[]forchunk_fileinchunk_files:taskprocess_data_chunk.delay(chunk_file,target_table)tasks.append(task)# 等待所有任务完成results[]fortaskintasks:resulttask.get(timeout3600)# 1小时超时results.append(result)# 清理临时文件forchunk_fileinchunk_files:os.remove(chunk_file)returnresults第四步监控和告警defmonitor_etl_pipeline(dag_id,execution_date):监控ETL管道执行状态fromairflow.modelsimportDagRunfromairflow.api.common.experimental.get_dag_runsimportget_dag_runsimportrequests# 获取DAG运行状态dag_runsget_dag_runs(dag_iddag_id)forrunindag_runs:ifrun.statefailed:# 发送告警send_etl_alert(dag_iddag_id,execution_daterun.execution_date,staterun.state,reason任务执行失败)elifrun.statesuccess:# 记录成功指标log_success_metrics(dag_id,run.execution_date)defsend_etl_alert(dag_id,execution_date,state,reason):发送ETL告警到企微/钉钉webhook_urlhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyYOUR_KEYcontentf **ETL管道告警** **DAG ID**{dag_id}**执行日期**{execution_date}**状态**{state}**原因**{reason}**时间**{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}请及时处理 payload{msgtype:markdown,markdown:{content:content}}responserequests.post(webhook_url,jsonpayload)ifresponse.json().get(errcode)0:print(fETL告警已发送:{dag_id})else:print(f发送ETL告警失败:{response.text})deflog_success_metrics(dag_id,execution_date):记录成功指标到监控系统# 可以集成Prometheus、CloudWatch等print(fETL执行成功:{dag_id}-{execution_date})第五步影刀RPA完整流程编排【定时触发】每天凌晨2:00 ↓ 【启动】Airflow DAG执行 ↓ 【执行任务】extract_data → transform_data → load_data ↓ 【监控状态】实时监控任务执行 ↓ 【条件判断】任务是否成功 ├─ 否 → 【企微告警】发送ETL失败通知 └─ 是 → 继续 ↓ 【记录日志】写入执行结果到数据库 ↓ 【生成报告】ETL执行日报.xlsx → 包含执行时间、处理记录数、状态 ↓ 【发送邮件】将报告发送给数据团队有什么坑坑1数据依赖管理复杂ETL任务之间有复杂依赖关系手动管理容易出错。解决方案使用Airflow DAG定义依赖清晰表达任务依赖实现数据感知调度下游任务检测上游数据可用性使用传感器Sensor等待外部数据可用fromairflow.sensors.filesystemimportFileSensorfromairflow.sensors.sqlimportSqlSensor# 文件传感器等待数据文件到达wait_for_data_fileFileSensor(task_idwait_for_data_file,filepath/data/input/daily_data.csv,poke_interval300,# 每5分钟检查一次timeout3600,# 1小时超时dagetl_dag)# SQL传感器等待数据满足条件wait_for_data_readySqlSensor(task_idwait_for_data_ready,conn_idsource_db,sql SELECT COUNT(*) FROM daily_data WHERE date {{ ds }} -- 使用Airflow宏 ,dagetl_dag)# 设置依赖wait_for_data_filewait_for_data_readyextract_task坑2大数据量处理内存溢出处理GB级别数据文件时可能内存溢出。解决方案分块处理使用Pandas的chunksize参数使用Dask或Spark分布式处理大数据流式处理使用生成器逐个处理记录# 分块读取大文件defprocess_large_file(file_path,chunk_size10000):分块处理大文件importpandasaspd chunk_iteratorpd.read_csv(file_path,chunksizechunk_size)results[]fori,chunkinenumerate(chunk_iterator):print(f处理第{i1}块记录数{len(chunk)})# 处理当前块processedtransform_chunk(chunk)# 保存到临时文件temp_filef/tmp/processed_chunk_{i}.csvprocessed.to_csv(temp_file,indexFalse)results.append(temp_file)# 合并所有处理后的块merge_processed_chunks(results)坑3任务失败重试策略不当重试次数过多或过少都可能导致问题。解决方案TEMU店群如何管理运营指数退避重试retry_delay 2^n区分可重试和不可重试错误实现死信队列处理多次重试失败的任务fromairflow.utils.emailimportsend_emailimporttimedefrobust_etl_task(**context):健壮的ETL任务带智能重试max_retries3base_delay60# 基础延迟60秒forattemptinrange(max_retries1):try:# 执行ETL逻辑resultdo_etl_work()returnresultexceptTemporaryErrorase:# 临时性错误网络连接、超时等ifattemptmax_retries:# 指数退避delaybase_delay*(2**attempt)print(f临时性错误{delay}秒后重试:{e})time.sleep(delay)continueelse:# 超过重试次数send_email(to[data-teamcompany.com],subjectfETL任务最终失败:{context[dag_run].dag_id},html_contentf错误:{e})raiseexceptPermanentErrorase:# 永久性错误数据格式错误、权限问题等# 不重试直接失败send_email(to[data-teamcompany.com],subjectfETL任务永久失败:{context[dag_run].dag_id},html_contentf永久性错误:{e})raise坑4数据质量监控不足ETL过程中数据质量问题缺失值、异常值、重复记录未及时发现。解决方案实现数据质量检查在转换步骤后添加验证设置数据质量阈值超出阈值触发告警数据质量仪表板可视化数据质量趋势defvalidate_data_quality(df,rules): 数据质量验证 rules: 验证规则列表 quality_report{total_rows:len(df),quality_issues:[]}forruleinrules:rule_typerule[type]ifrule_typenull_check:# 空值检查columnrule[column]null_countdf[column].isna().sum()ifnull_countrule.get(threshold,0):quality_report[quality_issues].append({rule:rule_type,column:column,null_count:null_count,severity:highifnull_countlen(df)*0.1elsemedium})elifrule_typerange_check:# 数值范围检查columnrule[column]min_valrule[min]max_valrule[max]out_of_rangedf[(df[column]min_val)|(df[column]max_val)]iflen(out_of_range)0:quality_report[quality_issues].append({rule:rule_type,column:column,out_of_range_count:len(out_of_range),severity:medium})# 如果有质量问题发送告警ifquality_report[quality_issues]:send_data_quality_alert(quality_report)returnquality_report# 在数据转换后调用quality_rules[{type:null_check,column:email,threshold:10},{type:range_check,column:age,min:0,max:120}]quality_reportvalidate_data_quality(transformed_df,quality_rules)总结功能节省时间附加价值自动化ETL调度每天省2小时减少人为错误提高数据及时性分布式任务处理每任务省30分钟处理大数据量提高处理速度智能监控告警—及时发现问题减少数据丢失数据质量验证—提高数据可靠性减少业务影响实际落地建议先小范围测试在一个非关键数据流上测试完整流程使用增量更新避免每次全量处理提高效率做好版本控制ETL脚本和配置纳入Git管理定期审查日志分析失败模式持续优化流程ETL自动化能为数据团队节省70%以上的数据处理时间同时提高数据质量和可靠性。
