本地AWS Glue流作业通过Spark SQL读取Glue Catalog表失败

问题描述 #

使用amazon/aws-glue-streaming-libs:glue_streaming_libs_4.0.0_image_01本地运行PySpark Glue流作业时，GlueContext API可正常从Glue Catalog表创建DataFrame，但Spark SQL API查询表内容时报错。已配置Spark会话使用Glue Catalog作为元存储，能成功列出数据库和表，但执行SELECT *语句时抛出NoSuchMethodError，该查询在AWS Glue控制台可正常运行。

glue_script.py #

import sys

from awsglue.context import GlueContext
from pyspark.sql import SparkSession
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(
    args=sys.argv,
    options=['JOB_NAME'],
)

spark = SparkSession.builder \
    .config('spark.sql.defaultCatalog', 'glue_catalog') \
    .config(f'spark.sql.catalog.glue_catalog', 'org.apache.iceberg.spark.SparkCatalog') \
    .config(f'spark.sql.catalog.glue_catalog.catalog-impl', 'org.apache.iceberg.aws.glue.GlueCatalog') \
    .config(f'spark.sql.catalog.glue_catalog.io-impl', 'org.apache.iceberg.aws.s3.S3FileIO') \
    .config(f'spark.sql.catalog.glue_catalog.warehouse', 's3://glue-iceberg-streaming-test') \
    .config('spark.sql.sources.partitionOverwriteMode', 'dynamic') \
    .config("hive.imetastoreclient.factory.class", "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory") \
    .enableHiveSupport() \
    .getOrCreate()

glue_context = GlueContext(spark.sparkContext)

spark.sql(f'SHOW DATABASES').show()  # 正常运行
spark.sql(f'SHOW TABLES in streaming_database').show()  # 正常运行

glue_context.create_data_frame_from_catalog(
    'streaming_database',
    'iceberg_table',
).show()  # 正常运行

spark.sql(f'SELECT * FROM glue_catalog.streaming_database.iceberg_table').show()  # 抛出异常

docker-compose.yml #

services:
  glue_spark_submit:
    image: amazon/aws-glue-streaming-libs:glue_streaming_libs_4.0.0_image_01
    environment:
      AWS_PROFILE: ${AWS_PROFILE}
      AWS_REGION: us-east-1
      DISABLE_SSL: true
      DATALAKE_FORMATS: iceberg
    volumes:
      - ./:/home/glue_user/workspace/
      - ~/.aws:/home/glue_user/.aws
    ports:
      - 4040:4040
      - 18080:18080
    command: >
      spark-submit
        --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
      /home/glue_user/workspace/src/glue_script.py
        --JOB_NAME=glue_script
        --TempDir=/tmp
        --job-bookmark-option=job-bookmark-disable

报错堆栈 #

Traceback (most recent call last):
glue_spark_submit-1  | Traceback (most recent call last):
glue_spark_submit-1  |   File "/home/glue_user/workspace/src/glue_script.py", line 33, in <module>
glue_spark_submit-1  |     spark.sql(f'SELECT * FROM glue_catalog.streaming_database.iceberg_table').show()
glue_spark_submit-1  |   File "/home/glue_user/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 607, in show
glue_spark_submit-1  |   File "/home/glue_user/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1321, in __call__
glue_spark_submit-1  |   File "/home/glue_user/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 190, in deco
glue_spark_submit-1  |   File "/home/glue_user/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/protocol.py", line 326, in get_return_value
glue_spark_submit-1  | py4j.protocol.Py4JJavaError: An error occurred while calling o73.showString.
glue_spark_submit-1  | : java.lang.NoSuchMethodError: software.amazon.awssdk.utils.IoUtils.closeQuietly(Ljava/lang/AutoCloseable;Lsoftware/amazon/awssdk/thirdparty/org/slf4j/Logger;)V
glue_spark_submit-1  |       at software.amazon.awssdk.protocols.xml.internal.unmarshall.AwsXmlPredicatedResponseHandler.lambda$closeInputStreamIfNeeded$0(AwsXmlPredicatedResponseHandler.java:176)
glue_spark_submit-1  |       at java.util.Optional.ifPresent(Optional.java:159)
glue_spark_submit-1  |       at software.amazon.awssdk.protocols.xml.internal.unmarshall.AwsXmlPredicatedResponseHandler.closeInputStreamIfNeeded(AwsXmlPredicatedResponseHandler.java:176)
...
glue_spark_submit-1  |       at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
glue_spark_submit-1  |       at java.lang.Thread.run(Thread.java:750)
glue_spark_submit-1  | 
glue_spark_submit-1  | 23/10/19 22:18:36 WARN S3InputStream: Unclosed input stream created by:
glue_spark_submit-1  |       org.apache.iceberg.aws.s3.S3InputStream.<init>(S3InputStream.java:72)
glue_spark_submit-1  |       org.apache.iceberg.aws.s3.S3InputFile.newStream(S3InputFile.java:83)
...
glue_spark_submit-1  |       py4j.ClientServerConnection.run(ClientServerConnection.java:106)
glue_spark_submit-1  |       at java.lang.Thread.run(Thread.java:750)
glue_spark_submit-1 exited with code 1

解决方案 #

该报错由AWS SDK版本冲突导致：Glue镜像自带的AWS SDK版本与Iceberg依赖的SDK版本不匹配，IoUtils.closeQuietly方法签名不一致。

修复步骤 #

1. 指定兼容的AWS SDK版本
   在docker-compose.yml的spark-submit命令中添加--packages参数，强制引入与Iceberg 1.3.0（Glue 4.0默认集成版本）兼容的AWS SDK依赖：

   command: >
     spark-submit
       --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
       --packages software.amazon.awssdk:aws-sdk-java:2.20.131,software.amazon.awssdk:s3:2.20.131
     /home/glue_user/workspace/src/glue_script.py
       --JOB_NAME=glue_script
       --TempDir=/tmp
       --job-bookmark-option=job-bookmark-disable
   

2. 移除冲突的Hive元存储配置
   已通过Iceberg的Glue Catalog配置对接元存储，重复配置Hive元存储会引发冲突，删除SparkSession中的对应配置项：

   spark = SparkSession.builder \
       .config('spark.sql.defaultCatalog', 'glue_catalog') \
       .config(f'spark.sql.catalog.glue_catalog', 'org.apache.iceberg.spark.SparkCatalog') \
       .config(f'spark.sql.catalog.glue_catalog.catalog-impl', 'org.apache.iceberg.aws.glue.GlueCatalog') \
       .config(f'spark.sql.catalog.glue_catalog.io-impl', 'org.apache.iceberg.aws.s3.S3FileIO') \
       .config(f'spark.sql.catalog.glue_catalog.warehouse', 's3://glue-iceberg-streaming-test') \
       .config('spark.sql.sources.partitionOverwriteMode', 'dynamic') \
       # 移除该行配置
       # .config("hive.imetastoreclient.factory.class", "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory") \
       .enableHiveSupport() \
       .getOrCreate()
   

3. 验证修复
   重新启动容器，执行Spark SQL查询即可正常读取Iceberg表数据。

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