Holo Client Python - Hologres

holo-client-py是Hologres官方Python用戶端，基於psycopg3實現，提供同步（HoloClient）與非同步（AsyncHoloClient）兩種API，適用於大批量資料寫入和高QPS點查情境。本文為您介紹holo-client-py的選型、安裝、資料寫入、Copy寫入、資料查詢、效能比較與HoloConfig參數配置方法。

holo-client-py是Hologres官方Python用戶端，基於psycopg3實現，適用於大批量資料寫入（批量、即時同步至Hologres）和高QPS點查（維表關聯）情境。holo-client-py同時提供同步（HoloClient）和非同步（AsyncHoloClient）兩種API，支援普通寫入、主鍵衝突處理、Copy高速寫入、完整主鍵查詢與Scan查詢等能力，相較於直接使用原生psycopg或JDBC，批量寫入和點查效能更優。源碼詳情請參見holo-client-py。

Hologres已與Python生態實現高效整合，協助您在Python應用中快速構建即時資料倉庫。如果您使用Java語言串連Hologres，請參見Holo Client。

說明

本文程式碼範例中的<HOST>、<PORT>、<DATABASE>、<USERNAME>和<PASSWORD>均為預留位置，請分別替換為Hologres執行個體的網路地址、連接埠、資料庫名稱，以及登入Hologres的使用者名稱和密碼。使用阿里雲帳號或RAM使用者登入時，使用者名稱和密碼分別為AccessKey ID和AccessKey Secret；使用BASIC帳號登入時，為BASIC帳號的使用者名稱和密碼。

選型指南

holo-client-py與Psycopg 3均可用於在Python應用中訪問Hologres，兩者的主要區別如下。

寫入效能：holo-client-py提供批量Put、Fixed Copy流式協議、Stage寫入等最佳化，單進程可達數萬TPS；Psycopg 3需要手動管理批量寫入和串連池，效能調優複雜。
查詢模式：holo-client-py專註主鍵點查和範圍掃描，API簡潔；Psycopg 3支援完整SQL，適合複雜查詢，但需要手寫SQL。
異常處理：holo-client-py內建重試和批量錯誤彙總；Psycopg 3需要手動處理串連異常和交易回復。

結合Hologres DataFrame，不同情境下推薦的訪問方式如下。

情境	推薦方式
高吞吐批量寫入（日誌、監控資料）	Holo Client Python（`Put` API+多進程）
高QPS主鍵點查（即時特徵查詢）	Holo Client Python（`Get` API）
流式資料寫入（Kafka、Flink消費）	Holo Client Python（Fixed Copy協議）
資料分析與ETL（彙總、JOIN）	Hologres DataFrame
複雜SQL查詢（報表、BI）	Psycopg 3或Hologres DataFrame
pandas生態整合	Hologres DataFrame

Hologres DataFrame的使用方法，請參見Hologres DataFrame；通過Psycopg 3串連Hologres，請參見Python。

使用限制

同步模式下，HoloClient使用多進程和共用記憶體實現並行寫入；非同步模式下，AsyncHoloClient使用多個asyncio writer task並發寫入。
Python版本為3.8及以上，psycopg[binary]版本為3.1及以上。
Fixed Copy自Hologres V1.3版本起支援，適用於無DELETE操作的即時寫入情境。

安裝Connector

您可以通過pip直接安裝holo-client-py。

pip install hologres-client

如需從源碼安裝，請執行如下命令。

cd holo-client-py
pip install -e .

說明

pip安裝包名為hologres-client，開源倉庫中的源碼目錄名為holo-client-py。

串連數說明

holo-client-py會根據配置建立對應數量的資料庫連接，串連數說明如下。

HoloClient會根據write_parallelism和read_parallelism建立對應數量的串連。
空閑超過connection_max_idle_ms的串連會被釋放。
寫入資料前，建議確認執行個體的剩餘可用串連數，詳情請參見串連數管理。

-- 查看最大串連數
select instance_max_connections();

-- 查看已使用串連數
select count(*) from pg_stat_activity where backend_type='client backend';

資料寫入

建議在專案中建立HoloClient或AsyncHoloClient單例，並通過write_parallelism控制寫入並發。

同步用戶端（HoloClient）：內部使用多進程和共用記憶體實現並行寫入。
非同步用戶端（AsyncHoloClient）：內部使用多個asyncio writer task並發寫入，適用於asyncio應用。

寫入普通表

同步介面

使用HoloClient寫入普通表的程式碼範例如下。

from hologres import HoloClient, HoloConfig, Put

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
    write_parallelism=4,  # 同步模式下啟動4個worker進程並行寫入
)

with HoloClient(config) as client:
    schema = client.get_table_schema("t0")
    put = Put(schema)
    put.set_object("id", 1)
    put.set_object("name", "name0")
    put.set_object("address", "address0")
    client.put(put)
    # 強制提交所有未提交put請求；HoloClient內部也會根據write_batch_size、write_batch_byte_size、write_max_interval_ms自動認可
    # client.flush()

非同步介面

使用asyncio.create_task批量提交put請求，讓writer task充分攢批，以獲得最高吞吐。請避免逐個await client.put()，否則每批僅包含1條請求。

import asyncio
from hologres import AsyncHoloClient, HoloConfig, Put

async def main():
    config = HoloConfig(
        host="<HOST>",
        port=<PORT>,
        database="<DATABASE>",
        username="<USERNAME>",
        password="<PASSWORD>",
        write_parallelism=4,  # 非同步模式下啟動4個writer task並發刷入
    )

    async with AsyncHoloClient(config) as client:
        schema = await client.get_table_schema("t0")

        # 批量提交put請求，不逐個await
        tasks = []
        for i in range(100):
            put = Put(schema)
            put.set_object("id", i)
            put.set_object("name", f"name{i}")
            put.set_object("address", f"address{i}")
            tasks.append(asyncio.create_task(client.put(put)))

        # 統一等待所有結果
        await asyncio.gather(*tasks)
        await client.flush()

asyncio.run(main())

寫入含主鍵表

同步介面

寫入含主鍵表時，您可以通過on_conflict_action配置主鍵衝突時的處理策略，程式碼範例如下。

from hologres import HoloClient, HoloConfig, Put, OnConflictAction

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
    on_conflict_action=OnConflictAction.INSERT_OR_REPLACE,  # 配置主鍵衝突時策略
)

with HoloClient(config) as client:
    # create table t0(id int not null, name0 text, address text, primary key(id))
    schema = client.get_table_schema("t0")
    put = Put(schema)
    put.set_object("id", 1)
    put.set_object("name0", "name0")
    put.set_object("address", "address0")
    client.put(put)

    put = Put(schema)
    put.set_object("id", 1)
    put.set_object("name0", "newName")
    put.set_object("address", "newAddress")
    client.put(put)
    # client.flush()

非同步介面

使用asyncio.create_task批量提交put請求，讓writer task充分攢批，以獲得最高吞吐。

import asyncio
from hologres import AsyncHoloClient, HoloConfig, Put, OnConflictAction

async def main():
    config = HoloConfig(
        host="<HOST>",
        port=<PORT>,
        database="<DATABASE>",
        username="<USERNAME>",
        password="<PASSWORD>",
        on_conflict_action=OnConflictAction.INSERT_OR_REPLACE,
    )

    async with AsyncHoloClient(config) as client:
        schema = await client.get_table_schema("t0")

        tasks = []
        for i in range(100):
            put = Put(schema)
            put.set_object("id", i)
            put.set_object("name0", f"name{i}")
            put.set_object("address", f"address{i}")
            tasks.append(asyncio.create_task(client.put(put)))

        await asyncio.gather(*tasks)
        await client.flush()

asyncio.run(main())

局部更新主鍵表

同步介面

將on_conflict_action設定為INSERT_OR_UPDATE可實現局部更新，主鍵衝突時僅更新Put中已設定的列，程式碼範例如下。

from hologres import HoloClient, HoloConfig, Put, OnConflictAction

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
    on_conflict_action=OnConflictAction.INSERT_OR_UPDATE,  # 局部更新
)

with HoloClient(config) as client:
    # create table t0(id int not null, name0 text, address text, primary key(id))
    schema = client.get_table_schema("t0")

    # 只put id和name0兩個欄位，當主鍵衝突則更新對應欄位，否則寫入
    put = Put(schema)
    put.set_object("id", 1)
    put.set_object("name0", "name0")
    client.put(put)

    put = Put(schema)
    put.set_object("id", 1)
    put.set_object("name0", "newName")
    client.put(put)
    # client.flush()

非同步介面

import asyncio
from hologres import AsyncHoloClient, HoloConfig, Put, OnConflictAction

async def main():
    config = HoloConfig(
        host="<HOST>",
        port=<PORT>,
        database="<DATABASE>",
        username="<USERNAME>",
        password="<PASSWORD>",
        on_conflict_action=OnConflictAction.INSERT_OR_UPDATE,
    )

    async with AsyncHoloClient(config) as client:
        schema = await client.get_table_schema("t0")

        tasks = []
        for i in range(100):
            put = Put(schema)
            put.set_object("id", i)
            put.set_object("name0", f"name{i}")
            tasks.append(asyncio.create_task(client.put(put)))

        await asyncio.gather(*tasks)
        await client.flush()

asyncio.run(main())

基於主鍵刪除

說明

DELETE佔比提高會降低整體的每秒寫入量。

同步介面

將mutation_type設定為MutationType.DELETE，即可基於完整主鍵刪除資料，程式碼範例如下。

from hologres import HoloClient, HoloConfig, Put, MutationType, OnConflictAction

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
    on_conflict_action=OnConflictAction.INSERT_OR_REPLACE,
)

with HoloClient(config) as client:
    # create table t0(id int not null, name0 text, address text, primary key(id))
    schema = client.get_table_schema("t0")
    put = Put(schema)
    put.mutation_type = MutationType.DELETE
    put.set_object("id", 1)
    client.put(put)
    # client.flush()

非同步介面

import asyncio
from hologres import AsyncHoloClient, HoloConfig, Put, MutationType, OnConflictAction

async def main():
    config = HoloConfig(
        host="<HOST>",
        port=<PORT>,
        database="<DATABASE>",
        username="<USERNAME>",
        password="<PASSWORD>",
        on_conflict_action=OnConflictAction.INSERT_OR_REPLACE,
    )

    async with AsyncHoloClient(config) as client:
        schema = await client.get_table_schema("t0")

        tasks = []
        for i in range(100):
            put = Put(schema)
            put.mutation_type = MutationType.DELETE
            put.set_object("id", i)
            tasks.append(asyncio.create_task(client.put(put)))

        await asyncio.gather(*tasks)
        await client.flush()

asyncio.run(main())

Copy寫入

Fixed Copy自Hologres V1.3版本起支援。相比HoloClient.put方法，Fixed Copy方式可以獲得更高的吞吐（流模式）、更低的資料延遲和更低的用戶端記憶體消耗（不攢批）。對於無DELETE操作的即時寫入情境，建議使用Fixed Copy寫入。

Fixed Copy寫入普通表

同步介面

copy_writer使用HoloConfig中配置的衝突策略，程式碼範例如下。

from hologres import HoloClient, HoloConfig, Put, CopyFormat, CopyMode, OnConflictAction

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
    on_conflict_action=OnConflictAction.INSERT_OR_UPDATE,  # copy_writer使用HoloConfig中的衝突策略
)

with HoloClient(config) as client:
    # CREATE TABLE copy_demo (id INT NOT NULL, name TEXT NOT NULL, address TEXT, PRIMARY KEY(id));
    columns = ["id", "name"]

    with client.copy_writer(
        "copy_demo",
        columns=columns,
        fmt=CopyFormat.BINARY,
        mode=CopyMode.STREAM,
    ) as writer:
        for i in range(10):
            put = Put(writer.schema)
            put.set_object("id", i)
            put.set_object("name", "name0")
            writer.write(put)

非同步介面

import asyncio
from hologres import AsyncHoloClient, HoloConfig, Put, CopyFormat, CopyMode, OnConflictAction

async def main():
    config = HoloConfig(
        host="<HOST>",
        port=<PORT>,
        database="<DATABASE>",
        username="<USERNAME>",
        password="<PASSWORD>",
        on_conflict_action=OnConflictAction.INSERT_OR_UPDATE,
    )

    async with AsyncHoloClient(config) as client:
        columns = ["id", "name"]

        async with await client.copy_writer(
            "copy_demo",
            columns=columns,
            fmt=CopyFormat.BINARY,
            mode=CopyMode.STREAM,
        ) as writer:
            for i in range(10):
                put = Put(writer.schema)
                put.set_object("id", i)
                put.set_object("name", "name0")
                await writer.write(put)

asyncio.run(main())

通過Stage寫入

holo-client-py支援通過CopyStageWriter將資料寫入Hologres內部Stage，再通過INSERT語句將資料從Stage載入到目標表。該方式適合需要先暫存再大量匯入的情境，具有原子性。Stage的詳細說明，請參見通過臨時儲存Stage近即時匯入。

from hologres import HoloClient, HoloConfig, Put

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
)

with HoloClient(config) as client:
    # CREATE TABLE copy_stage_demo (id INT NOT NULL, name TEXT NOT NULL, address TEXT, PRIMARY KEY(id));
    columns = ["id", "name", "address"]
    stage_name = "my_stage"

    # 建立內部Stage，TTL為7200秒（2小時），逾時後自動清理
    client.create_stage(stage_name, ttl_seconds=7200)

    try:
        with client.copy_stage_writer(
            "copy_stage_demo",
            stage_name,
            columns=columns,
        ) as writer:
            for i in range(10):
                put = Put(writer.schema)
                put.set_object("id", i)
                put.set_object("name", f"name{i}")
                put.set_object("address", f"address{i}")
                writer.write(put)
    finally:
        # 清理臨時Stage（不清理也會根據TTL自動清理）
        client.drop_stage(stage_name)

資料查詢

holo-client-py提供基於完整主鍵的點查（Get、async_get）和按主鍵首碼範圍掃描（Scan）兩類查詢介面，均支援同步與非同步呼叫。

基於完整主鍵查詢

同步介面

使用Get基於完整主鍵查詢資料的程式碼範例如下。

from hologres import HoloClient, HoloConfig, Get

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
    read_parallelism=4,  # 同步模式下啟動4個reader線程並行查詢
)

with HoloClient(config) as client:
    # create table t0(id int not null, name0 text, address text, primary key(id))
    schema = client.get_table_schema("t0")

    get = Get.builder(schema).set_primary_key("id", 1).build()
    record = client.get(get)
    if record:
        print(record.get_object("name0"))

非阻塞介面（async_get）

async_get()返回concurrent.futures.Future，適用於同步用戶端下需要高QPS點查的情境。您可以先批量提交多個Get請求，再統一等待結果，從而讓reader線程充分攢批執行。

from hologres import HoloClient, HoloConfig, Get

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
    read_parallelism=4,
)

with HoloClient(config) as client:
    schema = client.get_table_schema("t0")

    # 批量提交get請求，不阻塞等待結果
    futures = []
    for i in range(100):
        get = Get.builder(schema).set_primary_key("id", i).build()
        futures.append(client.async_get(get))

    # 統一等待所有結果
    for future in futures:
        record = future.result()
        if record:
            print(record.get_object("name0"))

非同步介面

使用asyncio.create_task批量提交Get請求，讓reader task充分攢批，以獲得最高吞吐。請避免逐個await client.get()，否則每批僅包含1條請求。

import asyncio
from hologres import AsyncHoloClient, HoloConfig, Get

async def main():
    config = HoloConfig(
        host="<HOST>",
        port=<PORT>,
        database="<DATABASE>",
        username="<USERNAME>",
        password="<PASSWORD>",
        read_parallelism=4,  # 非同步模式下啟動4個reader task並發查詢
    )

    async with AsyncHoloClient(config) as client:
        schema = await client.get_table_schema("t0")

        tasks = []
        for i in range(100):
            get = Get.builder(schema).set_primary_key("id", i).build()
            tasks.append(asyncio.create_task(client.get(get)))

        records = await asyncio.gather(*tasks)
        for record in records:
            if record:
                print(record.get_object("name0"))

asyncio.run(main())

Scan查詢

同步介面

Scan用於按主鍵首碼範圍掃描資料，可返回多行，程式碼範例如下。

from hologres import HoloClient, HoloConfig, Scan, SortKeys

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
)

with HoloClient(config) as client:
    # create table t0(id int not null, name0 text, address text, primary key(id))
    schema = client.get_table_schema("t0")

    scan = (Scan.builder(schema)
            .add_equal_filter("id", 102)
            .add_range_filter("name0", start="3", end="4")
            .with_selected_columns(["address"])
            .build())
    # 等同於 select address from t0 where id=102 and name0>='3' and name0<'4' order by id
    records = client.scan(scan)
    for record in records:
        print(record.get_object("address"))

    # 不排序
    scan = (Scan.builder(schema)
            .add_equal_filter("id", 102)
            .add_range_filter("name0", start="3", end="4")
            .with_selected_columns(["address"])
            .set_sort_keys(SortKeys.NONE)
            .build())
    records = client.scan(scan)

非同步介面

import asyncio
from hologres import AsyncHoloClient, HoloConfig, Scan

async def main():
    config = HoloConfig(
        host="<HOST>",
        port=<PORT>,
        database="<DATABASE>",
        username="<USERNAME>",
        password="<PASSWORD>",
    )

    async with AsyncHoloClient(config) as client:
        schema = await client.get_table_schema("t0")

        scan = (Scan.builder(schema)
                .add_equal_filter("id", 102)
                .add_range_filter("name0", start="3", end="4")
                .with_selected_columns(["address"])
                .build())
        records = await client.scan(scan)
        for record in records:
            print(record.get_object("address"))

asyncio.run(main())

效能比較

不同寫入和查詢方式在吞吐與延遲上存在差異，請根據業務情境選擇合適的方式。

寫入效能

寫入方式	吞吐	延遲	適用情境
Fixed Copy（同步）	最高	最低	無DELETE操作的即時寫入，追求最大吞吐。
`Put`（同步，`write_parallelism`>1）	高	低	通用寫入，支援DELETE和主鍵衝突策略。
`Put`（非同步，`write_parallelism`>1）	中	中	`asyncio`應用，受GIL限制吞吐低於同步多進程。
`Put`（同步，`write_parallelism`=1）	低	低	低吞吐情境，資源佔用最少。

說明

同步Put通過多進程和共用記憶體實現並行，不受GIL限制，提高write_parallelism可線性提升吞吐。非同步Put的生產者與flush共用同一事件迴圈，write_parallelism提升帶來的收益有限。Fixed Copy使用流式協議，繞過SQL構建和攢批，單串連即可獲得最高吞吐。寫入請求中DELETE佔比提高會降低整體的每秒寫入量。

查詢效能

查詢方式	吞吐	延遲	適用情境
`async_get`（非阻塞Future，`read_parallelism`>1）	最高	高	多線程生產者高QPS點查，追求最大吞吐。
`AsyncHoloClient.get` + `create_task`批量提交	最高	中	`asyncio`應用高QPS點查。
`get`（同步阻塞，`read_parallelism`>1）	中	低	同步應用的普通點查，延遲優先。
`Scan`	低	中	首碼範圍掃描，返回多行。

說明

async_get()返回concurrent.futures.Future，多線程可批量提交後統一等待，reader線程充分攢批，吞吐最高。AsyncHoloClient.get()配合asyncio.create_task批量提交可達同等吞吐；若逐個await則每批僅1條請求，吞吐退化。get()每次阻塞等待單條結果返回，reader線程無法充分攢批，吞吐受限於單條往返延遲。提高read_parallelism增加reader線程/task數，對批量提交模式吞吐提升顯著。Scan為單次RPC返回多行，吞吐取決於結果集大小和網路延遲。

異常處理

holo-client-py的異常分為兩類：HoloClientWithDetailsException包含寫入失敗記錄的詳情，HoloClientException通常為不可恢複的Fatal異常，程式碼範例如下。

from hologres import HoloClient, HoloConfig, Put
from hologres.exceptions import HoloClientException, HoloClientWithDetailsException

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
)

with HoloClient(config) as client:
    schema = client.get_table_schema("t0")
    put = Put(schema)
    put.set_object("id", 1)
    put.set_object("name", "name0")

    try:
        client.put(put)
        client.flush()
    except HoloClientWithDetailsException as e:
        # 包含失敗記錄詳情的異常
        for record, cause in e.details:
            print(f"寫入失敗的記錄: {record}, 原因: {cause}")
    except HoloClientException as e:
        # 非HoloClientWithDetailsException的異常一般是fatal的
        raise

自訂動作

您可以通過sql方法執行DDL等自訂SQL語句，程式碼範例如下。

from hologres import HoloClient, HoloConfig

config = HoloConfig(
    host="<HOST>",
    port=<PORT>,
    database="<DATABASE>",
    username="<USERNAME>",
    password="<PASSWORD>",
)

with HoloClient(config) as client:
    client.sql("CREATE TABLE t0(id int)")

附錄：HoloConfig參數說明

HoloConfig的完整參數說明如下。

基礎配置

參數名	預設值	說明
`host`	無	必填，Hologres執行個體地址。
`port`	無	必填，連接埠號碼。
`database`	無	必填，資料庫名。
`username`	無	必填，使用者名稱。
`password`	無	必填，密碼。
`app_name`	`holo-client-py`	串連的application_name參數。

寫入配置

參數名	預設值	說明
`write_parallelism`	`4`	處理`put`方法請求的最大並發數。同步模式下為worker進程數，非同步模式下為writer task數。
`on_conflict_action`	`INSERT_OR_REPLACE`	當INSERT目標表為有主鍵的表時，主鍵衝突的處理策略。取值如下。 `INSERT_OR_IGNORE`：主鍵衝突時不寫入。 `INSERT_OR_UPDATE`：主鍵衝突時更新相應列。 `INSERT_OR_REPLACE`：主鍵衝突時更新所有列。
`write_batch_size`	`512`	每個寫入線程的最大批次大小。
`write_batch_byte_size`	`2097152（2 MB）`	每個寫入線程的最大批次位元組大小。
`write_batch_total_byte_size`	`20971520（20 MB）`	所有表最大批次位元組大小。
`write_max_interval_ms`	`10000`	距離上次提交超過此值會觸發一次批量提交（毫秒）。
`enable_deduplication`	`true`	寫入時是否對攢批資料做去重。
`enable_generate_binlog`	`true`	關閉時，寫入的資料不會產生`binlog`。
`remove_u0000_in_text`	`true`	寫入Text列時，是否剔除字串中的`\u0000`。

查詢配置

參數名	預設值	說明
`read_parallelism`	`4`	點查並發數。同步模式下為reader線程數，非同步模式下為reader task數。
`read_batch_size`	`128`	reader每次攢批執行的最大`Get`數量。
`read_batch_queue_size`	`256`	非同步模式下的請求緩衝隊列大小。
`read_timeout_ms`	`0`	`Get`操作的逾時時間，0表示不逾時。

串連配置

參數名	預設值	說明
`retry_count`	`3`	當串連故障時，寫入和查詢的重試次數。
`retry_sleep_init_ms`	`1000`	重試等待時間基礎值（毫秒）。
`retry_sleep_step_ms`	`10000`	重試等待時間遞增步長（毫秒）。
`connection_max_idle_ms`	`60000`	資料庫連接的最大空閑時間，超過將被釋放（毫秒）。
`meta_cache_ttl_ms`	`60000`	`getTableSchema`資訊的本機快取時間（毫秒）。
`use_fixed_fe`	`false`	開啟後，`Get`/`Put`將使用Fixed FE輕量級串連。