高性能写入 apache Iceberg

Question

我正在尝试直接在 Python 环境中实现从 pandas（或者理想情况下是 Polars，如果可能）到我们的 Apache Iceberg 部署（使用 hive 元存储）的高性能记录写入，或者通过基于性能最高的 Trino 查询引擎进行记录写入。

鉴于我已经尝试过的方法，我当前的记录写入吞吐量仍然相当垃圾。如果有人可以请我指出如何直接与 Iceberg 建立高性能写入连接的正确方向（或者如果期望通过查询引擎工作，则通过 Trino）...因为我也没有看到指导文档。

我的尝试如下：

• 最初通过Trino python 包将 pandas df 的内容解析为批量的 sql 语法查询。- 笨重且最慢
• 然后挖掘并找到了一个异步 Trino python 包，它至少允许我使用 asyncio 进行调用，并从基于部署的 trino 工作人员数量的调用并发中受益。
• 利用 Trino 的 jdbc 连接器类型，以及底层的 trino-jdbc.jar 和 python 的jaydebeapi 包，看看底层 java 运行时执行是否可以提高性能 - 数据读取成功，但写入不成功
• Iceberg 通过pySpark直接连接- 本文中描述的连接问题
• 从 0.4.0 版开始，pyIceberg 仍然不包含我从文档或挖掘包中收集到的记录写入功能
• 使用pandas.to_sql()方法偶然发现了这种方法 - 迄今为止最好的方法，但仍然不够

通过一个通信线程，我还了解了如何将数据帧作为镶木地板文件格式直接上传到 S3，然后从那里可以通过元数据调用与 Iceberg 的相关目录表链接（明智的压缩传输是有意义的）...但也有避风港无法以这种方式开始工作

Trino 的 pandas.sql() 代码片段是唯一值得分享的：

import warnings
# Ignore all warnings
warnings.filterwarnings("ignore")
import logging
logging.getLogger().setLevel(logging.DEBUG)

from sqlalchemy import create_engine
from trino.auth import BasicAuthentication
import pandas as pd 
from datetime import datetime

trino_target_host = "..." 
trino_target_port = 443
trino_target_catalog = 'iceberg'
trino_target_schema = '...'

def write_content(table: str, df, batch_size: int):
    print(f"size: {len(df)}")
    try:
        start = datetime.utcnow()
        engine = create_engine(
            f"trino://{trino_target_host}:{trino_target_port}/{trino_target_catalog}/{trino_target_schema}",
            connect_args={
                "auth": BasicAuthentication("...", "..."),
                "http_scheme": "https"
            }
        )
        # chunksize = writing df in batches of size - saving memory
        # method = instead of writing a single record at a time, 'multi' will insert multiple rows as 1 statement
        output = df.to_sql(table, engine, if_exists='append', index=False, chunksize=batch_size, method='multi')
        print(f"elapsed: {datetime.utcnow() - start}")
        return output
    except Exception as e:
        if 'Request Entity Too Large' in str(e):
            print(f"Batch size '{batch_size}' too large. Reduce size")
            return None
        print(f"failed: {e}")
        return None  

sample = df[:70000].copy(deep=True)
write_content('table_name', sample, 1000)

规模：70000

立即执行不适用于 trino.dp.iotnxt.io:443；默认为旧的准备好的语句（TrinoUserError（type = USER_ERROR，name = SYNTAX_ERROR，message =“line 1:19：不匹配的输入''SELECT 1''。期望：'USING'，”，query_id = 20230629_161633_03350_rptf8））

已过去：0:14:05.315631

这是非常特定于部署的，但只是为了提供一个相对的比较想法：

数数	读	写入（批量 = 1000）	写入（批量 = 1500）
10		3.67 / 3.7 / 3.6 秒
100	2.9 / 4.0 / 2.8 / 2.2 秒	4.4 / 4.2 / 4.1 秒
1000	2.2 / 3.8 / 2.4 秒	11.2 / 11.2 / 11.0 秒
10 000	3.2 / 2.8 / 2.9 秒	118.4 / 105.7 / 108.8 秒	98.2 / 104.6 / 108.5 秒
70 000	6.8 / 6.5 / 7.2 秒	845.31 秒
140 000	9.8秒