标签: apache-iceberg

我正在尝试使用 Apache Iceberg 将数据写入指定位置（S3/本地）。以下是下面使用的配置。

科学技术试验：

libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.2.1" % "provided",
libraryDependencies += "org.apache.iceberg" % "iceberg-spark-runtime-3.2_2.12" % "0.13.2"

火花：

val builder = SparkSession
      .builder()
      .config("spark.sql.extensions","org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
      .config("spark.sql.catalog.spark_catalog","org.apache.iceberg.spark.SparkSessionCatalog")
      .config("spark.sql.catalog.spark_catalog.type","hive")
      .config("spark.sql.catalog.local","org.apache.iceberg.spark.SparkCatalog")
      .config("spark.sql.catalog.local.type","hadoop")
      .config("spark.sql.catalog.local.warehouse","/Users/tom/Documents/hive/warehouse")
      .getOrCreate()

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions._

// Create a DataFrame
val data = Seq(
("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
("101", "2015-01-01", "2015-01-01T12:14:58.597216Z"),
("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
("103", "2015-01-01", "2015-01-01T13:51:40.519832Z")
).toDF("id", "creation_date", "last_update_time")

data.write
.format("iceberg")
.save("/Users/tom/Documents/data") 

当我运行上面的代码时，我遇到了缺少 HiveCatalog 依赖项的情况。不知道为什么，但iceberg-spark-runtime包含所有必需的 JAR。我也尝试使用iceberg-hive-runtime但仍然面临下面相同的错误。

Cannot initialize Catalog implementation org.apache.iceberg.hive.HiveCatalog: Cannot find constructor …

我正在尝试直接在 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
# …

我正在尝试使用 Apache Iceberg 目录和 Hive Metastore 设置 Flink SQL，但没有成功。以下是我在干净的 Flink 1.18.1 安装中采取的步骤，以及我得到的错误。

设置组件

运行 Hive MetaStore：

docker run --rm --detach --name hms-standalone \\\n           --publish 9083:9083 \\\n           ghcr.io/recap-build/hive-metastore-standalone:latest \n

使用 Docker 运行 MinIO：

docker run --rm --detach --name minio \\\n            -p 9001:9001 -p 9000:9000 \\\n            -e "MINIO_ROOT_USER=admin" \\\n            -e "MINIO_ROOT_PASSWORD=password" \\\n            minio/minio server /data --console-address ":9001"\n

配置一个存储桶：

docker exec minio \\\n    mc config host add minio http://localhost:9000 admin password\ndocker exec minio \\\n    mc mb …

我正在尝试在云对象存储上创建冰山表格式。

在下图中我们可以看到冰山表格式需要一个目录。该目录存储当前元数据指针，该指针指向最新的元数据。Iceberg 快速入门文档列出了 JDBC、Hive MetaStore、AWS Glue、Nessie 和 HDFS 作为可以使用的目录列表。

我的目标是将当前元数据指针（version-hint.text）以及表数据的其余部分（元数据、清单列表、清单、镶木地板数据文件）存储在对象存储本身中。

\n

以HDFS为目录，table\xe2\x80\x99s元数据文件夹中有一个名为version-hint.text的文件，其内容为当前元数据文件的版本号。

\n

将 HDFS 视为可能的目录之一，我应该能够使用 ADLS 或 S3 来存储当前元数据指针以及其余数据。例如：spark 使用 ABFSS 接口连接到 ADLS 并创建冰山表和目录。

我的问题是

\n
• 使用版本提示文件作为 ADLS/S3 中的当前元数据指针是否安全？如果我这样做，我会失去任何冰山功能吗？查看一位贡献者的评论表明它对于生产来说并不理想。
\n

\n

版本提示文件用于 Hadoop 表，之所以这样命名，是因为它们适用于 HDFS。我们还将它们用于本地 FS\n测试，但它们不能与 S3 安全地同时使用。对于 S3，您需要一个元存储来在交换表元数据位置时强制执行原子性。您可以使用iceberg-hive 中的\nHive 元存储。

\n

\n
• 看看这个线程的评论，version-hint.text 文件是可选的吗？
\n

\n

我们遍历可能的元数据位置，仅当\n没有新快照可用时才停止

\n

有人可以澄清一下吗？

我正在尝试使用 Iceberg 进行 POC。此时的要求是能够至少每10分钟将新数据从数据块写入到表中。这个频率将来可能会增加。

数据一旦写入，将被databricks和dremio读取。