如何在K8s上让Spark对接Unity Catalog并关联MinIO存储数据？

问题排查与修复步骤 #

1. 验证权限 #

你能查看schema和现有表，不代表拥有创建新表的权限。先执行以下SQL确认权限：

SHOW GRANTS ON SCHEMA my_catalog.catalog_sales_schema;

如果没有CREATE TABLE权限，在Unity Catalog中给当前用户/角色授权：

GRANT CREATE TABLE ON SCHEMA my_catalog.catalog_sales_schema TO `你的用户名或角色名`;

2. 补充MinIO访问配置 #

Spark会话缺少MinIO的访问参数，会导致无法定位存储路径，间接触发schema找不到的误报。在Spark初始化时添加以下MinIO相关配置（替换为你的实际信息）：

spark = SparkSession.builder \
    .appName("UnityCatalogTest") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .config("spark.sql.catalog.my_catalog", "io.unitycatalog.spark.UCSingleCatalog") \
    .config("spark.sql.catalog.my_catalog.uri", unity_catalog_endpoint) \
    .config("spark.sql.catalog.my_catalog.token", unity_catalog_token) \
    .config("spark.sql.defaultCatalog", "my_catalog") \
    # MinIO核心配置
    .config("spark.hadoop.fs.s3a.endpoint", "http://你的MinIO服务地址:9000")
    .config("spark.hadoop.fs.s3a.access.key", "你的MinIO访问密钥")
    .config("spark.hadoop.fs.s3a.secret.key", "你的MinIO秘密密钥")
    .config("spark.hadoop.fs.s3a.path.style.access", "true")
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .getOrCreate()

3. 优化SQL语句的严谨性 #

• 先确保schema存在，避免创建表时的依赖问题：

  CREATE SCHEMA IF NOT EXISTS my_catalog.catalog_sales_schema;
  

• 切换到目标schema后再创建表，减少全限定名的潜在问题：

  USE my_catalog.catalog_sales_schema;
  
  CREATE TABLE IF NOT EXISTS catalog_sales (
      cs_sold_date_sk INTEGER,
      cs_sold_time_sk INTEGER,
      cs_ship_date_sk INTEGER,
      cs_bill_customer_sk INTEGER,
      cs_bill_cdemo_sk INTEGER,
      cs_bill_hdemo_sk INTEGER,
      cs_bill_addr_sk INTEGER,
      cs_ship_customer_sk INTEGER,
      cs_ship_cdemo_sk INTEGER,
      cs_ship_hdemo_sk INTEGER,
      cs_ship_addr_sk INTEGER,
      cs_call_center_sk INTEGER,
      cs_catalog_page_sk INTEGER,
      cs_ship_mode_sk INTEGER,
      cs_warehouse_sk INTEGER,
      cs_item_sk INTEGER,
      cs_promo_sk INTEGER,
      cs_order_number LONG,
      cs_quantity INTEGER,
      cs_wholesale_cost DECIMAL(7,2),
      cs_list_price DECIMAL(7,2),
      cs_sales_price DECIMAL(7,2)
  )
  USING DELTA
  LOCATION 's3a://test/tpcds_10_delta/catalog_sales';
  

正确的注册表方式 #

方式1：通过saveAsTable关联现有Delta数据 #

如果MinIO中已存在Delta格式的数据，用以下代码注册外部表：

df = spark.read.format("delta").load("s3a://test/tpcds_10_delta/catalog_sales")
df.write.format("delta") \
    .option("path", "s3a://test/tpcds_10_delta/catalog_sales")  # 必须显式指定MinIO路径
    .partitionBy("cs_bill_customer_sk") \
    .mode("overwrite")  # 表已存在时可替换为ignore避免报错
    .saveAsTable("my_catalog.catalog_sales_schema.catalog_sales_2")

方式2：通过CREATE TABLE直接创建外部表 #

确保MinIO路径可访问、schema权限正常的前提下，执行前文优化后的CREATE TABLE语句即可。

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