加快基本的 QuestDB 摄取速度

Question

我正在使用 docker compose 启动一个基本的 QuestDB 实例（下面的 yaml）。我想测试摄取，但不确定我做错了什么，因为我看到声称每秒可以摄取超过一百万个事件，但我似乎无法使用非常基本的脚本每秒摄取超过 100K 的事件.

这是我的yaml：

services:
  questdb:
    image: questdb/questdb:8.1.1
    container_name: rta_questdb
    restart: always
    ports:
      - "8812:8812"
      - "9000:9000"
      - "9009:9009"
      - "9003:9003"
    extra_hosts:
      - "host.docker.internal:host-gateway"
    environment:
      - QDB_METRICS_ENABLED=TRUE      
    volumes:
      - ./questdb/questdb_root:/var/lib/questdb/:rw

这是我模拟物联网数据的Python基本脚本。我没有提前创建表，而是依靠 QuestDB 自动创建表：

from questdb.ingress import Sender, IngressError, TimestampNanos
import os
import sys
import random


DEVICE_TYPES = ["blue", "red", "green", "yellow"]
EVENTS = 10000000
MIN_LAT = 19.50139
MAX_LAT = 64.85694
MIN_LON = -161.75583
MAX_LON = -68.01197

try:
    conf = f'http::addr=host.docker.internal:9000;'
    with Sender.from_conf(conf) as sender:
        for it in range(EVENTS):                
            sender.row(
                'iot_data',
                symbols={'device_type': random.choice(DEVICE_TYPES)},
                columns={
                            'duration_ms': random.randint(0, 4000),
                            "lat": random.uniform(MIN_LAT, MAX_LAT),
                            "lon": random.uniform(MIN_LON, MAX_LON),
                            "measure1": random.randint(-2147483648, 2147483647),
                            "measure2": random.randint(-2147483648, 2147483647),
                            "speed": random.randint(0, 100)
                },
                at=TimestampNanos.now())                                
except IngressError as e:
    sys.stderr.write(f'Got error: {e}')

我在 Macbook Pro M1 上运行此程序。

有什么想法为什么我没有获得广告中的性能吗？

Answer 1

上面的代码每秒“仅”插入 100K 事件有多种原因。

Mac 上的 Docker

第一个是 Mac 上的 docker。 QuestDB 可以在 Docker 中运行，但 Docker for Mac 有一些限制（当您在 Mac 上使用 Docker 启动 QuestDB 时，您会注意到日志和 Web 控制台上都会出现警告）。在 Linux 机器上的 Docker 上运行 QuestDB，或者在 Mac 上从二进制文件运行 QuestDB 应该会给您带来更好的开箱即用性能。在 M1 Macbook Pro 上测试相同的脚本可以得到每秒约 230K 事件。

并行化

虽然 QuestDB 每秒可以摄取数百万行，但这些数字是在并行建立多个连接时测量的。在单个连接上你无法获得那么多的吞吐量。在生产中，理想情况下，您应该并行运行多个脚本（如果您的数据来自 Kafka，则可能是多个 Kafka Connect 工作线程）。我们可以通过修改摄取脚本来测试并行发送数据的效果

from questdb.ingress import Sender, IngressError, TimestampNanos
import sys
import random
from multiprocessing import Pool
import math

DEVICE_TYPES = ["blue", "red", "green", "yellow"]
TOTAL_EVENTS = 10000000
NUM_SENDERS = 8

def send(sender_id, total_events):
    sys.stdout.write(f"Sender {sender_id} will send {total_events} events\n")
    try:
        conf = 'http::addr=localhost:9000;'
        with Sender.from_conf(conf) as sender:
            for _ in range(total_events):
                sender.row(
                    'iot_data',
                    symbols={'device_type': random.choice(DEVICE_TYPES)},
                    columns={
                        'duration_ms': random.randint(0, 4000),
                        'measure1': random.randint(-2147483648, 2147483647),
                        'measure2': random.randint(-2147483648, 2147483647),
                        'speed': random.randint(0, 100)
                    },
                    at=TimestampNanos.now()
                )
    except IngressError as e:
        sys.stderr.write(f'Sender {sender_id} got error: {e}\n')

if __name__ == '__main__':
    events_per_sender = TOTAL_EVENTS // NUM_SENDERS
    remaining_events = TOTAL_EVENTS % NUM_SENDERS

    sender_events = [events_per_sender] * NUM_SENDERS
    for i in range(remaining_events):
        sender_events[i] += 1

    with Pool(processes=NUM_SENDERS) as pool:
        sender_ids = range(NUM_SENDERS)
        pool.starmap(send, [(sender_id, sender_events[sender_id]) for sender_id in sender_ids])

运行脚本时，我可以看到每秒约 1,499,000 个事件。

配置设置

在生产环境中，您可能需要调整

QDB_LINE_TCP_WRITER_WORKER_COUNT

等配置参数和其他相关参数，这使我们能够配置共享池中有多少工作人员以及保留多少工作人员用于写入、预处理，等等。

这些都是低级参数，一旦你已经启动并运行，你就可以对其进行微调，但在这种特殊情况下，我相信 Mac 上的 Docker 和缺乏并行化是你摄取的最大瓶颈。