欢迎关注大数据技术架构与案例微信公众号:过往记忆大数据
过往记忆博客公众号iteblog_hadoop
欢迎关注微信公众号:
过往记忆大数据
大数据技术博客公众号bigdata_ai
开发爱好者社区:
Java技术范

Apache Iceberg 快速入门

导言

本文主要介绍如何快速的通过Spark访问 Iceberg table。

Apache Iceberg 快速入门
如果想及时了解Spark、Hadoop或者HBase相关的文章,欢迎关注微信公众号:iteblog_hadoop

Spark通过DataSource和DataFrame API访问Iceberg table,或者进行Catalog相关的操作。由于Spark Data Source V2 API还在持续的演进和修改中,所以Iceberg在不同的Spark版本中的使用方式有所不同。

版本对比

功能Spark 2.4Spark 3.0
基于DataFrame
- 读数据支持支持
- 读元数据支持支持
- 追加(append)支持支持
- 覆盖(Overwrite)支持支持
- V2 source专属操作,如create, overwrite不支持支持
基于Spark SQL
- SELECT通过DataFrame的temporary view支持
- DDL不支持(仅能通过Iceberg API)支持(通过Catalog)
- DML不支持支持

Spark 2.4 环境的使用

配置

Hive MetaStore

Iceberg 内部支持 Hive 和 Hadoop 两种 catalog:

Catalog类型Metadata JSON管理Namespace
Hive catalogHive MetaStore1级,即DB
Hadoop catalog文件系统上的某个文件多级,对应多级目录

后文以Hive catalog为主做介绍。Hive catalog需要Hive MetaStore的支持。注意其有多种配置方式,其中内嵌的Derby数据库仅仅用于实验和学习,不能用于生产环境。

Spark

<SPARK_HOME>/conf/spark-defaults.conf需要加入如下配置,使Iceberg能够访问Hive MetaStore:

spark.hadoop.hive.metastore.uris           thrift://<HiveMetaStore>:9083
spark.hadoop.hive.metastore.warehouse.dir  hdfs://<NameNode>:8020/path

部署

如何使用社区正式发布的版本:

spark-shell --packages org.apache.iceberg:iceberg-spark-runtime:0.7.0-incubating

如何本地打包,并把Iceberg放入Spark的classpath:

git clone https://github.com/apache/incubator-iceberg.git
cd incubator-iceberg
# master branch supports Spark 2.4.4
./gradlew assemble
spark-shell --jars <iceberg-git-working-directory>/spark-runtime/build/libs/iceberg-spark-runtime-<version>.jar

读Iceberg table

通过DataFrame

Spark 2.4只能读写已经存在的Iceberg table。在后续的操作前,需要先通过Iceberg API来创建table。

读取是通过DataFrameReader并指定iceberg作为format来访问Iceberg table,随后Iceberg内部的逻辑会根据path来判断访问的是Hive catalog下的table,还是用文件系统的路径表示的Hadoop table。

// Table managed by Hive catalog
spark.read
     .format("iceberg")
     .load("db.table")

// Hadoop table, identified by a path 
spark.read
     .format("iceberg")
     .load("hdfs://<NameNode>:8020/<path_to_table>")

Iceberg会判断path中是否含有"/"。如果是,则认为是一个用路径表示Hadoop table;否则,会去Hive catalog中寻找。

利用time travel回溯某一个snapshot的数据

在读取时,通过option指定as-of-timestamp或者snapshot-id来访问之前某一个snapshot中的数据:

// Time travel to October 26, 1986 at 01:21:00
spark.read
     .format("iceberg")
     .option("as-of-timestamp", "499162860000")
     .load("db.table")

// Time travel to snapshot with ID 10963874102873L
spark.read
     .format("iceberg")
     .option("snapshot-id", 10963874102873L)
     .load("db.table")

snapshot-id的获取方法,可以参考后文中访问元数据中snapshot的部分,或者直接查看元数据文件的内容。

在DataFrame基础上使用SQL SELECT
在DataFrame的基础上,创建local temporary view后,也可以通过SQL SELECT来读取Iceberg table的内容:

val df = spark.read
              .format("iceberg")
              .load("db.table")
df.createOrReplaceTempView("view")
spark.sql("""SELECT * FROM view""")
     .show()

写Iceberg table

Spark 2.4可以通过DataFrameWriter并指定iceberg作为format来写入Iceberg table,并支持append和overwrite两种模式:

// Append
df.write
  .format("iceberg")
  .mode("append")
  .save("db.table")

// Overwrite
df.write
  .format("iceberg")
  .mode("overwrite")
  .save("db.table")

有如下几点需要注意:

  1. Overwrit的行为dynamic overwrite,即当某个partition中含有输入DataFrame中的行的时候,该partition才会被新数据完全覆盖;其他partition则保持不变。而Spark 2.4中原生数据源(如parquet)的默认行为是static overwrite;
  2. 操作粒度是文件级别,并不是行级别;
  3. mode必须显式指定,没有默认行为。

访问Iceberg table的元数据

Iceberg支持通过DataFrameReader访问table的元数据,如snapshot,manifest等。对于Hive table,可以在原table name后面加.history、.snapshots等表示要访问元数据;对于用路径来表示的Hadoop table,需要在原路径后面加#history等。例如:

// Read snapshot history of db.table
spark.read
     .format("iceberg")
     .load("db.table.history")

结果

+-------------------------+---------------------+---------------------+---------------------+
| made_current_at         | snapshot_id         | parent_id           | is_current_ancestor |
+-------------------------+---------------------+---------------------+---------------------+
| 2019-02-08 03:29:51.215 | 5781947118336215154 | NULL                | true                |
| 2019-02-08 03:47:55.948 | 5179299526185056830 | 5781947118336215154 | true                |
| 2019-02-09 16:24:30.13  | 296410040247533544  | 5179299526185056830 | false               |
| 2019-02-09 16:32:47.336 | 2999875608062437330 | 5179299526185056830 | true                |
| 2019-02-09 19:42:03.919 | 8924558786060583479 | 2999875608062437330 | true                |
| 2019-02-09 19:49:16.343 | 6536733823181975045 | 8924558786060583479 | true                |
+-------------------------+---------------------+---------------------+---------------------+

又如:

// Read snapshot list of db.table
spark.read
     .format("iceberg")
     .load("db.table.snapshots")

// Read manifest files of db.table
spark.read
     .format("iceberg")
     .load("db.table.manifests")

// Read data file list of db.tabe
spark.read
     .format("iceberg")
     .load("db.table.files")

可以进一步将history和snapshot按照snapshot id做join,来查找snapshot id对应的application id:

spark.read
     .format("iceberg")
     .load("db.table.history")
     .createOrReplaceTempView("history")
spark.read
     .format("iceberg")
     .load("db.table.snapshots")
     .createOrReplaceTempView("snapshots")
SELECT
    h.made_current_at,
    s.operation,
    h.snapshot_id,
    h.is_current_ancestor,
    s.summary['spark.app.id']
FROM history h
JOIN snapshots s
  ON h.snapshot_id = s.snapshot_id
ORDER BY made_current_at

结果如下:

-------------------------+-----------+----------------+---------------------+----------------------------------+
| made_current_at         | operation | snapshot_id    | is_current_ancestor | summary[spark.app.id]            |
+-------------------------+-----------+----------------+---------------------+----------------------------------+
| 2019-02-08 03:29:51.215 | append    | 57897183625154 | true                | application_1520379288616_155055 |
| 2019-02-09 16:24:30.13  | delete    | 29641004024753 | false               | application_1520379288616_151109 |
| 2019-02-09 16:32:47.336 | append    | 57897183625154 | true                | application_1520379288616_155055 |
| 2019-02-08 03:47:55.948 | overwrite | 51792995261850 | true                | application_1520379288616_152431 |
+-------------------------+-----------+----------------+---------------------+----------------------------------+

Spark 3.0 环境的使用

Iceberg在Spark 3.0中,作为V2 Data Source,除了上述Spark 2.4所有的访问能力外,还可以通过V2 Data Source专属的DataFrame API访问;同时,受益于external catalog的支持,Spark SQL的DDL功能也可以操作Iceberg table,并且DML语句支持也更加丰富。

配置external catalog

在<SPARK_HOME>/conf/spark-defaults.conf加入如下配置:

spark.sql.catalog.catalog-name=com.example.YourCatalogClass

通过V2 Data Source专属DataFrame API访问

df.writeTo("catalog-name.db.table")
  .overwritePartitions()

通过Spark SQL访问

相较于Spark 2.4,Spark 3.0可以省去DataFrameReader和创建local temporary view的步骤,直接通过Spark SQL进行操作:

-- Create table
CREATE TABLE catalog-name.db.tabe
    (id INT, data STRING)
    USING iceberg
    PARTITIONED BY (id)

-- Insert
INSERT INTO catalog-name.db.table
    VALUES (1, 'a'), (2, 'b'), (3, 'c')

-- Delete
DELETE FROM catalog-name.db.table
    WHERE id <> 1

-- Update
UPDATE catalog-name.db.table
    SET data = 'C' WHERE id = 3

-- Create table as select
CREATE TABLE catalog-name.db.table 
    USING iceberg
    AS SELECT id, data 
       FROM catalog-name.db.table1 
       WHERE id <= 2

我们作为社区中spark-3分支的维护者,正在持续推进新功能的开发和合入,让更多的人受益。

总结

本文作为Iceberg的快速入门,介绍了如何通过Spark访问Iceberg table,以及不同Spark版本的支持情况:

  • Spark 2.4可以通过DataFrame读取或修改已经存在的Iceberg table中的数据,但建表、删表等DDL操作只能通过Iceberg API完成;
  • Spark 3.0访问Iceberg table的能力是Spark 2.4的超集,可以通过Spark SQL配合catalog,进行SELECT、DDL和DML等更多的操作。

随着Iceberg自身功能的完善(如向量化读取,merge on read等),以及上下游对接和生态的丰富,Iceberg作为优秀的表格式抽象,在大数据领域必然会有更好的发展。

本文原文:https://mp.weixin.qq.com/s/vvsnHrbzxJ3Gno1XtzHO7g

本博客文章除特别声明,全部都是原创!
转载本文请加上:转载自过往记忆(https://www.iteblog.com/)
本文链接: 【Apache Iceberg 快速入门】(https://www.iteblog.com/archives/9824.html)
喜欢 (1)
分享 (0)
发表我的评论
取消评论

表情
本博客评论系统带有自动识别垃圾评论功能,请写一些有意义的评论,谢谢!