如何在Python中自动化修改Zeppelin Notebook的SQL查询日期条件？

一、在Zeppelin SQL中定义变量的可行方式 #

1. 段落内固定变量名定义（适配你现有写法） #

统一使用固定变量名（比如date1、start_date）定义日期，后续替换时只匹配变量行，完全规避索引失效问题：

date1 = '2023-01-01'
select e_segment_map['style_guide'] style, count(f_device_id) good
    from events
    where e_key = 'User_Feedback'
    and e_segment_map['feedback'] = 'good'
    and a_receive_day >= date1
    and e_segment_map['type'] = '6'
group by e_segment_map['style_guide']

2. Zeppelin原生变量替换（更推荐） #

改用Zeppelin SQL原生支持的${变量名}格式，无需在SQL内写赋值行，直接在查询中引用：

select e_segment_map['style_guide'] style, count(f_device_id) good
    from events
    where e_key = 'User_Feedback'
    and e_segment_map['feedback'] = 'good'
    and a_receive_day >= ${date1}
    and e_segment_map['type'] = '6'
group by e_segment_map['style_guide']

这种方式支持通过API全局设置变量，不用逐个修改段落SQL。

二、修改Python脚本实现自动化替换 #

基于你的现有代码，调整为变量名匹配替换，同时让非技术用户无需改动代码核心逻辑：

1. 优化后的脚本（适配自定义变量定义） #

import requests
from requests.auth import HTTPBasicAuth
import re
import json

def updateZeppelinQueries(target_date):
    # 配置信息集中管理，用户无需修改这里
    config = {
        "zeppelin_url": "https://query-ntu.perfectcorp.com/zeppelin/api",
        "notebook_id": "2J4GY9RVJ",
        "auth": ("sinfulheinz", "Tj7g&tENQ/d-PFnX"),
        "variable_name": "date1"  # 统一的日期变量名
    }

    # 获取Notebook所有段落
    notebook_url = f"{config['zeppelin_url']}/notebook/{config['notebook_id']}"
    notebook_info = requests.get(notebook_url, auth=HTTPBasicAuth(*config['auth'])).json()['body']
    paragraphs = notebook_info['paragraphs']

    # 遍历更新每个段落的日期变量
    for para in paragraphs:
        para_id = para['id']
        original_text = para['text']
        # 正则匹配变量定义行，替换为目标日期
        updated_text = re.sub(
            rf"{config['variable_name']}\s*=\s*'[^']+'",
            f"{config['variable_name']} = '{target_date}'",
            original_text
        )
        if updated_text == original_text:
            continue
        # 推送更新到Zeppelin
        update_url = f"{notebook_url}/paragraph/{para_id}"
        requests.put(update_url, auth=HTTPBasicAuth(*config['auth']),
                     json={"text": updated_text})
        print(f"段落 {para_id} 日期已更新为 {target_date}")

# 非技术用户仅需修改目标日期，或自动计算日期
if __name__ == "__main__":
    # 自动计算上周一例（无需手动输入日期）：
    # from datetime import datetime, timedelta
    # last_monday = (datetime.now() - timedelta(days=datetime.now().weekday())).strftime('%Y-%m-%d')
    # updateZeppelinQueries(last_monday)
    
    # 手动指定日期（用户仅改此行即可）
    updateZeppelinQueries('2023-10-09')

2. 极简配置方式（适合纯业务用户） #

把日期放在单独的config.json文件中，用户只需修改这个文件，完全不用碰Python代码：

{
    "target_date": "2023-10-09"
}

脚本中读取配置的代码：

if __name__ == "__main__":
    with open('config.json', 'r') as f:
        config = json.load(f)
    updateZeppelinQueries(config['target_date'])

三、进阶：全局变量批量更新 #

如果所有SQL都用相同日期变量，直接通过API设置Zeppelin全局变量，无需修改任何段落内容：

def setZeppelinGlobalVariable(variable_name, variable_value):
    config = {
        "zeppelin_url": "https://query-ntu.perfectcorp.com/zeppelin/api",
        "notebook_id": "2J4GY9RVJ",
        "auth": ("sinfulheinz", "Tj7g&tENQ/d-PFnX")
    }
    set_var_url = f"{config['zeppelin_url']}/notebook/{config['notebook_id']}/set"
    requests.post(set_var_url, auth=HTTPBasicAuth(*config['auth']),
                 json={"name": variable_name, "value": f"'{variable_value}'"})

# 调用示例：设置全局date1变量
setZeppelinGlobalVariable('date1', '2023-10-09')

所有引用${date1}的SQL段落会自动生效。

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