如何通过EMR spark-submit步骤连接Glue数据目录？

我们在EMR集群中使用PySpark查询Glue数据库，采用Zeppelin笔记本和EMR步骤两种方式执行Python脚本。在Zeppelin中连接Glue数据库一切正常，但在EMR步骤中执行查询时出现如下错误：

pyspark.sql.utils.AnalysisException: Database '{glue_database_name}' does not exist.

执行的PySpark脚本配置和boto3提交步骤的代码如描述所示，且两种方式使用的EC2用户均已授予Glue和S3权限。以下是具体解决方法：

可能的解决方法 #

1. 确认EMR集群已启用Glue Data Catalog作为Hive元存储 #

• EMR集群创建时，需在元存储选项中选择AWS Glue Data Catalog，否则集群默认使用本地Hive元存储，无法识别Glue中的数据库。
• 登录EMR主节点，检查/etc/hive/conf/hive-site.xml文件，确认hive.metastore.client.factory.class的值为com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory。若不一致，需修改配置或重新创建集群时启用Glue Catalog。

2. 在spark-submit命令中显式指定Glue Catalog配置 #

代码中设置的SparkSession配置可能被spark-submit的全局配置覆盖，需在提交参数中添加Glue相关配置：
修改boto3提交代码中的Args部分：

response = emr_client.add_job_flow_steps(
    JobFlowId=cluster_id,
    Steps=[
        {   'Name': name,
            'ActionOnFailure': 'CONTINUE',
            'HadoopJarStep': {
                'Jar': 'command-runner.jar',
                'Args': [
                    'spark-submit',
                    '--deploy-mode', 'client',
                    '--master', 'yarn',
                    '--conf', 'hive.metastore.client.factory.class=com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory',
                    '--conf', 'spark.sql.catalogImplementation=hive',
                    "{path to .py script}"
                ]
            }
        },
    ]
)

3. 验证Glue数据库名称的准确性 #

• 检查代码中{glue_database_name}是否与Glue控制台中的数据库名称完全一致（Glue数据库名称区分大小写）。
• 在EMR主节点执行命令验证：

  aws glue get-database --name {glue_database_name}
  

  若命令返回数据库信息，说明名称正确；若报错，需修正代码中的数据库名称。

4. 检查EMR步骤的执行角色权限 #

EMR步骤的执行角色（非EC2实例角色）需具备Glue访问权限，确保角色包含以下权限：

• glue:GetDatabase
• glue:GetTable
• glue:GetPartition（若查询分区表）
  可直接附加GlueDataCatalogFullAccess策略进行测试。

5. 避免SparkSession配置冲突 #

移除代码中手动设置的hive.metastore.client.factory.class配置，让SparkSession继承EMR集群的默认Glue Catalog配置，修改后的SparkSession构建代码：

from pyspark import SparkContext
from pyspark.sql import SparkSession

sc = SparkContext.getOrCreate()
spark = SparkSession(sc)\
    .builder\
    .appName("dfp.sln.kunderelation.work")\
    .config("spark.sql.broadcastTimeout", "36000")\
    .config("spark.sql.legacy.parquet.int96RebaseModeInRead", "CORRECTED")\
    .config("spark.sql.legacy.parquet.int96RebaseModeInWrite", "CORRECTED")\
    .config("spark.sql.legacy.parquet.datetimeRebaseModeInRead", "CORRECTED")\
    .enableHiveSupport()\
    .getOrCreate()
spark.conf.set("spark.sql.sources.ignoreDataLocality.enabled", "true")

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