如何使用Python SDK从Kusto查询数据时实现分页功能？

可以实现，主要有两种可靠的分页方式，分别基于Kusto查询语法和SDK原生的游标机制：

方法1：通过Kusto查询语法实现分页 #

Kusto原生支持用limit和offset运算符实现分页（如果数据有序，也可以用更高效的range运算符）。你只需在原查询语句后动态拼接分页参数即可：

// 示例分页查询
YourTargetTable
| order by Timestamp desc  // 建议先排序保证分页一致性
| limit {page_size}
| offset {offset}

这种方式简单直接，适合一次性获取指定页码的数据。

方法2：通过SDK的ContinuationToken实现游标分页 #

Kusto Python SDK支持通过结果集的continuation_token来分批获取数据，无需修改原查询，SDK会自动跟踪分页位置，适合大数据量场景（避免一次性加载所有数据到内存）。

整合分页功能的完整代码 #

下面是修改后的代码，同时支持两种分页方式，且复用了你原有的认证逻辑：

from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.kusto.data.helpers import dataframe_from_result_table
import pandas as pd

def execute_kusto_query(
    authentication, 
    cluster: str, 
    database: str, 
    query: str, 
    tenant_id: str, 
    client_id: str = None, 
    client_secret: str = None, 
    page_size: int = 10,
    offset: int = 0,
    use_cursor_pagination: bool = False
):
    try:
        # 构建连接字符串
        if authentication == 'client_secret':
            connection_string = KustoConnectionStringBuilder.with_aad_application_key_authentication(
                cluster,
                client_id,
                client_secret,
                authority_id=f"{tenant_id}"
            )
        elif authentication == 'managed_identity':
            connection_string = KustoConnectionStringBuilder.with_aad_managed_service_identity_authentication(cluster)                    
        else:
            raise ValueError("Invalid authentication method.")
        
        with KustoClient(connection_string) as client:
            if use_cursor_pagination:
                # 游标分页：循环获取数据直到无更多结果
                all_pages = []
                continuation_token = None
                while True:
                    response = client.execute_query(database, query, continuation_token=continuation_token)
                    table = response.primary_results[0]
                    if table.row_count == 0:
                        break
                    all_pages.append(dataframe_from_result_table(table))
                    continuation_token = response.continuation_token
                    # 仅获取指定页数（按需取消注释）
                    # if len(all_pages) >= page_count:
                    #     break
                return pd.concat(all_pages, ignore_index=True) if all_pages else pd.DataFrame()
            else:
                # 查询层面分页：拼接分页语句
                paginated_query = f"{query} | limit {page_size} | offset {offset}"
                response = client.execute_query(database, paginated_query)
                return dataframe_from_result_table(response.primary_results[0])

    except Exception as ex:
        print(f"Error executing Kusto query: {str(ex)}")

代码说明 #

• 查询层面分页：通过offset（跳过的行数）和page_size（每页行数）控制返回数据，需保证查询有稳定排序（否则分页结果可能混乱）。
• 游标分页：无需修改原查询，SDK通过continuation_token自动记录读取位置，循环获取下一页，直到结果集为空，适合处理百万级以上数据。
• 原代码中未使用的page_size参数已被整合到两种分页逻辑中。

内容的提问来源于stack exchange，提问作者manav sharma