Diese Seite wurde von der Cloud Translation API übersetzt.

Auf Metadaten in Apache Spark zugreifen

Auf dieser Seite wird beschrieben, wie Sie einen Dataproc-Cluster mit Spark erstellen.

Übersicht

Sie erstellen einen Cluster nach dem Dataproc Metastore-Dienst ist mit dem Dataplex-Lake verknüpft, damit der kann sich der Cluster auf den Hive Metastore-Endpunkt verlassen, um Zugriff auf Dataplex-Metadaten.

Auf Metadaten, die in Dataplex verwaltet werden, kann standardmäßig über Schnittstellen wie Hive Metastore zur Unterstützung von Spark-Abfragen. Die Abfragen werden im Dataproc-Cluster ausgeführt.

Legen Sie für Parquet-Daten das Spark-Attribut spark.sql.hive.convertMetastoreParquet auf false, um Ausführungsfehler zu vermeiden. Weitere Informationen

Dataproc-Cluster erstellen

Führen Sie die folgenden Befehle aus, um einen Dataproc-Cluster zu erstellen und den mit dem Dataplex-Datensee verknüpften Dataproc Metastore-Dienst anzugeben:

  GRPC_ENDPOINT=$(gcloud metastore services describe SERVICE_ID \
    --location LOCATION \
    --format "value(endpointUri)" | cut -c9-)

  WHDIR=$(gcloud metastore services describe SERVICE_ID \
    --location LOCATION \
    --format "value(hiveMetastoreConfig.configOverrides.'hive.metastore.warehouse.dir')")

  METASTORE_VERSION=$(gcloud metastore services describe SERVICE_ID \
    --location LOCATION \
    --format "value(hiveMetastoreConfig.version)")

  # This command  creates a cluster with default settings. You can customize
  # it as needed. The --optional-components, --initialization-actions,
  # --metadata and --properties flags are used to to connect with
  # the associated metastore.
  gcloud dataproc clusters create CLUSTER_ID \
    --project PROJECT \
    --region LOCATION \
    --scopes "https://www.googleapis.com/auth/cloud-platform" \
    --image-version 2.0-debian10 \
    --optional-components=DOCKER \
    --initialization-actions "gs://metastore-init-actions/metastore-grpc-proxy/metastore-grpc-proxy.sh" \
    --metadata "proxy-uri=$GRPC_ENDPOINT,hive-version=$METASTORE_VERSION" \
    --properties "hive:hive.metastore.uris=thrift://localhost:9083,hive:hive.metastore.warehouse.dir=$WHDIR"

Metadaten untersuchen

Sie können DQL-Abfragen ausführen, um die Metadaten zu untersuchen, und Spark-Abfragen, um Daten abzufragen.

Hinweis

Öffnen Sie eine SSH-Sitzung auf dem primären Knoten des Dataproc-Clusters.

VM_ZONE=$(gcloud dataproc clusters describe CLUSTER_ID \
  --project PROJECT \
  --region LOCATION \
  --format "value(config.gceClusterConfig.zoneUri)")
gcloud compute ssh CLUSTER_ID-m --project PROJECT --zone $VM_ZONE

Öffnen Sie in der Eingabeaufforderung des primären Knotens eine neue Python-REPL.
```
python3
```

Datenbanken auflisten

Jede Dataplex-Zone im Data Lake wird einer Metastore-Datenbank zugeordnet.

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("SHOW DATABASES")
  df.show()

Tabellen auflisten

Tabellen in einer der Zonen auflisten

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("SHOW TABLES IN ZONE_ID")
  df.show()

Daten abfragen

Fragen Sie die Daten in einer der Tabellen ab.

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  # Modify the SQL statement to retrieve or filter on table columns.
  df = session.sql("SELECT COLUMNS FROM ZONE_ID.TABLE_ID WHERE QUERY LIMIT 10")
  df.show()

Tabellen und Partitionen in Metadaten erstellen

Führen Sie DDL-Abfragen aus, um mit Apache Spark Tabellen und Partitionen in Dataplex-Metadaten zu erstellen.

Weitere Informationen zu den unterstützten Datentypen, Dateiformaten und Zeilenformaten finden Sie unter Unterstützte Werte.

Hinweis

Erstellen Sie vor dem Erstellen einer Tabelle ein Dataplex-Asset, das dem Cloud Storage-Bucket zugeordnet ist mit den zugrunde liegenden Daten. Weitere Informationen finden Sie unter Assets hinzufügen.

Tabelle erstellen

Parquet-, ORC-, AVRO-, CSV- und JSON-Tabellen werden unterstützt.

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("CREATE TABLE ZONE_ID.TABLE_ID (COLUMNS DATA_TYPE) PARTITIONED BY (COLUMN) STORED AS FILE_FORMAT ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION 'gs://MY_GCP_BUCKET/TABLE_LOCATION' TBLPROPERTIES('dataplex.entity.partition_style' = 'HIVE_COMPATIBLE')")
  df.show()

Tabelle ändern

Dataplex lässt es nicht zu, dass Sie den Standort einer Tabelle ändern oder die Partitionsspalten für eine Tabelle. Durch das Ändern einer Tabelle wird userManaged nicht automatisch auf true gesetzt.

In Spark SQL können Sie Tabellen umbenennen, Spalten hinzufügen und das Dateiformat einer Tabelle festlegen.

Tabelle umbenennen

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("ALTER TABLE OLD_TABLE_NAME RENAME TO NEW_TABLE_NAME")
  df.show()

Spalten hinzufügen

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("ALTER TABLE TABLE_NAME ADD COLUMN (COLUMN_NAME DATA_TYPE"))
  df.show()

Dateiformat festlegen

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("ALTER TABLE TABLE_NAME SET FILEFORMAT FILE_FORMAT")
  df.show()

Tabelle ablegen

Durch das Löschen einer Tabelle aus der Metadaten-API von Dataplex werden die zugrunde liegenden Daten in Cloud Storage nicht gelöscht.

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("DROP TABLE ZONE_ID.TABLE_ID")
  df.show()

Partition hinzufügen

Dataplex lässt nicht zu, dass eine Partition nach dem Erstellen geändert wird. Die Partition kann jedoch gelöscht werden.

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("ALTER TABLE ZONE_ID.TABLE_ID ADD PARTITION (COLUMN1=VALUE1) PARTITION (COLUMN2=VALUE2)")
  df.show()

Sie können mehrere Partitionen mit demselben Partitionsschlüssel und unterschiedlichen Partitionswerten hinzufügen, wie im vorherigen Beispiel gezeigt.

Partition erstellen

Führen Sie den folgenden Befehl aus, um eine Partition zu erstellen:

  import pyspark.sql as sql

  session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()

  df = session.sql("ALTER TABLE ZONE_ID.TABLE_ID DROP PARTITION (COLUMN=VALUE)")
  df.show()

Iceberg-Tabellen abfragen

Sie können Iceberg-Tabellen mit Apache Spark abfragen.

Hinweis

Spark SQL-Sitzung mit Iceberg einrichten

  spark-sql --packages org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:0.13.1 --conf
  spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf
  spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog --conf
  spark.sql.catalog.spark_catalog.type=hive --conf
  spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog --conf
  spark.sql.catalog.local.type=hadoop --conf
  spark.sql.catalog.local.warehouse=$PWD/warehouse

Iceberg-Tabelle erstellen

Führen Sie den folgenden Befehl aus, um eine Eisbergtabelle zu erstellen:

  CREATE TABLE ZONE_ID.TABLE_ID (COLUMNS DATA_TYPE) USING ICEBERG PARTITIONED BY (COLUMN) LOCATION 'gs://MY_GCP_BUCKET/TABLE_ID' TBLPROPERTIES ('write.format.default' = 'TABLE_FORMAT');

Überblick über die Eisberge und Geschichte

Mit Apache Spark können Sie Snapshots und den Verlauf von Iceberg-Tabellen abrufen.

Hinweis

Richten Sie eine PySpark-Sitzung mit der Iceberg-Unterstützung ein.

  pyspark --packages org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:0.14.1 --conf
  spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf
  spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog --conf
  spark.sql.catalog.spark_catalog.type=hive --conf
  spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog --conf
  spark.sql.catalog.local.type=hadoop --conf
  spark.sql.catalog.local.warehouse=$PWD/warehouse

Geschichte der Eisbergtische

Führen Sie den folgenden Befehl aus, um den Verlauf einer Iceberg-Tabelle abzurufen:

  spark.read.format("iceberg").load("ZONE_ID.TABLE_ID.history").show(truncate=False)

Übersichten von Iceberg-Tabellen abrufen

Führen Sie den folgenden Befehl aus, um einen Snapshot einer Iceberg-Tabelle abzurufen:

  spark.read.format("iceberg").load("ZONE_ID.TABLE_ID.snapshots").show(truncate=False, vertical=True)

Unterstützte Datentypen und Dateiformate

Die unterstützten Datentypen sind so definiert:

Datentyp	Werte
Primitive	`TINYINT` `SMALLINT` `INT` `BIGINT` `BOOLEAN` `FLOAT` `DOUBLE` `DOUBLE PRECISION` `STRING` `BINARY` `TIMESTAMP` `DECIMAL` `DATE`
Array	`ARRAY < DATA_TYPE >`
Struktur	`STRUCT < COLUMN : DATA_TYPE >`

Die unterstützten Dateiformate sind so definiert:

TEXTFILE
ORC
PARQUET
AVRO
JSONFILE

Weitere Informationen zu den Dateiformaten finden Sie unter Speicherformate.

Folgende Zeilenformate werden unterstützt:

EINGESCHRÄNKT [FIELDS TERMINATED BY CHAR]
SERDE SERDE_NAME [MIT SERDEPROPERTIES (PROPERTY_NAME=PROPERTY_VALUE, PROPERTY_NAME=PROPERTY_VALUE, ...)]

Nächste Schritte

Weitere Informationen zum Verwalten von Metadaten