Hologres：Hologres DataFrame

基本用法

Session管理串連池、資料庫上下文、臨時Stage或暫存資料表以及Function Compute配置。建議作為上下文管理器使用，退出時自動釋放串連並清理臨時對象。

import hologres.dataframe as hg

with hg.connect(
    host="<ENDPOINT>",
    port=80,
    dbname="my_db",
    user="<ACCESS_KEY_ID>",
    password="<ACCESS_KEY_SECRET>",
    sslmode="require",
) as session:
    session.table("public.orders").limit(10).show()

串連池參數

其他關鍵字參數會透傳給psycopg或libpq，例如sslmode、connect_timeout、keepalives和options。

切換Schema和Database

session.use_schema("analytics")
print(session.current_schema())

session.use_database("another_db")
print(session.current_database())

從Hologres表建立

orders = session.table("public.orders")

session.table()會讀取並緩衝表Schema，返回惰性DataFrame。

從SQL查詢建立

paid_orders = session.sql(
    "SELECT order_id, user_id, amount FROM public.orders WHERE status = 'paid'"
)

session.sql()返回惰性DataFrame，原生SQL在觸發Action時執行。

從Python資料建立

推薦通過Session建立，使DataFrame可以直接執行Action或寫入操作。

df = session.from_dict({"id": [1, 2], "name": ["alice", "bob"]})

df = session.from_records(
    [(1, "alice"), (2, "bob")],
    columns=["id", "name"],
)

小資料會編譯為SQL VALUES。超過inline_limit時自動轉為Arrow資料並上傳到Hologres。

從pandas DataFrame建立

df = session.from_pandas(pdf)
df.write("public.target")

資料在首次Action時上傳。Hologres V4.1及以上版本使用Internal Stage，更低版本使用暫存資料表相容。Session關閉時自動清理架構建立的Stage或暫存資料表。

從本地檔案建立

支援CSV、JSON和Parquet格式，通過pyarrow讀取。檔案讀取發生在調用read_files()時，上傳延遲到首次Action。

csv_df = session.read_files("data/*.csv")
parquet_df = session.read_files("data/orders.parquet")

csv_df.group_by("region").count().show()

列運算式

hg.col()用於引用列，hg.lit()用於建立字面量。運算式中包含Column時只構建SQL運算式，不會在Python中逐行計算。

amount = hg.col("amount")

amount * 0.06
amount.between(100, 1000)
hg.col("region").isin("east", "west")
hg.col("name").ilike("ali%")
(amount > 100) & (hg.col("status") == "paid")

投影、過濾與排序

result = (
    session.table("public.orders")
    .filter((hg.col("amount") > 100) & hg.col("region").is_not_null())
    .with_columns(
        {
            "tax": hg.col("amount") * 0.06,
            "amount_with_tax": hg.col("amount") * 1.06,
        }
    )
    .drop("internal_note")
    .rename({"region": "sales_region"})
    .select("order_id", "sales_region", "amount", "tax")
    .distinct()
    .order_by(hg.col("amount").desc())
    .limit(100)
    .offset(20)
)

分組彙總

使用group_by()和agg()進行分組彙總計算。

summary = (
    session.table("public.orders")
    .group_by("region")
    .agg(
        hg.sum("amount").alias("total_amount"),
        hg.avg("amount").alias("avg_amount"),
        hg.count("*").alias("order_count"),
    )
)

單一彙總可使用快捷方法。

session.table("public.orders").group_by("region").sum("amount")
session.table("public.orders").group_by("region").count()

對於不常用的SQL函數或特殊SQL文法，可以使用hg.function()和hg.expr()。

df.group_by("region").agg(
    hg.function("stddev", hg.col("amount")).alias("amount_stddev"),
    hg.expr("count(*) filter (where status = 'paid')").alias("paid_count"),
)

Join與Union

orders = session.table("public.orders")
users = session.table("public.users")

same_name_key = orders.join(users, on="user_id", how="left")

different_key = orders.join(
    users,
    on=orders["buyer_id"] == users["user_id"],
    how="inner",
)

all_rows = orders.union_all(session.table("public.orders_archive"))
deduplicated = orders.union(session.table("public.orders_archive"))

數組展開

session.table("public.documents").explode("tags").show()
session.table("public.documents").explode("tags", outer=True).show()

explode()編譯為Hologres支援的SELECT-list unnest()。設定outer=True會保留數組為空白或為NULL的原始行。

追加寫入

source = session.table("public.orders_staging").select("order_id", "user_id", "amount")
source.write("public.orders")

記憶體資料也使用相同介面。

session.from_records(
    [(1, "east", 100), (2, "west", 200)],
    columns=["id", "region", "amount"],
).write("public.orders")

主鍵衝突處理

通過on_conflict參數指定主鍵衝突時的處理策略。

df.write("public.target", on_conflict=("ignore", ["id"]))

df.write("public.target", on_conflict=("update", ["id"]))

df.write("public.target", on_conflict=("replace", ["id"]))

覆蓋寫入

df.write("public.target", overwrite=True)

df.write(
    "public.partitioned_target",
    overwrite=True,
    partition={"ds": "2026-01-01"},
)

Hologres V3.1及以上版本使用原生INSERT OVERWRITE，更低版本自動使用事務內TRUNCATE/DELETE + INSERT的相容方案。

讀寫OSS檔案

Hologres V4.1及以上版本可以通過EXTERNAL_FILES讀取和匯出OSS中的檔案。

讀取OSS檔案

支援CSV、Parquet和ORC格式。

oss_df = session.read_files(
    "oss://bucket/input/",
    format="parquet",
    oss_endpoint="oss-cn-hangzhou-internal.aliyuncs.com",
    role_arn="acs:ram::<ACCOUNT_ID>:role/hologres-oss-role",
)

oss_df.filter(hg.col("amount") > 100).show()

匯出查詢結果到OSS

session.table("public.orders").write_files(
    "oss://bucket/output/",
    format="csv",
    oss_endpoint="oss-cn-hangzhou-internal.aliyuncs.com",
    role_arn="acs:ram::<ACCOUNT_ID>:role/hologres-oss-role",
    target_file_size_mb=128,
)

Serverless Computing

通過serverless()方法將查詢標記為使用Serverless Computing資源執行，Serverless Computing的詳細說明請參見Serverless Computing概述。該方法為惰性標記，返回新的DataFrame，執行Action時架構在同一事務和串連上設定SET LOCAL，不會將配置泄漏到串連池中的後續查詢。

rows = (
    session.table("public.orders")
    .filter(hg.col("amount") > 100)
    .serverless(priority=4, required_cores=64, max_cores=128)
    .collect()
)

參數說明如下。

Dynamic Table

通過as_dynamic_table()將DataFrame查詢定義為Dynamic Table，Dynamic Table的詳細說明請參見Dynamic Table。該方法是立即執行的DDL Action，返回None。

summary = (
    session.table("public.orders")
    .join(session.table("public.users"), on="user_id", how="left")
    .select("order_id", "user_id", "user_name")
)

summary.as_dynamic_table(
    "public.dt_orders",
    freshness="10 minutes",
    refresh_mode="incremental",
    auto_refresh=True,
    mode="ignore",
    options={"cdc_format": "binlog"},
)

mode參數控制目標表已存在時的行為。

手動重新整理和查詢Dynamic Table。

session.refresh_table("public.dt_orders")
session.table("public.dt_orders").show()

AI Function

hg.ai將Hologres AI Function暴露為惰性Column運算式，可以在DataFrame查詢中直接使用AI能力。

questions = session.table("public.questions")
questions.select(
    "question",
    hg.ai.gen(
        hg.function("concat", hg.lit("請簡要回答："), hg.col("question"))
    ).alias("answer"),
).show()

session.table("public.docs").select(
    "id",
    hg.ai.embed(hg.col("content")).alias("embedding"),
).show()

session.table("public.comments").select(
    "comment",
    hg.ai.sentiment(hg.col("comment")).alias("sentiment"),
).show()

除上述樣本外，還支援rank、chunk、classify、extract、mask、fix_grammar、summarize、translate、similarity、parse_document、to_file和prompt。具體函數的可用性取決於Hologres執行個體版本和模型配置，支援的函數列表請參見AI Function列表。

Function Compute UDF與UDTF

Hologres沒有內嵌Python Runtime。@hg.udf和@hg.udtf會在首次使用它們的Action執行前，將Python代碼自動部署到Function Compute，再註冊為LANGUAGE function_compute的Remote UDX。

配置Function Compute

Function Compute配置放在Session中，AccessKey使用阿里雲標準憑據鏈，不要寫在裝飾器中。

session = hg.connect(
    host="<ENDPOINT>",
    port=80,
    dbname="my_db",
    user="<ACCESS_KEY_ID>",
    password="<ACCESS_KEY_SECRET>",
    fc_config={
        "endpoint": "<ACCOUNT_ID>.cn-hangzhou-internal.fc.aliyuncs.com",
        "region": "cn-hangzhou",
        "role_arn": "acs:ram::<ACCOUNT_ID>:role/fc-execution-role",
    },
)

標量UDF

@hg.udf(packages=["numpy"])
def grade(score: int) -> str:
    return "high" if score >= 80 else "low"

session.table("public.scores").select(
    "id", grade("score").alias("grade")
).show()

UDTF

UDTF使用類和process()方法定義。

@hg.udtf(output_schema=hg.StructType([hg.StructField("word", hg.StringType())]))
class SplitWords:
    def process(self, text: str):
        for word in text.split():
            yield (word,)

session.table("public.docs").select("id", SplitWords("text")).show()