Aplicar políticas de proteção de dados ao consultar tabelas Apache Iceberg™ pelo Apache Spark™¶

Baixar o conector Snowflake para Spark¶

Para baixar a versão 3.1.6 ou uma versão posterior do conector Snowflake para Spark, siga as instruções em Instalação e configuração do conector Spark.

Conectar o Spark a tabelas Iceberg¶

Nesta etapa, você conecta o Spark às tabelas Iceberg por meio do Horizon Catalog. Essa conexão inclui configurações para você usar o conector Snowflake para Spark com o Horizon Catalog para consultar as tabelas protegidas pelas políticas de proteção de dados do Snowflake.

• Para conectar o Spark às tabelas Iceberg usando um token de acesso programático (Programmatic Access Token, PAT), use o seguinte código PySpark de exemplo:

  from pyspark.sql import SparkSession
  
  # Snowflake Horizon Catalog Configuration, change as per your environment
  
  CATALOG_URI = "https://<account_identifier>.snowflakecomputing.com/polaris/api/catalog"
  ROLE = "<role>"
  HORIZON_SESSION_ROLE = f"session:role:{ROLE}"
  CATALOG_NAME = "<database_name>" #provide in UPPER CASE
  SF_URL= "<account_identifier>.snowflakecomputing.com"
  SF_USER = "<user_name>" #provide in UPPER CASE
  SF_PASSWORD = "<user_password>"
  SF_SCHEMA = "<schema_name>" #provide in UPPER CASE
  SF_WAREHOUSE = "<warehouse_name>" #provide in UPPER CASE
  
  
  # Cloud Service Provider Region Configuration (where the Iceberg data is stored)
  REGION = "<region_name>"
  
  # Paste the External Oauth Access token that you generated in Snowflake here
  ACCESS_TOKEN = "<your_access_token>"
  
  # Paste the PAT you generated in Snowflake here
  PAT_TOKEN = "<your_PAT_token>"
  
  # Iceberg Version
  ICEBERG_VERSION = "1.9.1"
  
  #Snowflake Connector for Spark
  DRIVER_VERSION = "3.24.0" # (or above)
  SNOWFLAKE_CONNECTOR_VERSION = "3.1.6"
  
  
  try:
      spark.stop()
  except:
      pass
  
    spark = (
        SparkSession.builder
  
        .master("local[*]")
  .config("spark.ui.port", "0")
        .config("spark.driver.bindAddress", "127.0.0.1")
        .config("spark.driver.host", "127.0.0.1")
        .config("spark.driver.port", "0")
        .config("spark.blockManager.port", "0")
  
  
  # JAR Dependencies for Iceberg, Azure and Snowflake Connector for Spark
        .config(
   "spark.jars.packages",
   f"org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:{ICEBERG_VERSION},"
   f"org.apache.iceberg:iceberg-aws-bundle:{ICEBERG_VERSION},"
  
     # for Azure storage, use the below package and comment above azure bundle
            # f"org.apache.iceberg:iceberg-azure-bundle:{ICEBERG_VERSION}"
  # for Snowflake Connector for Spark
   f"net.snowflake:snowflake-jdbc:{DRIVER_VERSION},"
   f"net.snowflake:spark-snowflake_2.12:{SNOWFLAKE_CONNECTOR_VERSION}"
  
  )
        # Iceberg SQL Extensions
        .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
        .config("spark.sql.defaultCatalog", "horizoncatalog")
  .config("spark.sql.catalog.horizoncatalog", "org.apache.spark.sql.snowflake.catalog.SnowflakeFallbackCatalog")
  
    #Horizon REST Catalog Configuration
     .config(f"spark.sql.catalog.horizoncatalog.catalog-impl", "org.apache.iceberg.spark.SparkCatalog")
        .config(f"spark.sql.catalog.horizoncatalog.type", "rest")
        .config(f"spark.sql.catalog.horizoncatalog.uri", CATALOG_URI)
        .config(f"spark.sql.catalog.horizoncatalog.warehouse", CATALOG_NAME)
        .config(f"spark.sql.catalog.horizoncatalog.scope", HORIZON_SESSION_ROLE)
        .config(f"spark.sql.catalog.horizoncatalog.client.region", REGION)
        .config(f"spark.sql.catalog.horizoncatalog.credential", PAT_TOKEN)
  # for External Oauth use below and comment above configuration .token
  #.config(f"spark.sql.catalog.horizoncatalog.token", ACCESS_TOKEN)
  
  .config("spark.sql.catalog.horizoncatalog.io-impl","org.apache.iceberg.aws.s3.S3FileIO")
  # Enforcing policies using Snowflake Connector for Spark
  .config("spark.snowflake.sfURL", SF_URL)
  .config("spark.snowflake.sfUser", SF_USER)
  .config("spark.snowflake.sfPassword", SF_PASSWORD)
  # for External Oauth uncomment below and comment above configurations for user and password
  #.config("spark.snowflake.sfAuthenticator","oauth")
  #.config("spark.snowflake.sfToken",ACCESS_TOKEN)
  .config("spark.snowflake.sfDatabase", CATALOG_NAME)
  .config("spark.snowflake.sfSchema",SF_SCHEMA) # Optional
  .config("spark.snowflake.sfRole",ROLE)
  .config("spark.snowflake.sfWarehouse",SF_WAREHOUSE)
  
    # Required for vended credentials
   .config(f"spark.sql.catalog.horizoncatalog.header.X-Iceberg-Access-Delegation", "vended-credentials")
        .config("spark.sql.iceberg.vectorization.enabled", "false")
        .getOrCreate()
    )
    spark.sparkContext.setLogLevel("ERROR")
  

  Copy

  Onde:

  • <account_identifier> é o identificador da sua conta Snowflake que contém as tabelas Iceberg que você deseja consultar. Para encontrar esse identificador, consulte Identificadores de conta.

  • <your_access_token> é o seu token de acesso que você obteve. Para obter um token de acesso, consulte Obter um token de acesso para autenticação.

    Nota

    Para o OAuth externo, como alternativa, você pode configurar sua conexão com o mecanismo com atualização automática de token em vez de especificar um token de acesso.

  • <database_name> é o nome do banco de dados em sua conta Snowflake que contém as tabelas Iceberg gerenciadas pelo Snowflake que você deseja consultar.

    Nota

    As seguintes propriedades no Spark esperam o nome do seu banco de dados Snowflake, não o nome do seu warehouse Snowflake:

    • .warehouse

    • .sfDatabase

  • <role> é a função no Snowflake que está configurada com acesso às tabelas Iceberg que você deseja consultar. Por exemplo: DATA_ENGINEER.

  • <user_name> é o nome de usuário usado para acessar tabelas no Snowflake.

  • <user_password> é a senha do usuário que acessa as tabelas.

    Nota

    Essa senha pode ser o token de acesso programático (PAT) que você obteve para autenticação, se aplicável.

  • <schema_name> é o esquema no Snowflake em que as tabelas são armazenadas. Ele é opcional.

  • <warehouse_name> é o nome do warehouse Snowflake (instância de computação) que você deseja usar para avaliar as políticas.

  Importante

  Por padrão, o exemplo de código está configurado para tabelas Apache Iceberg™ armazenadas no Amazon S3. Se suas tabelas Iceberg estiverem armazenadas no Armazenamento do Azure (ADLS), execute as seguintes etapas:

  1. Comente a seguinte linha: f"org.apache.iceberg:iceberg-aws-bundle:{ICEBERG_VERSION}"

  2. Remova o comentário da seguinte linha: # f"org.apache.iceberg:iceberg-azure-bundle:{ICEBERG_VERSION}"

Consulta a uma tabela¶

spark.sql("SHOW NAMESPACES").show(truncate=False)
spark.sql("USE horizoncatalog.<schema_name>")
spark.sql("SHOW TABLES").show(truncate=False)
spark.sql("Select * from <your_table_name_in_snowflake>").show(truncate=False)