首先，使用pydruid库操作apache druid需构建json查询并发送至druid集群；1. 安装pydruid：pip install pydruid；2. 使用querybuilder或直接构造json发送请求；3. 查询包含datasource、intervals、granularity、aggregations和dimensions等核心字段；4. 针对查询慢问题，优化方法包括：优化索引、合理分片、避免全表扫描、减少返回列数、使用limit、启用近似查询、开启缓存、调优资源、优化数据模型、避免复杂join、升级pydruid版本；5. 处理时间戳需注意iso 8601格式、数据摄入时配置timestampspec、查询时使用datetime对象、处理时区转换、合理设置时间粒度、处理null值及使用bound过滤；6. 对于复杂聚合如百分位数，需构造含quantilesdoublessketch聚合器和quantilesdoublessketchtoquantiles后聚合器的json查询，并确保datasketches扩展已加载；7. 可考虑使用druidapi等更高级库简化复杂查询构建。
Python操作Apache Druid，核心在于构建查询并发送给Druid集群。通常，你会使用
pydruid这个库，它简化了与Druid交互的过程。
pydruid查询的重点在于构造合适的JSON查询体，并处理Druid返回的结果。
解决方案
首先，你需要安装
pydruid：
立即学习“Python免费学习笔记（深入）”；
pip install pydruid
然后，就可以开始构建查询了。一个基本的Druid查询包含以下几个部分：
dataSource（数据源），
intervals（时间范围），
granularity（时间粒度），
aggregations（聚合操作）和
dimensions（维度）。
例如，要查询名为
wikipedia的数据源，统计过去24小时内每个小时的事件数量，可以这样写：
from pydruid.client import *
from pydruid.query import QueryBuilder
import datetime
client = PyDruid('http://your_druid_host:8082', 'druid/v2') # 替换为你的Druid Coordinator地址
q = QueryBuilder()
q.datasource('wikipedia')
q.intervals(datetime.datetime.now() - datetime.timedelta(days=1), datetime.datetime.now())
q.granularity('hour')
q.aggregator('count', 'events')
results = client.query(q)
for row in results:
print(row)这段代码首先连接到Druid Coordinator，然后使用
QueryBuilder构建一个查询。
dataSource指定了数据源，
intervals指定了查询的时间范围，
granularity指定了时间粒度为小时，
aggregator指定了聚合操作为统计事件数量。最后，使用
client.query(q)执行查询，并打印结果。
除了
QueryBuilder，你也可以直接构建JSON查询体，然后发送给Druid。这种方式更灵活，可以支持更复杂的查询。
import requests
import json
druid_url = 'http://your_druid_host:8082/druid/v2/?pretty' # 替换为你的Druid Coordinator地址
query = {
"queryType": "groupBy",
"dataSource": "wikipedia",
"intervals": [
f"{datetime.datetime.now() - datetime.timedelta(days=1)}/{datetime.datetime.now()}"
],
"granularity": "hour",
"dimensions": [],
"aggregations": [
{"type": "count", "name": "events"}
],
"limit": 10
}
headers = {'Content-Type': 'application/json'}
response = requests.post(druid_url, data=json.dumps(query), headers=headers)
if response.status_code == 200:
results = response.json()
for row in results:
print(row)
else:
print(f"Error: {response.status_code} - {response.text}")这段代码直接构建了一个JSON查询体，并使用
requests库发送给Druid。注意，你需要根据你的Druid集群配置修改
druid_url。
pydruid查询慢，有哪些优化方法？
索引优化： Druid的性能很大程度上取决于索引。确保你的数据源配置了合适的索引，特别是针对经常用于过滤和分组的维度。检查
indexSpec
，确保使用了合适的bitmap索引和column索引。例如，string
类型的维度列，可以考虑使用invertedIndex
。数据分片： Druid通过segment来组织数据。合理的分片策略可以提高查询性能。过大或过小的segment都会影响性能。一般来说，每个segment的大小在300MB到700MB之间比较合适。检查你的
segmentGranularity
配置。查询优化： 避免全表扫描。尽量使用时间范围过滤，缩小查询范围。尽量减少返回的列数，只选择需要的列。合理使用
limit
，避免返回过多的数据。资源调优： 确保你的Druid集群有足够的资源。增加Historical节点的数量，提高查询并发能力。调整Historical节点的内存大小，提高数据缓存能力。检查Coordinator和Overlord节点的配置，确保它们能够及时处理任务。
使用近似查询： 对于一些不需要精确结果的查询，可以使用近似查询，例如
approxCountDistinct
。近似查询可以显著提高查询性能。缓存： Druid有两级缓存：query cache和result cache。确保query cache开启，可以缓存最近的查询结果。Result cache可以缓存更细粒度的结果，但需要额外的配置。
监控和诊断： 使用Druid的监控工具，例如Druid Console，监控查询性能。分析查询日志，找出慢查询的原因。使用Druid的查询分析工具，例如
explain
命令，分析查询计划。数据建模： 优化数据模型。如果你的数据源包含多个维度，可以考虑使用rollup，预先计算一些聚合结果。Rollup可以显著提高查询性能，但会增加数据摄入的复杂度。
避免使用复杂的JOIN操作： Druid本身并不擅长复杂的JOIN操作。尽量在数据摄入阶段完成JOIN操作，或者使用lookup external。
Pydruid版本： 确保使用的
pydruid
是最新版本，新版本可能包含性能优化。
如何处理pydruid查询中的时间戳问题？
时间戳在Druid中至关重要，因为它用于数据分片和查询过滤。在
pydruid中处理时间戳，需要注意以下几点：
Druid的时间戳格式： Druid默认使用ISO 8601格式的时间戳，例如
2023-10-27T10:00:00.000Z
。确保你的时间戳数据符合这个格式。-
数据摄入： 在数据摄入阶段，需要指定时间戳列。通常，你会使用
timestampSpec
来配置时间戳列的名称和格式。如果你的时间戳数据不是ISO 8601格式，你需要指定format
参数。{ "type": "index_parallel", "spec": { "dataSchema": { "dataSource": "your_data_source", "timestampSpec": { "column": "your_timestamp_column", "format": "yyyy-MM-dd HH:mm:ss" }, "dimensionsSpec": { "dimensions": [ "dimension1", "dimension2" ] }, "metricsSpec": [ { "type": "count", "name": "count" } ] }, "ioConfig": { "type": "index_parallel", "inputSource": { "type": "local", "baseDir": "/path/to/your/data", "filter": "*.csv" }, "inputFormat": { "type": "csv", "columns": [ "your_timestamp_column", "dimension1", "dimension2" ] } }, "tuningConfig": { "type": "index_parallel", "maxRowsInMemory": 75000, "forceGuaranteedRollup": true } } } -
查询时间范围： 在
pydruid
查询中，你需要使用datetime
对象来指定时间范围。pydruid
会自动将datetime
对象转换为Druid需要的ISO 8601格式。from pydruid.client import * from pydruid.query import QueryBuilder import datetime client = PyDruid('http://your_druid_host:8082', 'druid/v2') q = QueryBuilder() q.datasource('your_data_source') q.intervals(datetime.datetime(2023, 10, 26), datetime.datetime(2023, 10, 27)) q.granularity('day') q.aggregator('count', 'events') results = client.query(q) for row in results: print(row) 时区问题： Druid内部使用UTC时间。如果你的时间戳数据不是UTC时间，你需要进行时区转换。可以在数据摄入阶段进行时区转换，也可以在查询阶段进行时区转换。
时间粒度： 在查询时，你需要指定时间粒度。时间粒度决定了Druid如何聚合数据。常见的时间粒度包括
all
，year
，month
，day
，hour
，minute
，second
。处理Null时间戳： 有时，你的数据可能包含Null时间戳。你需要决定如何处理这些Null时间戳。可以选择忽略这些数据，也可以选择使用默认时间戳填充。
-
时间戳过滤： 在查询时，可以使用时间戳过滤来缩小查询范围。可以使用
bound
过滤器来指定时间范围。{ "type": "timeseries", "dataSource": "your_data_source", "intervals": [ "2023-10-26T00:00:00.000Z/2023-10-27T00:00:00.000Z" ], "granularity": "day", "aggregations": [ { "type": "count", "name": "events" } ], "filters": [ { "type": "bound", "dimension": "__time", "lower": "2023-10-26T12:00:00.000Z", "upper": "2023-10-26T18:00:00.000Z", "ordering": "numeric" } ] }
pydruid如何进行更复杂的聚合查询，例如计算百分位数？
对于更复杂的聚合查询，例如计算百分位数，
pydruid可能不够直接。你需要构造更底层的JSON查询，利用Druid的
quantilesDoublesSketch聚合器。
首先，你需要确保你的Druid集群已经加载了
datasketches扩展。
然后，你可以构建如下的JSON查询：
import requests
import json
import datetime
druid_url = 'http://your_druid_host:8082/druid/v2/?pretty'
query = {
"queryType": "groupBy",
"dataSource": "your_data_source",
"intervals": [
f"{datetime.datetime.now() - datetime.timedelta(days=1)}/{datetime.datetime.now()}"
],
"granularity": "all",
"dimensions": [],
"aggregations": [
{
"type": "quantilesDoublesSketch",
"name": "value_sketch",
"fieldName": "your_value_column",
"k": 128 # 可选，控制精度，默认值是128
}
],
"postAggregations": [
{
"type": "quantilesDoublesSketchToQuantiles",
"name": "quantiles",
"field": {
"type": "fieldAccess",
"fieldName": "value_sketch"
},
"fractions": [0.25, 0.5, 0.75, 0.9, 0.99] # 要计算的百分位数
}
]
}
headers = {'Content-Type': 'application/json'}
response = requests.post(druid_url, data=json.dumps(query), headers=headers)
if response.status_code == 200:
results = response.json()
for row in results:
print(row)
else:
print(f"Error: {response.status_code} - {response.text}")这个查询首先使用
quantilesDoublesSketch聚合器计算
your_value_column的sketch。然后，使用
quantilesDoublesSketchToQuantilespost-aggregator计算指定的百分位数。
fieldName指定了要计算百分位数的列。
fractions指定了要计算的百分位数，例如
[0.25, 0.5, 0.75]表示计算25%，50%和75%的百分位数。
注意，
k参数控制了
quantilesDoublesSketch的精度。
k越大，精度越高，但内存消耗也越大。
除了百分位数，Druid还支持其他的复杂聚合操作，例如
approxCountDistinct（近似去重计数），
thetaSketch（用于集合操作）等。你可以根据你的需求选择合适的聚合器。
使用
pydruid构建复杂的JSON查询可能比较繁琐。你可以考虑使用其他的Python库，例如
druidapi，它提供了更高级的API，可以更方便地构建复杂的查询。
