dnf手游360客户端
1.71 GB · 2025-11-12
传统爬虫调度常面临两个痛点：要么用crontab这种简单工具，但缺乏任务依赖管理；要么用Jenkins等CI工具，却不够灵活。Airflow的出现解决了这些矛盾——它用有向无环图（DAG）管理任务依赖，支持分钟级调度，还能通过Web界面监控任务状态。
举个真实案例：某电商公司需要每天采集竞品价格，涉及3个爬虫（列表页→详情页→价格校验）。用crontab时，详情页爬虫常因列表页未完成而报错。改用Airflow后，通过设置depends_on_past=True和wait_for_downstream=True，任务自动按顺序执行，错误率下降90%。
想象把爬虫任务拆解成乐高积木：每个积木块是一个Task，用箭头连接表示执行顺序。比如：
with DAG('ecommerce_spider',
schedule_interval='0 8 * * *', # 每天8点执行
catchup=False) as dag:
task1 = PythonOperator(task_id='fetch_list', python_callable=spider_list)
task2 = PythonOperator(task_id='fetch_detail', python_callable=spider_detail)
task3 = PythonOperator(task_id='validate_price', python_callable=validate_price)
task1 >> task2 >> task3 # 定义执行顺序
| 参数 | 作用 | 示例 |
|---|---|---|
schedule_interval | 执行频率 | '@daily' 或 '0 */6 * * *'（每6小时） |
start_date | 首次执行时间 | datetime(2023,1,1) |
retries | 失败重试次数 | retries=3 |
retry_delay | 重试间隔 | retry_delay=timedelta(minutes=5) |
将Scrapy/Requests爬虫封装成可调用函数：
def spider_list(ds, **kwargs):
# ds是执行日期参数，可用于动态构造URL
url = f"https://example.com/products?date={ds}"
response = requests.get(url, proxies=get_proxy()) # 使用代理
save_to_db(response.json())
通过template_fields实现动态参数：
class DynamicSpiderOperator(PythonOperator):
template_fields = ('url', 'date') # 这些字段会被渲染
def execute(self, context):
url = self.url.format(date=context['ds'])
# 执行爬取...
场景1：详情页必须等列表页完成
解决方案：在详情页Task中设置trigger_rule='all_done'
detail_task = PythonOperator(
task_id='fetch_detail',
trigger_rule='all_done', # 即使上游失败也执行
python_callable=spider_detail
)
场景2：周末不执行校验任务
解决方案：用TimeSensor或自定义BranchPythonOperator
def should_run(**context):
return context['ds'].weekday() < 5 # 周一到周五
branch_task = BranchPythonOperator(
task_id='check_weekday',
python_callable=should_run,
trigger_rule='all_success'
)
当单节点性能不足时，可采用：
配置示例（airflow.cfg）：
[core]
executor = CeleryExecutor
[celery]
broker_url = redis://localhost:6379/0
result_backend = redis://localhost:6379/0
通过on_failure_callback实现邮件报警：
def send_failure_email(context):
task_id = context['task_instance'].task_id
error = context['exception']
send_mail(
subject=f"Airflow任务失败: {task_id}",
body=str(error),
to_emails=["[email protected]"]
)
task = PythonOperator(
task_id='critical_spider',
on_failure_callback=send_failure_email,
# ...其他参数
)
在爬取后添加校验Task：
def validate_data(ds, **kwargs):
df = pd.read_sql("SELECT * FROM products WHERE date=?", params=[ds])
if len(df) < 100: # 低于阈值报警
raise ValueError("数据量不足")
通过Prometheus+Grafana监控：
airflow_task_instance_successairflow_task_instance_durationairflow_scheduler_heartbeat并行度调整：
[core]
parallelism = 32 # 默认32，可根据CPU核心数调整
结果持久化：
task = PythonOperator(
task_id='save_results',
python_callable=save_data,
provide_context=True,
output_encoding='utf-8' # 避免编码问题
)
日志分级：
import logging
logging.getLogger("airflow.task").setLevel(logging.WARNING) # 减少日志量
Q1：被网站封IP怎么办？
A：立即启用备用代理池，建议使用住宅代理（如站大爷IP代理），配合每请求更换IP策略。代码示例：
import random
PROXY_POOL = [
"http://1.1.1.1:8080",
"http://2.2.2.2:8080",
# ...更多代理
]
def get_proxy():
return {"http": random.choice(PROXY_POOL)}
Q2：如何避免重复爬取？
A：使用execution_date作为唯一标识，结合数据库去重：
def spider_with_dedup(ds, **kwargs):
if db.exists(url=f"https://example.com/item/{ds}"):
return # 已爬取则跳过
# 执行爬取...
Q3：Airflow和Scrapy如何配合？
A：两种方式：
封装Scrapy为命令行：
BashOperator(
task_id='run_scrapy',
bash_command='scrapy crawl myspider -a date={ds}'
)
直接调用Scrapy API：
from scrapy.crawler import CrawlerProcess
from myproject.spiders import MySpider
def run_scrapy(ds):
process = CrawlerProcess()
process.crawl(MySpider, start_date=ds)
process.start()
Q4：任务卡住不执行怎么办？
A：按以下步骤排查：
airflow-scheduler日志airflow workers）Q5：如何实现补数（回填历史数据）？
A：修改DAG的catchup参数并指定start_date：
with DAG('historical_spider',
schedule_interval='@daily',
start_date=datetime(2023,1,1),
catchup=True) as dag: # catchup=True会生成所有未执行的任务
# ...任务定义
doc_md注释说明业务逻辑Airflow不是银弹，但它是目前最平衡的爬虫调度解决方案。通过合理设计DAG和参数，可以构建出既稳定又灵活的定时采集系统。实际部署时建议先在测试环境运行一周，观察任务成功率、执行时间分布等指标后再上线生产。
