Apache Spark™에서 Apache Iceberg™ 테이블 쿼리 시 데이터 보호 정책 적용¶

Spark용 Snowflake Connector 다운로드하기¶

Spark용 Snowflake Connector 3.1.6 이상 버전을 다운로드하려면 :doc:`/user-guide/spark-connector-install`의 지침을 따릅니다.

Spark를 Iceberg 테이블에 연결하기¶

이 단계에서는 Horizon Catalog를 통해 Spark를 Iceberg 테이블에 연결합니다. 이 연결에는 Horizon Catalog가 포함된 Spark용 Snowflake Connector를 사용하여 Snowflake 데이터 보호 정책으로 보호되는 테이블을 쿼리하기 위한 구성이 포함됩니다.

• 프로그래밍 방식 액세스 토큰(PAT)을 사용하여 Spark를 Iceberg 테이블에 연결하려면 다음 예제 PySpark 코드를 사용합니다.

  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

  여기서

  • <account_identifier>`는 쿼리하려는 Iceberg 테이블이 포함된 Snowflake 계정의 Snowflake 계정 식별자입니다. 이 식별자를 찾으려면 :doc:/user-guide/admin-account-identifier` 섹션을 참조하세요.

  • :code:`<your_access_token>`은 획득한 액세스 토큰입니다. 액세스 토큰을 얻으려면 :ref:`인증을 위한 액세스 토큰 얻기<label-tables_iceberg_query_using_external_query_engine_snowflake_horizon_generate_access_token>`를 참조하세요.

    참고

    또는 외부 OAuth의 경우 액세스 토큰을 지정하는 대신 자동 토큰 새로 고침을 사용하여 엔진에 대한 연결을 구성할 수 있습니다.

  • :code:`<database_name>`은 쿼리하려는 Snowflake 관리 Iceberg 테이블이 포함된 Snowflake 계정의 데이터베이스 이름입니다.

    참고

    Spark의 다음 속성에는 Snowflake 웨어하우스 이름이 아닌 Snowflake 데이터베이스 이름이 필요합니다.

    • .warehouse

    • .sfDatabase

  • :code:`<role>`은 쿼리하려는 Iceberg 테이블에 대한 액세스 권한으로 구성된 Snowflake의 역할입니다. 예를 들어 DATA_ENGINEER 과 같습니다.

  • :code:`<user_name>`은 Snowflake에서 테이블에 액세스하는 데 사용되는 사용자 이름입니다.

  • :code:`<user_password>`는 테이블에 액세스하는 사용자의 비밀번호입니다.

    참고

    이 비밀번호는 해당하는 경우 인증을 위해 얻은 프로그래밍 방식 액세스 토큰(PAT)일 수 있습니다.

  • :code:`<schema_name>`은 테이블이 저장되는 Snowflake의 스키마입니다. 이는 선택 사항입니다.

  • :code:`<warehouse_name>`은 정책 평가에 사용할 Snowflake 웨어하우스(컴퓨팅 인스턴스) 이름입니다.

  중요

  기본적으로 코드 예제는 Amazon S3에 저장된 Apache Iceberg™ 테이블에 대해 설정됩니다. Iceberg 테이블이 Azure Storage(ADLS)에 저장되는 경우 다음 단계를 수행합니다.

  1. 다음 줄을 주석으로 처리합니다. f"org.apache.iceberg:iceberg-aws-bundle:{ICEBERG_VERSION}"

  2. 다음 줄의 주석 처리를 제거합니다. # f"org.apache.iceberg:iceberg-azure-bundle:{ICEBERG_VERSION}"

테이블 쿼리하기¶

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)