Python怎样操作Apache Druid？pydruid查询-Python教程-PHP中文网

首先，使用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？pydruid查询

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。

PhotoG
PhotoG是全球首个内容营销端对端智能体

121

查看详情
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}")

登录后复制

这个查询首先使用