如何配置Oozie Coordinator应用以触发外部数据源驱动的作业？

核心思路 #

要实现这个需求，核心是定期校验目标资源的版本状态，当状态变化时触发作业，常用的两种高效校验方式：

• 利用HTTP响应头的ETag或Last-Modified字段：这两个是服务器标识资源版本的标准字段，无需下载整个文件，带宽占用极低
• 对比文件哈希值：如果服务器不支持上述响应头，可下载文件后计算哈希（如MD5、SHA256），与上次记录的哈希对比（仅适合小文件）

代码示例（基于HTTP响应头校验） #

以下是用Python实现的轻量脚本，通过HEAD请求获取资源状态，判断更新并触发作业：

import requests
import json
from datetime import datetime

# 配置参数
TARGET_URL = "http://www.ic.gc.ca/folder/filename.zip"
STATE_FILE = "last_resource_state.json"

def get_resource_state(url):
    try:
        # 发送HEAD请求仅获取响应头，避免下载大文件
        response = requests.head(url, allow_redirects=True)
        response.raise_for_status()
        
        # 提取版本标识字段，优先ETag（更精准），其次Last-Modified
        state = {}
        if "ETag" in response.headers:
            state["etag"] = response.headers["ETag"]
        if "Last-Modified" in response.headers:
            state["last_modified"] = response.headers["Last-Modified"]
        
        return state
    except requests.exceptions.RequestException as e:
        print(f"资源状态检查失败: {e}")
        return None

def load_last_state():
    try:
        with open(STATE_FILE, "r") as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        # 首次运行无历史状态，返回空字典
        return {}

def save_current_state(state):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, indent=2)

def run_target_job():
    # 替换为你实际要执行的作业逻辑
    print(f"[{datetime.now()}] 检测到资源更新，开始执行作业...")
    # 示例：调用数据处理脚本、下载并解析文件等
    # import subprocess
    # subprocess.run(["python", "your_data_processing_script.py"])

def main():
    current_state = get_resource_state(TARGET_URL)
    if not current_state:
        return
    
    last_state = load_last_state()
    
    # 对比状态，判断资源是否更新
    if current_state != last_state:
        run_target_job()
        save_current_state(current_state)
    else:
        print(f"[{datetime.now()}] 资源未更新，跳过作业")

if __name__ == "__main__":
    main()

定时执行配置 #

脚本本身是单次检查，需要搭配定时工具实现周期性检测：

• Linux/macOS：使用cron定时任务。执行crontab -e添加规则，比如每天凌晨1点检查一次：

  0 1 * * * /usr/bin/python3 /path/to/your/script.py >> /path/to/logfile.log 2>&1
  

• Windows：使用「任务计划程序」，创建基本任务，设置触发周期（如每天），操作选择启动程序，指向Python解释器并传入脚本路径

注意事项 #

• 如果目标服务器不支持HEAD请求，可修改为GET请求并开启stream=True，获取响应头后立即关闭连接：

  response = requests.get(url, stream=True, allow_redirects=True)
  response.close()
  

• 对于大文件，优先用响应头校验，避免频繁下载浪费带宽
• 可扩展脚本的异常处理逻辑，比如添加邮件告警、日志归档等

内容的提问来源于stack exchange，提问作者Yan