如何在AWS EMR上构建Iceberg数据湖

测试环境

• AWS EMR : 6.2.0
• Iceberg : 0.11.1
• EMR Spark: 3.0.1-amzn-0

注意：如果使用 AWS EMR 6.3.0 版本，使用 EMR 默认的 iceberg 0.11.1 版本则可能碰到这个 issue。这时，应该手动把如下iceberg jar删除掉，然后指定用master分支的iceberg runtime jar来运行spark-sql:

[hadoop@ip-172-31-33-131 ~]$ locate iceberg
/emr/instance-controller/lib/bootstrap-actions/1/bootstrap-with-iceberg.sh
/usr/share/aws/aws-java-sdk/iceberg-flink-runtime-0.11.1.jar
/usr/share/aws/aws-java-sdk/iceberg-spark3-runtime-0.11.1.jar

运行 spark-sql 的方式：

spark-sql --jars /path/to/iceberg-spark3-runtime-25eaeba.jar \ 
    --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
    --conf spark.sql.catalog.hive_prod=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.hive_prod.type=hive \
    --conf spark.sql.catalog.hive_prod.uri=thrift://ip-172-31-34-107.ap-northeast-1.compute.internal:9083 \
    --conf spark.sql.catalog.hive_prod.warehouse=s3://dw-ali/warehouse

第一步：打开spark-sql

在启动spark-sql时指定如下参数：

spark-sql --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
    --conf spark.sql.catalog.hive_prod=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.hive_prod.type=hive \
    --conf spark.sql.catalog.hive_prod.uri=thrift://ip-172-31-34-107.ap-northeast-1.compute.internal:9083 \
    --conf spark.sql.catalog.hive_prod.warehouse=s3://dw-ali/warehouse

• hive_prod 表示你想要创建的catalog名字
• spark.sql.catalog.hive_prod.type: 表示创建的catalog类型，目前iceberg支持：hive、hadoop、custom三种。
• spark.sql.catalog.hive_prod.uri : 表示hive-metastore的thrift地址。对于AWS EMR集群来说，可以在 /usr/lib/hive/conf/hive-site.xml 文件中的搜索 ‘hive.metastore.uris’ ，获取到对应的value。
• spark.sql.catalog.hive_prod.warehouse: 表示 iceberg 的数据存在在哪个目录下。

第二步：创建对应的 database

执行如下 SQL:

CREATE DATABASE hive_prod.iceberg_db

这时候，可以在 s3 的路径下看到具体的文件：

[hadoop@ip-172-31-33-131 ~]$ hdfs dfs -ls s3://dw-ali/warehouse
Found 1 items
drwxrwxrwx   - hadoop hadoop          0 1970-01-01 00:00 s3://dw-ali/warehouse/iceberg_db.db
[hadoop@ip-172-31-33-131 ~]$ hdfs dfs -ls -R s3://dw-ali/warehouse
drwxrwxrwx   - hadoop hadoop          0 1970-01-01 00:00 s3://dw-ali/warehouse/iceberg_db.db
drwxrwxrwx   - hadoop hadoop          0 1970-01-01 00:00 s3://dw-ali/warehouse/iceberg_db.db/sample
drwxrwxrwx   - hadoop hadoop          0 1970-01-01 00:00 s3://dw-ali/warehouse/iceberg_db.db/sample/data
-rw-rw-rw-   1 hadoop hadoop        636 2021-07-12 11:37 s3://dw-ali/warehouse/iceberg_db.db/sample/data/00000-0-7072f1c0-330c-4eaf-9096-47fd5b4e3cf6-00001.parquet
drwxrwxrwx   - hadoop hadoop          0 1970-01-01 00:00 s3://dw-ali/warehouse/iceberg_db.db/sample/metadata
-rw-rw-rw-   1 hadoop hadoop        825 2021-07-12 11:37 s3://dw-ali/warehouse/iceberg_db.db/sample/metadata/00000-1a9e72e3-8549-4713-8e3b-22a1b76bd49e.metadata.json
-rw-rw-rw-   1 hadoop hadoop       1739 2021-07-12 11:37 s3://dw-ali/warehouse/iceberg_db.db/sample/metadata/00001-06443d34-8ada-40b1-8d3d-2c8381073ed2.metadata.json
-rw-rw-rw-   1 hadoop hadoop       5570 2021-07-12 11:37 s3://dw-ali/warehouse/iceberg_db.db/sample/metadata/830162e3-ed03-42f0-8ad5-ffebdb44706b-m0.avro
-rw-rw-rw-   1 hadoop hadoop       3500 2021-07-12 11:37 s3://dw-ali/warehouse/iceberg_db.db/sample/metadata/snap-929163357221233223-1-830162e3-ed03-42f0-8ad5-ffebdb44706b.avro

第三步：创建测试用的 iceberg 表

执行如下 SQL:

CREATE TABLE hive_prod.iceberg_db.sample 
    (
    id bigint,
    data string
  ) 
USING iceberg;

第四步：插入测试数据

执行如下 SQL:

INSERT INTO hive_prod.iceberg_db.sample VALUES (1, 'aaa');

第五步：查询测试数据

执行如下 SQL:

SELECT * FROM hive_prod.iceberg_db.sample;