如何在Python脚本中通过cURL执行Druid查询并导出CSV

1. 安装依赖 #

首先确保安装requests库（Python内置的csv模块无需额外安装）：

pip install requests

2. 完整实现代码 #

以下代码完全对应你提供的cURL逻辑，同时处理了嵌套字段的展开，并将结果导出为CSV：

import requests
import csv
from itertools import chain

def flatten_dict(d, parent_key='', sep='_'):
    """展开嵌套字典，将嵌套键转换为`父键_子键`的格式"""
    items = []
    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten_dict(v, new_key, sep=sep).items())
        else:
            items.append((new_key, v))
    return dict(items)

def main():
    # Druid SQL接口地址
    druid_url = 'http://druid.domain.vpc/druid/v2/sql'
    # 请求头（与cURL中的--header一致）
    headers = {'Content-Type': 'application/json'}
    # 查询参数（与cURL中的--data内容一致）
    query_payload = {
        "query": "Select * from demo_table"
    }

    try:
        # 发送POST请求，自动序列化字典为JSON
        response = requests.post(druid_url, headers=headers, json=query_payload)
        # 检查请求是否成功，失败则抛出异常
        response.raise_for_status()
        # 解析响应为JSON格式
        raw_results = response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {str(e)}")
        return

    # 展开所有结果中的嵌套字段（比如extended_attributes下的子字段）
    flattened_results = [flatten_dict(item) for item in raw_results]
    if not flattened_results:
        print("查询结果为空")
        return

    # 获取所有列名并排序
    all_columns = list(set(chain.from_iterable(res.keys() for res in flattened_results)))
    all_columns.sort()

    # 导出到CSV文件
    output_filename = 'druid_query_results.csv'
    with open(output_filename, 'w', newline='', encoding='utf-8') as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=all_columns)
        writer.writeheader()
        writer.writerows(flattened_results)

    print(f"结果已成功导出到文件: {output_filename}")

if __name__ == "__main__":
    main()

3. 代码说明 #

• 请求部分：用requests.post()替代cURL，json参数自动将字典转为JSON字符串，对应cURL中的--data参数
• 嵌套字段处理：flatten_dict函数将extended_attributes.aid这类嵌套键转为extended_attributes_aid，确保CSV能正常写入
• 错误处理：捕获请求过程中的网络错误、HTTP错误，避免程序崩溃
• CSV导出：使用csv.DictWriter自动处理表头和行数据，保证格式正确

4. 使用方法 #

1. 将代码保存为druid_query.py
2. 在终端运行：python druid_query.py
3. 执行完成后，当前目录会生成druid_query_results.csv文件

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