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()