Föderierte Kataloge mit Spark und dem Lakehouse-Laufzeitkatalog abfragen

Nachdem Sie den cloudübergreifenden Datenzugriff im Lakehouse-Laufzeitkatalog eingerichtet haben, können Sie Remote-Iceberg-Tabellen über den Managed Service for Apache Spark abfragen, ohne Dateien migrieren oder separate Cloud Storage-Anmeldedaten verwalten zu müssen.

Der Lakehouse-Laufzeitkatalog übernimmt die Metadatensynchronisierung und den sicheren Datenzugriff über die Bereitstellung von Anmeldedaten. Wenn Spark eine föderierte Tabelle abfragt, übersetzt der Lakehouse-Laufzeitkatalog die Katalogmetadaten und stellt kurzlebige, bereichsbezogene Anmeldedaten für Spark bereit, um Daten aus Amazon S3 oder Cloud Storage zu lesen. Die Tabellen werden dabei so behandelt, als wären sie lokal in IhrerGoogle Cloud -Umgebung vorhanden.

Unterstützte föderierte Katalogtypen

Managed Service for Apache Spark unterstützt das Abfragen der folgenden föderierten Katalogtypen über den Iceberg REST-Katalog-Endpunkt des Lakehouse-Laufzeitkatalogs:

Typ des föderierten Katalogs Unterstützte Speicher-Backends Einrichtungsleitfaden im Lakehouse-Laufzeitkatalog
AWS Glue Data Catalog Amazon S3 (s3:// – Standard-AWS-Kontokatalog oder Amazon S3-Tabellen-Bucket) Mit AWS Glue verbinden
Databricks Unity Catalog Externe Amazon S3- (s3://) oder Cloud Storage-Speicherorte (gs://) Verbindung zu Databricks Unity Catalog herstellen
Snowflake Horizon Catalog Amazon S3 (s3://) oder Cloud Storage (gs://) Mit Snowflake Horizon Catalog verbinden

Informationen zu den Anforderungen an die Katalogbereitschaft und den zugrunde liegenden Spark-Parameterdefinitionen für jeden Speicheranbieter finden Sie unter Remote-Daten im Lakehouse-Laufzeitkatalog abfragen und Spark-Konfigurationsparameter.

Hinweis

  1. Melden Sie sich in Ihrem Google Cloud -Konto an. Wenn Sie mit Google Cloudnoch nicht vertraut sind, erstellen Sie einfach ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that you have the permissions required to complete this guide.

  4. Verify that billing is enabled for your Google Cloud project.

  5. Enable the Dataproc, BigLake, and BigQuery APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

  6. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  7. Verify that you have the permissions required to complete this guide.

  8. Verify that billing is enabled for your Google Cloud project.

  9. Enable the Dataproc, BigLake, and BigQuery APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

    10.
  10. Prüfen Sie, ob in Ihrem Remote-Katalog Iceberg-Tabellen vorhanden sind.
  11. Richten Sie im Lakehouse-Laufzeitkatalog eine cloudübergreifende Verbindung und einen föderierten Katalog für Ihren Remote-Kataloganbieter ein:

Erforderliche IAM-Rollen (Identity and Access Management)

Zum Ausführen der Beispiele auf dieser Seite sind bestimmte IAM-Rollen erforderlich. Je nach Organisationsrichtlinien wurden diese Rollen möglicherweise bereits gewährt. Informationen zum Prüfen von Rollenzuweisungen finden Sie unter Müssen Sie Rollen zuweisen?. Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Welche Rollen Sie benötigen, hängt davon ab, wie Ihre Arbeitslast ausgeführt wird. In Managed Service for Apache Spark-Clustern und Managed Service for Apache Spark-serverlosen Runtimes 2.3 und früher werden Arbeitslasten als Dienstkonto ausgeführt. In Managed Service for Apache Spark-serverlosen Laufzeiten ab Version 3.0 werden Arbeitslasten standardmäßig mit Ihren Endnutzer-Anmeldedaten (EUC) ausgeführt. Daher weisen Sie Ihrem Nutzerkonto anstelle eines Dienstkontos Datenzugriffsrollen zu. Wenn Sie ein benutzerdefiniertes Dienstkonto für eine Laufzeit der Version 3.0 oder höher angeben, wird die Arbeitslast mit den Anmeldedaten des Dienstkontos ausgeführt. Weitere Informationen finden Sie unter Rollen für serverlose Bereitstellungen.

Nutzerrollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Erstellen eines Managed Service for Apache Spark-Clusters, zum Senden einer Batcharbeitslast oder zum Erstellen einer interaktiven Sitzung benötigen:

  • Verwenden Sie einen Cluster oder eine serverlose Laufzeitversion 2.3 oder früher:
    • Dataproc-Bearbeiter (roles/dataproc.editor) für das Projekt
    • Dienstkontonutzer (roles/iam.serviceAccountUser) für das Compute Engine-Standarddienstkonto oder ein benutzerdefiniertes Dienstkonto
  • Serverlose Laufzeit 3.0 oder höher verwenden: Dataproc Serverless-Bearbeiter (roles/dataproc.serverlessEditor) für das Projekt
  • Föderierte Tabellen mit Endnutzeranmeldedaten in einer serverlosen Laufzeitversion 3.0 oder höher lesen: BigLake-Betrachter (roles/biglake.viewer) im Projekt, das den föderierten Katalog enthält
  • Dienstkonto-Anmeldedaten in einer serverlosen Laufzeit 3.0 oder höher verwenden: Dienstkontonutzer (roles/iam.serviceAccountUser) für das benutzerdefinierte Dienstkonto

Dienstkontorollen

Weisen Sie dem Dienstkonto, mit dem Ihr Arbeitslast ausgeführt wird, die folgenden Rollen zu. Serverlose Laufzeiten von Managed Service for Apache Spark ab Version 3.0 benötigen diese Rollen nur, wenn die Arbeitslast mit Dienstkonto-Anmeldedaten ausgeführt wird.

Damit das Compute Engine-Standarddienstkonto oder ein benutzerdefiniertes Dienstkonto die erforderlichen Berechtigungen zum Ausführen von Managed Service for Apache Spark-Jobs und zum Lesen von föderierten Tabellen hat, bitten Sie Ihren Administrator, dem Compute Engine-Standarddienstkonto oder einem benutzerdefinierten Dienstkonto die folgenden IAM-Rollen zuzuweisen:

  • Führen Sie Managed Service for Apache Spark-Jobs in einem Cluster oder einer serverlosen Laufzeit der Version 2.3 oder früher aus: Dataproc Worker (roles/dataproc.worker) im Projekt
  • Managed Service for Apache Spark-Jobs in einer serverlosen Laufzeitversion 3.0 oder höher ausführen:
  • Metadaten föderierter Tabellen zur Laufzeit ermitteln und lesen: BigLake Viewer (roles/biglake.viewer) für das Projekt, das den föderierten Katalog enthält

Damit der Dienst-Agent für Managed Service for Apache Spark (service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com) die erforderlichen Berechtigungen zum Auflösen der vereinfachten Konfigurationseigenschaft dataproc.lakehouse.catalog. hat, wenn Sie einen Cluster erstellen, eine Batcharbeitslast senden oder eine interaktive Sitzung erstellen, bitten Sie Ihren Administrator, dem Dienst-Agent für Managed Service for Apache Spark (service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com) die IAM-Rolle BigLake-Betrachter (roles/biglake.viewer) für das Projekt zu gewähren, das den föderierten Katalog enthält.

Ausgehende Netzwerkverbindungen für Amazon S3 konfigurieren

Wenn Sie föderierte Kataloge abfragen, die von Amazon S3 unterstützt werden, können serverlose Batcharbeitslasten und interaktive Sitzungen von Managed Service for Apache Spark sowie Managed Service for Apache Spark-Cluster mit interner IP-Adresse keine Verbindung zu Amazon S3-Endpunkten mit ihrer standardmäßigen Netzwerkkonfiguration herstellen. Stellen Sie einen Cloud Router und Cloud NAT in Ihrem Arbeitslast-VPC-Netzwerk bereit, um ausgehenden Traffic zu Amazon S3 zuzulassen:

gcloud compute routers create lakehouse-router \
    --network=NETWORK_NAME \
    --region=REGION

gcloud compute routers nats create lakehouse-nat \
    --router=lakehouse-router \
    --auto-allocate-nat-external-ips \
    --nat-all-subnet-ip-ranges \
    --region=REGION

Ersetzen Sie Folgendes:

  • NETWORK_NAME: Das VPC-Netzwerk für Ihren Managed Service for Apache Spark-Cluster, Ihre Batcharbeitslast oder Ihre interaktive Sitzung, z. B. default.
  • REGION: Die Google Cloud-Region für Ihren Managed Service for Apache Spark-Cluster, Ihre Batcharbeitslast oder Ihre interaktive Sitzung.

Managed Service for Apache Spark konfigurieren und föderierte Tabellen abfragen

Sie können einen Managed Service for Apache Spark-Cluster, eine serverlose Batcharbeitslast oder eine interaktive Sitzung so konfigurieren, dass eine Verbindung zu Ihrem föderierten Katalog im Lakehouse-Laufzeitkatalog hergestellt wird. Diese Managed Service for Apache Spark-Ressourcen unterstützen die vereinfachte Konfigurationseigenschaft dataproc.lakehouse.catalog.CATALOG_NAME (empfohlen), die die Definition des föderierten Katalogs aus dem Lakehouse-Laufzeitkatalog liest und die erforderlichen Iceberg REST-Katalog-Spark-Eigenschaften generiert, einschließlich der Speicher-io-impl- und der Header für die Bereitstellung von Anmeldedaten (die Clusterform der Eigenschaft fügt das Präfix dataproc: hinzu).

Sowohl die vereinfachte als auch die manuelle Konfiguration verwenden die Google-Autorisierung für Iceberg (GoogleAuthManager). Dafür ist eine der folgenden Cluster-Image- oder serverlosen Laufzeitversionen erforderlich:

  • Managed Service for Apache Spark-Cluster: Image-Versionen 2.2.65 und höher oder Image-Versionen 2.3.11 und höher.
  • Serverlose Batcharbeitslasten und interaktive Sitzungen von Managed Service for Apache Spark: 2.2-Laufzeiten 2.2.60 und höher, 2.3-Laufzeiten 2.3.10 und höher oder 3.0-Laufzeiten.

Für serverlose Batch-Arbeitslasten und interaktive Sitzungen hängen die erforderlichen Rollen von der Laufzeitversion ab. In Laufzeiten ab Version 3.0 werden standardmäßig Anmeldedaten von Endnutzern verwendet. Weitere Informationen finden Sie unter Erforderliche IAM-Rollen und Rollen für serverlose Bereitstellungen.

Cluster

Erstellen Sie einen Managed Service for Apache Spark-Cluster mit der optionalen Iceberg-Komponente. Verwenden Sie dazu vereinfachte Konfigurationsattribute (empfohlen) oder geben Sie Katalogattribute manuell an.

Vereinfachte Konfiguration mit Properties (empfohlen)

Erstellen Sie den Cluster mit dem Attribut für den föderierten Katalog:

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=REGION \
  --image-version=IMAGE_VERSION \
  --optional-components=ICEBERG,JUPYTER \
  --enable-component-gateway \
  --properties="dataproc:dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,spark:spark.sql.defaultCatalog=CATALOG_NAME"

Ersetzen Sie Folgendes:

  • CLUSTER_NAME: ein Name für Ihren Cluster.
  • PROJECT_ID: IhreGoogle Cloud Projekt-ID.
  • REGION: Die Google Cloud-Region für den Cluster, z. B. us-central1.
  • IMAGE_VERSION: die Image-Version von Managed Service for Apache Spark, z. B. 2.3-debian12. Informationen zu unterstützten Versionen finden Sie weiter oben in diesem Abschnitt unter „Voraussetzungen“.
  • CATALOG_NAME: Ein Name für den lokalen Spark-Katalog, z. B. federated_catalog. Er kann mit FEDERATED_CATALOG_NAME identisch sein.
  • FEDERATED_CATALOG_NAME: die ID des föderierten Katalogs, den Sie unter Vorbereitung erstellt haben.

Snowflake Horizon Catalog abfragen

Wenn Sie eine Snowflake Horizon Catalog-Abfrage ausführen, beachten Sie Folgendes:

  • Snowflake-Speicher: Die vereinfachte Konfiguration unterstützt Snowflake-Tabellen, die in Amazon S3 gespeichert sind. Verwenden Sie für Snowflake-Tabellen, die in Cloud Storage gespeichert sind, die manuelle Konfiguration mit io-impl auf org.apache.iceberg.gcp.gcs.GCSFileIO gesetzt.

  • Snowflake-STRING-Spalten:Wenn beim Abfragen von Snowflake-STRING-Spalten Fehler auftreten, bei denen die Richtlinie für die Speicherserialisierung der Tabelle in Snowflake nicht COMPATIBLE ist, können Sie die Richtlinie für die Speicherserialisierung in Snowflake in COMPATIBLE ändern oder die Parquet-Vektorisierung in Spark deaktivieren, indem Sie dem --properties-Flag die folgenden Eigenschaften hinzufügen:

    spark:spark.sql.iceberg.vectorization.enabled=false,spark:spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false

Erstellen Sie in Ihrem Jupyter-Notebook oder Ihrer PySpark-Anwendung die SparkSession, ohne Katalogkonfigurationen anzugeben:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

Ersetzen Sie APP_NAME durch den Namen Ihrer Spark-Anwendung.

Manuelle Konfiguration

Wenn Sie die vereinfachte Konfigurationseigenschaft nicht verwenden, erstellen Sie den Cluster wie zuvor in diesem Abschnitt beschrieben, ohne die Eigenschaft dataproc:dataproc.lakehouse.catalog. an das Flag --properties zu übergeben, und konfigurieren Sie dann SparkSession manuell. Eine Liste der Parameterwerte für Speicheranbieter finden Sie unter Spark-Konfigurationsparameter.

from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"

spark = SparkSession.builder.appName("APP_NAME") \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

Ersetzen Sie IO_IMPL durch die FileIO-Implementierung für den Speicher, der Ihre Tabellen enthält:

  • Tabellen in Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
  • Tabellen in Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

Das Attribut s3.client-factory-impl gilt nur für Tabellen in Amazon S3. Lassen Sie diese Zeile für Cloud Storage-Tabellen weg.

Serverlos

Senden Sie eine PySpark-Batcharbeitslast an Managed Service for Apache Spark Serverless. Verwenden Sie dazu vereinfachte Konfigurationseigenschaften (empfohlen) oder geben Sie Katalogeigenschaften manuell an.

Vereinfachte Konfiguration mit Properties (empfohlen)

Übergeben Sie die Eigenschaft des föderierten Katalogs ohne das Präfix dataproc:, wenn Sie die Batcharbeitslast einreichen:

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,spark.sql.defaultCatalog=CATALOG_NAME"

Ersetzen Sie Folgendes:

  • PYSPARK_FILE: der gs://-Cloud Storage-Pfad zu Ihrer PySpark-Anwendungsdatei.
  • PROJECT_ID: IhreGoogle Cloud -Projekt-ID.
  • REGION: die Region für die Batch-Arbeitslast.
  • RUNTIME_VERSION: die serverlose Laufzeitversion von Managed Service for Apache Spark, z. B. 3.0. Informationen zu den unterstützten Versionen finden Sie weiter oben in diesem Abschnitt unter „Voraussetzungen“. Die erforderlichen IAM-Rollen hängen von der Laufzeitversion ab.
  • CATALOG_NAME: ein Name für den lokalen Spark-Katalog, z. B. federated_catalog. Er kann mit FEDERATED_CATALOG_NAME identisch sein.
  • FEDERATED_CATALOG_NAME: die ID des föderierten Katalogs, den Sie unter Vorbereitung erstellt haben.

Abfragen des Snowflake Horizon Catalog

Wenn Sie eine Snowflake Horizon Catalog-Abfrage ausführen, beachten Sie Folgendes:

  • Snowflake-Speicher:Die vereinfachte Konfiguration unterstützt Snowflake-Tabellen, die in Amazon S3 gespeichert sind. Verwenden Sie für Snowflake-Tabellen, die in Cloud Storage gespeichert sind, die manuelle Konfiguration mit io-impl auf org.apache.iceberg.gcp.gcs.GCSFileIO gesetzt.

  • Snowflake-STRING-Spalten:Wenn beim Abfragen von Snowflake-STRING-Spalten Fehler auftreten, bei denen die Richtlinie für die Speicherserialisierung der Tabelle in Snowflake nicht COMPATIBLE ist, können Sie die Richtlinie für die Speicherserialisierung in Snowflake in COMPATIBLE ändern oder die Parquet-Vektorisierung in Spark deaktivieren, indem Sie dem --properties-Flag die folgenden Eigenschaften hinzufügen:

    spark.sql.iceberg.vectorization.enabled=false,spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false

Erstellen Sie in der PySpark-Anwendungsdatei die SparkSession, ohne Katalogkonfigurationen anzugeben:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

Manuelle Konfiguration

Wenn Sie das vereinfachte Konfigurationsattribut nicht verwenden, geben Sie die Katalogattribute mit dem Flag --properties des Befehls gcloud dataproc batches submit an. Eine Liste der Parameterwerte für Cloud Storage-Anbieter finden Sie unter Spark-Konfigurationsparameter.

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="\
spark.sql.defaultCatalog=CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\
spark.sql.catalog.CATALOG_NAME.type=rest,\
spark.sql.catalog.CATALOG_NAME.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
spark.sql.catalog.CATALOG_NAME.warehouse=bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME.header.x-goog-user-project=PROJECT_ID,\
spark.sql.catalog.CATALOG_NAME.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
spark.sql.catalog.CATALOG_NAME.header.X-Iceberg-Access-Delegation=vended-credentials,\
spark.sql.catalog.CATALOG_NAME.io-impl=IO_IMPL,\
spark.sql.catalog.CATALOG_NAME.s3.client-factory-impl=org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

Ersetzen Sie IO_IMPL durch die FileIO-Implementierung für den Speicher, der Ihre Tabellen enthält:

  • Tabellen in Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
  • Tabellen in Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

Das Attribut s3.client-factory-impl gilt nur für Tabellen in Amazon S3. Lassen Sie dieses Attribut für Cloud Storage-Tabellen weg.

Serverlos

Erstellen Sie eine serverlose interaktive Managed Service for Apache Spark-Sitzung in einem Notebook über die DataprocSparkSession API. Verwenden Sie dazu vereinfachte Konfigurationseigenschaften (empfohlen) oder geben Sie Katalogeigenschaften manuell an.

Vereinfachte Konfiguration mit Properties (empfohlen)

In einem Notebook, in dem das Paket dataproc-spark-connect installiert ist, z. B. einem BigQuery Studio-Notebook oder einem Vertex AI Workbench-Notebook, übergeben Sie beim Erstellen der Sitzung das Attribut für den föderierten Katalog ohne das Präfix dataproc::

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config(f'dataproc.lakehouse.catalog.{catalog_name}', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config('spark.sql.defaultCatalog', catalog_name) \
.getOrCreate()

Ersetzen Sie Folgendes:

  • CATALOG_NAME: Ein Name für den lokalen Spark-Katalog, z. B. federated_catalog. Er kann mit FEDERATED_CATALOG_NAME identisch sein.
  • RUNTIME_VERSION: die serverlose Laufzeitversion von Managed Service for Apache Spark, z. B. 3.0. Informationen zu unterstützten Versionen finden Sie weiter oben in diesem Abschnitt unter „Anforderungen“. Die erforderlichen IAM-Rollen hängen von der Laufzeitversion ab.
  • APP_NAME: der Name Ihrer Spark-Anwendung.
  • PROJECT_ID: IhreGoogle Cloud -Projekt-ID.
  • FEDERATED_CATALOG_NAME: die ID des föderierten Katalogs, den Sie unter Vorbereitung erstellt haben.

Sie können diese Attribute auch in einer Sitzungsvorlage festlegen.

Abfragen des Snowflake Horizon Catalog

Wenn Sie eine Snowflake Horizon Catalog-Abfrage ausführen, beachten Sie Folgendes:

  • Snowflake-Speicher:Die vereinfachte Konfiguration unterstützt Snowflake-Tabellen, die in Amazon S3 gespeichert sind. Verwenden Sie für Snowflake-Tabellen, die in Cloud Storage gespeichert sind, die manuelle Konfiguration mit io-impl auf org.apache.iceberg.gcp.gcs.GCSFileIO gesetzt.

  • Snowflake-Spalten STRING:Wenn beim Abfragen von Snowflake-Spalten STRING Fehler auftreten und die Richtlinie für die Speicherserialisierung der Tabelle in Snowflake nicht COMPATIBLE ist, können Sie die Richtlinie für die Speicherserialisierung in Snowflake in COMPATIBLE ändern oder die Parquet-Vektorisierung in Spark deaktivieren, indem Sie dem Sitzungs-Builder die folgenden .config()-Aufrufe hinzufügen:

    .config('spark.sql.iceberg.vectorization.enabled', 'false') \
.config(f'spark.sql.catalog.{catalog_name}.table-override.read.parquet.vectorization.enabled', 'false')

Manuelle Konfiguration

Wenn Sie das vereinfachte Konfigurationsattribut nicht verwenden, geben Sie die Katalogattribute beim Erstellen der Sitzung an. Eine Liste der Parameterwerte für Speicheranbieter finden Sie unter Spark-Konfigurationsparameter.

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

Ersetzen Sie RUNTIME_VERSION wie im Beispiel für die vereinfachte Konfiguration weiter oben auf diesem Tab beschrieben.

Ersetzen Sie IO_IMPL durch die FileIO-Implementierung für den Speicher, der Ihre Tabellen enthält:

  • Tabellen in Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
  • Tabellen in Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

Das Attribut s3.client-factory-impl gilt nur für Tabellen in Amazon S3. Lassen Sie diese Zeile für Cloud Storage-Tabellen weg.

Föderierte Tabellen abfragen

Nachdem Sie Ihre Spark-Sitzung konfiguriert haben, können Sie Tabellen in Ihrem föderierten Katalog mit PySpark oder Spark SQL abfragen.

  1. Föderierte Tabelle mit der PySpark DataFrame API lesen:

    df = spark.table("CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME")
df.show(10, truncate=False)

    Ersetzen Sie Folgendes:

    • CATALOG_NAME: der Name des lokalen Spark-Katalogs. Wenn Sie spark.sql.defaultCatalog auf CATALOG_NAME setzen, können Sie dieses Präfix weglassen.
    • NAMESPACE_NAME: der Namespace (Datenbank oder Schema) im föderierten Katalog.
    • TABLE_NAME: der Tabellenname im föderierten Katalog.

  2. Fragen Sie die föderierte Tabelle mit Spark SQL ab:

    spark.sql("""
    SELECT *
    FROM CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME
    LIMIT 10
""").show()

Daten aus föderierten und lokalen Katalogen zusammenführen oder materialisieren

Föderierte Kataloge haben Lesezugriff. Wenn Sie Remote-Cloud-übergreifende Daten mit lokalen Tabellen zusammenführen oder föderierte Abfrageergebnisse in Cloud Storage materialisieren möchten, hängen Sie Ihren föderierten Katalog und einen Lese-/Schreibkatalog im Lakehouse-Laufzeitkatalog mit kommagetrennten dataproc.lakehouse.catalog.-Eigenschaften an:

Cluster

Beispiel für einen Managed Service for Apache Spark-Cluster:

--properties="dataproc:dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc:dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark:spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

Serverlos

Beispiel für eine serverlose Batcharbeitslast in Managed Service for Apache Spark:

--properties="dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

Serverlos

Beispiel für eine serverlose interaktive Sitzung von Managed Service for Apache Spark:

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
  .dataprocSessionConfig(session_config) \
  .config('dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
  .config('dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID') \
  .config('spark.sql.defaultCatalog', 'LOCAL_SPARK_CATALOG') \
  .getOrCreate()

Ersetzen Sie RUNTIME_VERSION wie im Beispiel für eine interaktive Sitzung weiter oben auf dieser Seite beschrieben.

Nachdem Sie diese Eigenschaften festgelegt haben, können Sie den föderierten Katalog abfragen und die Ergebnisse in eine Tabelle in Ihrem lokalen Lakehouse-Laufzeitkatalog schreiben.

Das Hauptkonto, das Ihre Arbeitslast ausführt, ist das Dienstkonto für Cluster, serverlose Laufzeiten 2.3 und niedriger sowie serverlose Laufzeiten 3.0 und höher mit Dienstkonto-Anmeldedaten. Bei serverlosen Runtimes 3.0 und höher mit Endnutzeranmeldedaten (Standard) ist das Hauptkonto Ihr Nutzerkonto.

Bitten Sie Ihren Administrator, dem Prinzipal, der Ihre Arbeitslast ausführt, die folgenden IAM-Rollen zuzuweisen, damit er die erforderlichen Berechtigungen zum Schreiben von Tabellendaten in Ihren lokalen Katalog hat:

  • Tabellendaten im Modus für die Bereitstellung von Anmeldedaten schreiben: BigLake-Editor (roles/biglake.editor) für das Projekt, das den lokalen Katalog enthält
  • Tabellendaten im Modus ohne Bereitstellung von Anmeldedaten schreiben:
    • BigLake Editor (roles/biglake.editor) für das Projekt, das den lokalen Katalog enthält
    • Storage Object User (roles/storage.objectUser) für alle Cloud Storage-Buckets, die mit dem lokalen Katalog verknüpft sind

Im folgenden Beispiel wird eine Tabelle in Ihrem lokalen Katalog aus einer föderierten Tabelle erstellt:

spark.sql("""
    CREATE TABLE IF NOT EXISTS LOCAL_SPARK_CATALOG.LOCAL_NAMESPACE.materialized_table
    USING iceberg
    AS SELECT *
    FROM FEDERATED_SPARK_CATALOG.REMOTE_NAMESPACE.REMOTE_TABLE
""")

Nächste Schritte