前置准备 #

• Apache IoTDB v1.3+ 已部署，且在iotdb-common.properties中设置enable_pipe_service=true
• PostgreSQL服务运行正常，已创建目标库与对应数据表（表结构需匹配IoTDB的时间序列模型）
• Python环境安装依赖包：

  pip install iotdb-pipe-client psycopg2-binary
  

1. IoTDB端创建并启动Pipe #

在IoTDB CLI中执行以下SQL，创建监听指定表的Pipe：

CREATE PIPE demo_pipe
WITH (
  'extractor' = 'iotdb-extractor',
  'extractor.iotdb.filter' = 'root.sg.demo.*', -- 监听root.sg.demo下所有时间序列
  'extractor.iotdb.fetchLatest' = 'false', -- 仅捕获实时新增数据，不回溯历史
  'processor' = 'default-processor',
  'sink' = 'rpc-sink' -- 通过RPC将数据推送给Python客户端
);

-- 启动Pipe
START PIPE demo_pipe;

可根据需求调整extractor.iotdb.filter，比如指定单个时间序列root.sg.demo.device1.temperature，或用通配符匹配更多路径。

2. Python端数据接收与批量写入PostgreSQL #

以下是完整的Python脚本，负责从IoTDB Pipe拉取数据并批量写入PostgreSQL：

from iotdb.pipe.client import PipeClient
import psycopg2
from psycopg2.extras import execute_values

# IoTDB Pipe连接配置
IOTDB_HOST = "localhost"
IOTDB_PORT = 6667
PIPE_NAME = "demo_pipe"

# PostgreSQL连接配置
PG_HOST = "localhost"
PG_PORT = 5432
PG_DB = "iotdb_sink"
PG_USER = "postgres"
PG_PWD = "your_password"
PG_TABLE = "iotdb_demo_data"

def get_pg_connection():
    """获取PostgreSQL连接"""
    return psycopg2.connect(
        host=PG_HOST,
        port=PG_PORT,
        database=PG_DB,
        user=PG_USER,
        password=PG_PWD
    )

def main():
    # 初始化IoTDB Pipe客户端
    pipe_client = PipeClient(IOTDB_HOST, IOTDB_PORT)
    pipe_client.connect()

    # 初始化PostgreSQL连接与游标
    pg_conn = get_pg_connection()
    pg_cursor = pg_conn.cursor()

    print(f"开始监听Pipe {PIPE_NAME} 的实时数据...")
    try:
        while True:
            # 批量拉取数据，batch_size可根据数据量调整
            batch_records = pipe_client.fetch(PIPE_NAME, batch_size=100)
            if not batch_records:
                continue

            # 转换数据格式为PostgreSQL插入所需的元组列表
            insert_rows = []
            for record in batch_records:
                ts = record.get_timestamp()
                device = record.get_device()
                measurements = record.get_measurements()
                values = record.get_values()

                # 适配你的PostgreSQL表字段，这里假设表有ts、device、temperature、humidity字段
                temp = values[measurements.index("temperature")] if "temperature" in measurements else None
                humi = values[measurements.index("humidity")] if "humidity" in measurements else None
                insert_rows.append((ts, device, temp, humi))

            # 批量插入，提升写入效率
            if insert_rows:
                insert_sql = f"""
                INSERT INTO {PG_TABLE} (ts, device, temperature, humidity)
                VALUES %s
                ON CONFLICT (ts) DO NOTHING; -- 可选：避免重复写入同一时间戳数据
                """
                execute_values(pg_cursor, insert_sql, insert_rows)
                pg_conn.commit()
                print(f"成功写入 {len(insert_rows)} 条数据")

    except KeyboardInterrupt:
        print("用户中断，停止数据同步")
    except Exception as e:
        print(f"同步出错：{str(e)}")
        pg_conn.rollback()
    finally:
        # 关闭所有连接
        pipe_client.close()
        pg_cursor.close()
        pg_conn.close()

if __name__ == "__main__":
    main()

3. 配套PostgreSQL表结构示例 #

提前创建对应的数据表，示例SQL：

CREATE TABLE iotdb_demo_data (
    ts BIGINT PRIMARY KEY,
    device VARCHAR(100) NOT NULL,
    temperature DOUBLE PRECISION,
    humidity DOUBLE PRECISION
);

4. 性能优化要点 #

• 批量参数调整：IoTDB端可在创建Pipe时添加'extractor.iotdb.batchSize' = '1000'，增大单次提取的数据量；Python端调整batch_size平衡实时性与写入性能
• PostgreSQL优化：调整max_wal_size、shared_buffers等参数提升写入能力；开启autovacuum维护表性能
• 连接池：生产环境建议使用psycopg2.pool.SimpleConnectionPool替代单次连接，减少连接开销
• 异常重试：可添加重试逻辑，处理网络波动或数据库临时不可用的情况

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