Durchsetzen von Datenschutzrichtlinien bei der Abfrage von Apache Iceberg™-Tabellen von Apache Spark™¶

Snowflake-Konnektor für Spark herunterladen¶

Um 3.1.6 oder eine neuere Version des Snowflake-Konnektors für Spark herunterzuladen, folgen Sie den Anweisungen unter Installieren und Konfigurieren des Spark-Konnektors.

Spark mit Iceberg-Tabellen verbinden¶

In diesem Schritt verbinden Sie Spark über Horizon Catalog mit Iceberg-Tabellen. Diese Verbindung enthält Konfigurationen für die Verwendung des Snowflake-Konnektors für Spark mit Horizon Catalog, um die Tabellen abzufragen, die durch Snowflake-Datenschutzrichtlinien geschützt sind.

• Um Spark mit Iceberg-Tabellen über ein programmgesteuertes Zugriffstoken (PAT) zu verbinden, verwenden Sie den folgenden Beispiel-PySpark-Code:

  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

  Wobei:

  • <account_identifier> ist Ihr Snowflake-Kontobezeichner für das Snowflake-Konto, das die Iceberg-Tabellen enthält, die Sie abfragen möchten. Wie Sie diesen Bezeichner finden, erfahren Sie unter Kontobezeichner.

  • <your_access_token> ist Ihr -Zugriffstoken, das Sie erhalten haben. Um ein Zugriffstoken zu erhalten, siehe Zugriffstoken für die Authentifizierung besorgen.

    Bemerkung

    Für External OAuth können Sie auch Ihre Verbindung zur Engine mit automatischer Tokenaktualisierung konfigurieren, anstatt ein Zugriffstoken anzugeben.

  • <database_name> ist der Name der Datenbank in Ihrem Snowflake-Konto, die von Snowflake verwaltete Iceberg-Tabellen enthält, die Sie abfragen möchten.

    Bemerkung

    Die folgenden Eigenschaften in Spark erwarten Ihren Snowflake-Datenbanknamen, nicht den Namen Ihres Snowflake-Warehouses:

    • .warehouse

    • .sfDatabase

  • <role> ist die Rolle in Snowflake, die mit Zugriff auf die Iceberg-Tabellen konfiguriert ist, die Sie abfragen möchten. Beispiel: DATA_ENGINEER.

  • <user_name> ist der Benutzername, der für den Zugriff auf Tabellen in Snowflake verwendet wird.

  • <user_password> ist das Kennwort für den Benutzer, der auf die Tabellen zugreift.

    Bemerkung

    Dieses Kennwort kann das programmgesteuerte Zugriffstoken (PAT) sein, das Sie ggf. zur Authentifizierung erhalten haben.

  • <schema_name> ist das Schema in Snowflake, in dem die Tabellen gespeichert sind. Dies ist optional.

  • <warehouse_name> ist der Name des Snowflake-Warehouses (der Computeinstanz), das für die Auswertung von Richtlinien verwendet werden soll.

  Wichtig

  Standardmäßig ist das Codebeispiel für Apache Iceberg™-Tabellen eingerichtet, die auf Amazon S3 gespeichert sind. Wenn Ihre Iceberg-Tabellen auf Azure Storage (ADLS) gespeichert sind, führen Sie die folgenden Schritte aus:

  1. Versehen Sie die folgende Zeile mit Kommentarmarkierungen: f"org.apache.iceberg:iceberg-aws-bundle:{ICEBERG_VERSION}"

  2. Entfernen Sie die Kommentarmarkierungen in der folgenden Zeile: # f"org.apache.iceberg:iceberg-azure-bundle:{ICEBERG_VERSION}"

Eine Tabelle abfragen¶

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)