Dataproc Persistent History Server

Übersicht

Der Dataproc Persistent History Server (PHS) bietet Weboberflächen, um den Jobverlauf für Jobs aufzurufen, die in aktiven oder gelöschten Dataproc-Clustern ausgeführt werden. Sie ist in der Dataproc-Image-Version 1.5 und höher verfügbar und wird auf einem Dataproc-Cluster mit nur einem Knoten ausgeführt. Es bietet Weboberflächen für die folgenden Dateien und Daten:

  • MapReduce- und Spark-Verlaufsdateien

  • Flink-Jobverlaufsdateien (siehe Optionale Dataproc Flink-Komponente, um einen Dataproc-Cluster zum Ausführen von Flink-Jobs zu erstellen)

  • Datendateien der Anwendungszeitachse, die vom YARN Timeline Service v2 erstellt und in einer Bigtable-Instanz gespeichert wurden.

  • YARN-Aggregationslogs

Der Persistent History Server greift auf Spark- und MapReduce-Jobverlaufsdateien, Flink-Jobverlaufsdateien und YARN-Logdateien zu, die während der Lebensdauer von Dataproc-Jobclustern in Cloud Storage geschrieben wurden, und zeigt sie an.

Beschränkungen

  • Die Image-Version des PHS-Clusters muss mit der Image-Version der Dataproc-Jobcluster übereinstimmen. Sie können beispielsweise einen PHS-Cluster mit der Dataproc 2.0-Imageversion verwenden, um Jobverlaufsdateien von Jobs aufzurufen, die in Jobclustern mit der Dataproc 2.0-Imageversion ausgeführt wurden, die sich im Projekt befinden, in dem sich der PHS-Cluster befindet.

  • Ein PHS-Cluster unterstützt weder Kerberos noch die persönliche Authentifizierung.

Dataproc-PHS-Cluster erstellen

Sie können den folgenden Befehl gcloud dataproc clusters create in einem lokalen Terminal oder in Cloud Shell mit den folgenden Flags und Clustereigenschaften ausführen, um einen Dataproc-Cluster mit einem einzelnen Knoten für den Persistent History Server zu erstellen.

gcloud dataproc clusters create CLUSTER_NAME \
    --project=PROJECT \
    --region=REGION \
    --single-node \
    --enable-component-gateway \
    --optional-components=COMPONENT \
    --properties=PROPERTIES
  • CLUSTER_NAME: Geben Sie den Namen des PHS-Clusters an.
  • PROJECT: Geben Sie das Projekt an, das mit dem PHS-Cluster verknüpft werden soll. Dieses Projekt sollte mit dem Projekt übereinstimmen, das mit dem Cluster verknüpft ist, in dem Ihre Jobs ausgeführt werden (siehe Dataproc-Jobcluster erstellen).
  • REGION: Geben Sie eine Compute Engine-Region an, in der sich der PHS-Cluster befinden soll.
  • --single-node: Ein PHS-Cluster ist ein Dataproc-Cluster mit einem einzelnen Knoten.
  • --enable-component-gateway: Dieses Flag aktiviert die Weboberflächen des Component Gateways im PHS-Cluster.
  • COMPONENT: Verwenden Sie dieses Flag, um eine oder mehrere optionale Komponenten auf dem Cluster zu installieren. Sie müssen die optionale Komponente FLINK angeben, um den Flink HistoryServer-Webdienst auf dem PHS-Cluster auszuführen und Flink-Jobverlaufsdateien aufzurufen.
  • PROPERTIES. Geben Sie eine oder mehrere Clustereigenschaften an.

  • Optional können Sie das Flag --image-version hinzufügen, um die PHS-Cluster-Imageversion anzugeben. Die PHS-Image-Version muss mit der Image-Version der Dataproc-Jobcluster übereinstimmen. Siehe Einschränkungen.

    Hinweise:

    • In den Beispielen für Property-Werte in diesem Abschnitt wird das Platzhalterzeichen „*“ verwendet, damit der PHS mehrere Verzeichnisse im angegebenen Bucket abgleichen kann, in die verschiedene Jobcluster geschrieben wurden. Siehe Effizienzaspekte von Platzhaltern.
    • In den folgenden Beispielen sind separate --properties-Flags zu sehen, um die Lesbarkeit zu verbessern. Wenn Sie gcloud dataproc clusters create zum Erstellen eines Dataproc in Compute Engine-Clusters verwenden, sollten Sie mit einem --properties-Flag eine Liste von durch Kommas getrennten Eigenschaften angeben (siehe Formatierung von Clustereigenschaften).

    Properties:

    • yarn:yarn.nodemanager.remote-app-log-dir=gs://bucket-name/*/yarn-logs: Fügen Sie diese Property hinzu, um den Cloud Storage-Speicherort anzugeben, unter dem die PHS auf YARN-Logs zugreifen, die von Jobclustern geschrieben wurden.

    • spark:spark.history.fs.logDirectory=gs://bucket-name/*/spark-job-history: Fügen Sie diese Property hinzu, um den nichtflüchtigen Spark-Jobverlauf zu aktivieren. Dieses Attribut gibt den Speicherort an, an dem der PHS auf Spark-Jobverlaufslogs zugreift, die von Jobclustern geschrieben wurden.

      In Dataproc-Clustern der Version 2.0 und höher müssen außerdem die folgenden beiden Eigenschaften festgelegt werden, um PHS-Spark-Verlaufsprotokolle zu aktivieren (siehe Konfigurationsoptionen für den Spark History Server). Der Wert spark.history.custom.executor.log.url ist ein Literalwert, der {{PLATZHOLDER}} für Variablen enthält, die vom Persistent History Server festgelegt werden. Diese Variablen werden nicht von Nutzern festgelegt. Geben Sie den Property-Wert wie gezeigt ein.

      --properties=spark:spark.history.custom.executor.log.url.applyIncompleteApplication=false

      --properties=spark:spark.history.custom.executor.log.url={{YARN_LOG_SERVER_URL}}/{{NM_HOST}}:{{NM_PORT}}/{{CONTAINER_ID}}/{{CONTAINER_ID}}/{{USER}}/{{FILE_NAME}}

    • mapred:mapreduce.jobhistory.read-only.dir-pattern=gs://bucket-name/*/mapreduce-job-history/done: Fügen Sie diese Property hinzu, um den nichtflüchtigen MapReduce-Jobverlauf zu aktivieren. Dieses Attribut gibt den Cloud Storage-Speicherort an, in dem die PHS auf MapReduce-Jobverlaufslogs zugreift, die von Jobclustern geschrieben wurden.

    • dataproc:yarn.atsv2.bigtable.instance=projects/project-id/instance_id/bigtable-instance-id: Nachdem Sie Yarn Timeline Service v2 konfiguriert haben, fügen Sie diese Property hinzu, um Zeitachsendaten über den PHS-Cluster in den Weboberflächen YARN Application Timeline Service v2 und Tez aufzurufen (siehe Weboberflächen des Komponenten-Gateways).

    • flink:historyserver.archive.fs.dir=gs://bucket-name/*/flink-job-history/completed-jobs: Mit dieser Eigenschaft können Sie den Flink-HistoryServer so konfigurieren, dass er eine durch Kommas getrennte Liste von Verzeichnissen überwacht.

    Beispiele für Properties:

    --properties=spark:spark.history.fs.logDirectory=gs://bucket-name/*/spark-job-history

    --properties=mapred:mapreduce.jobhistory.read-only.dir-pattern=gs://bucket-name/*/mapreduce-job-history/done

    --properties=flink:flink.historyserver.archive.fs.dir=gs://bucket-name/*/flink-job-history/completed-jobs

Dataproc-Jobcluster erstellen

Sie können den folgenden Befehl in einem lokalen Terminal oder in Cloud Shell ausführen, um einen Dataproc-Jobcluster zu erstellen, der Jobs ausführt und Jobverlaufsdateien in einen Persistent History Server (PHS) schreibt.

gcloud dataproc clusters create CLUSTER_NAME \
    --project=PROJECT \
    --region=REGION \
    --optional-components=COMPONENT \
    --enable-component-gateway \
    --properties=PROPERTIES \
    other args ...
  • CLUSTER_NAME: Geben Sie den Namen des Jobclusters an.
  • PROJECT: Geben Sie das Projekt an, das mit dem Jobcluster verknüpft ist.
  • REGION: Geben Sie die Compute Engine-Region an, in der sich der Jobcluster befinden soll.
  • --enable-component-gateway: Mit diesem Flag werden die Weboberflächen des Component Gateways im Jobcluster aktiviert.
  • COMPONENT: Verwenden Sie dieses Flag, um eine oder mehrere optionale Komponenten auf dem Cluster zu installieren. Geben Sie die optionale Komponente FLINK an, um Flink-Jobs auf dem Cluster auszuführen.

  • PROPERTIES: Fügen Sie eine oder mehrere der folgenden Clustereigenschaften hinzu, um nicht standardmäßige Cloud Storage-Standorte und andere Jobcluster-Eigenschaften für PHS festzulegen.

    Hinweise:

    • In den Beispielen für Property-Werte in diesem Abschnitt wird der Platzhalter „*“ verwendet, damit der PHS mehrere Verzeichnisse im angegebenen Bucket abgleichen kann, in die verschiedene Jobcluster geschrieben wurden. Siehe Effizienzaspekte von Platzhaltern.
    • In den folgenden Beispielen sind separate --properties-Flags zu sehen, um die Lesbarkeit zu erleichtern. Wenn Sie gcloud dataproc clusters create zum Erstellen eines Dataproc in Compute Engine-Clusters verwenden, wird empfohlen, mit einem --properties-Flag eine Liste von durch Kommas getrennten Eigenschaften anzugeben (siehe Formatierung von Clustereigenschaften).

    Properties:

    • yarn:yarn.nodemanager.remote-app-log-dir: Zusammengefasste YARN-Protokolle sind standardmäßig in Dataproc-Jobclustern aktiviert und werden in den temporären Bucket des Clusters geschrieben. Fügen Sie dieses Attribut hinzu, um einen anderen Cloud Storage-Speicherort anzugeben, an dem der Cluster Aggregationsprotokolle für den Zugriff durch den Persistent History Server schreibt. 
      --properties=yarn:yarn.nodemanager.remote-app-log-dir=gs://bucket-name/directory-name/yarn-logs
    • spark:spark.history.fs.logDirectory und spark:spark.eventLog.dir: Standardmäßig werden Spark-Jobverlaufsdateien im Cluster temp bucket im Verzeichnis /spark-job-history gespeichert. Sie können diese Properties hinzufügen, um unterschiedliche Cloud Storage-Speicherorte für diese Dateien anzugeben. Wenn beide Properties verwendet werden, müssen sie auf Verzeichnisse im selben Bucket verweisen. 
      --properties=spark:spark.history.fs.logDirectory=gs://bucket-name/directory-name/spark-job-history

      --properties=spark:spark.eventLog.dir=gs://bucket-name/directory-name/spark-job-history
    • mapred:mapreduce.jobhistory.done-dir und mapred:mapreduce.jobhistory.intermediate-done-dir: Standardmäßig werden MapReduce-Jobverlaufsdateien im Cluster temp bucket in den Verzeichnissen /mapreduce-job-history/done und /mapreduce-job-history/intermediate-done gespeichert. Im Zwischenspeicher mapreduce.jobhistory.intermediate-done-dir wird vorübergehend gespeichert. Die Zwischendateien werden nach Abschluss des MapReduce-Jobs an den Speicherort mapreduce.jobhistory.done-dir verschoben. Sie können diese Properties hinzufügen, um für diese Dateien unterschiedliche Cloud Storage-Speicherorte anzugeben. Wenn beide Properties verwendet werden, müssen sie auf Verzeichnisse im selben Bucket verweisen. 
      --properties=mapred:mapreduce.jobhistory.done-dir=gs://bucket-name/directory-name/mapreduce-job-history/done

      --properties=mapred:mapreduce.jobhistory.intermediate-done-dir=gs://bucket-name/directory-name/mapreduce-job-history/intermediate-done
    • spark:spark.history.fs.gs.outputstream.type und spark:spark.history.fs.gs.outputstream.sync.min.interval.ms: Mit diesen Cloud Storage-Connector-Eigenschaften können Sie das Standardverhalten ändern, mit dem der Jobcluster Daten an Cloud Storage sendet. Der Standardwert für spark:spark.history.fs.gs.outputstream.type ist BASIC. Dabei werden Daten nach Abschluss des Jobs an Cloud Storage gesendet. Sie können diese Einstellung in FLUSHABLE_COMPOSITE ändern, um das Löschen so zu ändern, dass Daten während der Ausführung des Jobs in regelmäßigen Intervallen in Cloud Storage kopiert werden. 
      --properties=spark:spark.history.fs.gs.outputstream.type=FLUSHABLE_COMPOSITE
      Die Standardeinstellung für spark:spark.history.fs.gs.outputstream.sync.min.interval.ms, die die Häufigkeit der Datenübertragung in Cloud Storage steuert, ist 5000ms. Sie kann in ein anderes ms-Zeitintervall geändert werden: 
      --properties=spark:spark.history.fs.gs.outputstream.sync.min.interval.ms=intervalms
      Hinweis:Wenn Sie diese Eigenschaften festlegen möchten, muss für die Image-Version des Dataproc-Jobclusters der Cloud Storage-Connector Version 2.2.0 oder höher verwendet werden. Auf der Seite Liste der Dataproc-Image-Versionen können Sie die Connectorversion prüfen, die auf Image-Versionen installiert ist.
    • dataproc:yarn.atsv2.bigtable.instance: Nachdem Sie Yarn Timeline Service v2 konfiguriert haben, fügen Sie diese Property hinzu, um YARN-Zeitachsendaten in die angegebene Bigtable-Instanz zu schreiben, damit sie in den Weboberflächen des PHS-Clusters YARN Application Timeline Service v2 und Tez angezeigt werden können. Hinweis: Die Clustererstellung schlägt fehl, wenn die Bigtable-Instanz nicht vorhanden ist. 
      --properties=dataproc:yarn.atsv2.bigtable.instance=projects/project-id/instance_id/bigtable-instance-id
    • flink:jobhistory.archive.fs.dir: Der Flink JobManager archiviert abgeschlossene Flink-Jobs, indem er archivierte Jobinformationen in ein Dateisystemverzeichnis hochlädt. Mit diesem Attribut kannst du das Archivverzeichnis in flink-conf.yaml festlegen. 
      --properties=flink:jobmanager.archive.fs.dir=gs://bucket-name/job-cluster-1/flink-job-history/completed-jobs

PHS mit Spark-Batcharbeitslasten verwenden

So verwenden Sie den Persistent History Server mit Dataproc Serverless für Spark-Batcharbeitslasten:

  1. Erstellen Sie einen PHS-Cluster.

  2. Wählen Sie den PHS-Cluster aus oder geben Sie ihn an, wenn Sie eine Spark-Batcharbeitslast einreichen.

PHS mit Dataproc in der Google Kubernetes Engine verwenden

So verwenden Sie den Persistent History Server mit Dataproc in GKE:

  1. Erstellen Sie einen PHS-Cluster.

  2. Wählen Sie den PHS-Cluster aus oder geben Sie ihn an, wenn Sie einen virtuellen Dataproc-Cluster auf GKE erstellen.

Component Gateway-Weboberflächen

Klicken Sie in der Google Cloud Console auf der Dataproc-Seite Cluster auf den Namen des PHS-Clusters, um die Seite Clusterdetails zu öffnen. Wählen Sie auf dem Tab Weboberflächen die Links zum Komponenten-Gateway aus, um Weboberflächen zu öffnen, die auf dem PHS-Cluster ausgeführt werden.

Weboberfläche des Spark-Verlaufsservers

Der folgende Screenshot zeigt die Weboberfläche des Spark-History-Servers mit Links zu Spark-Jobs, die auf job-cluster-1 und job-cluster-2 ausgeführt werden, nachdem die spark.history.fs.logDirectory und spark:spark.eventLog.dir der Jobcluster und die spark.history.fs.logDirectory-Speicherorte des PHS-Clusters folgendermaßen erstellt wurden:

job-cluster-1 gs://example-cloud-storage-bucket/job-cluster-1/spark-job-history
job-cluster-2 gs://example-cloud-storage-bucket/job-cluster-2/spark-job-history
phs-cluster gs://example-cloud-storage-bucket/*/spark-job-history

Sie können Jobs nach Anwendungsname in der Weboberfläche des Spark-Verlaufsservers auflisten, indem Sie einen App-Namen in das Suchfeld eingeben. App-Namen können auf eine der folgenden Arten festgelegt werden (nach Priorität aufgelistet):

  1. Wird im Anwendungscode beim Erstellen des Spark-Kontexts festgelegt
  2. Wird vom Attribut spark.app.name festgelegt, wenn der Job gesendet wird
  3. Setzen Sie von Dataproc den vollständigen REST-Ressourcennamen für den Job (projects/project-id/regions/region/jobs/job-id).

Nutzer können einen App- oder Ressourcennamen in das Feld Suchen eingeben, um Jobs zu finden und aufzulisten.

Ereignisprotokolle

Die Spark History Server-Weboberfläche bietet eine Schaltfläche Ereignislog, auf die Sie klicken können, um Spark-Ereignislogs herunterzuladen. Diese Logs sind nützlich, um den Lebenszyklus der Spark-Anwendung zu untersuchen.

Spark-Jobs

Spark-Anwendungen sind in mehrere Jobs unterteilt, die weiter in mehrere Phasen unterteilt werden. Jede Phase kann mehrere Aufgaben enthalten, die auf Executor-Knoten (Workern) ausgeführt werden.

  • Klicken Sie in der Weboberfläche auf eine Spark-Anwendungs-ID, um die Seite „Spark-Jobs“ zu öffnen, die eine Ereigniszeitachse und eine Zusammenfassung der Jobs innerhalb der Anwendung enthält.

  • Klicken Sie auf einen Job, um die Seite "Jobdetails" mit einem gerichteten azyklischen Graphen (Directed Acyclic Graph, DAG) und einer Zusammenfassung der Jobphasen zu öffnen.

  • Klicken Sie auf eine Phase oder verwenden Sie den Tab „Phasen“, um eine Phase auszuwählen, um die Seite „Phasendetails“ zu öffnen.

    Stage Details umfassen eine DAG-Visualisierung, eine Ereigniszeitachse und Messwerte für die Aufgaben innerhalb der Phase. Auf dieser Seite können Sie Probleme mit strangulierten Aufgaben, Planerverzögerungen und Speichermangel beheben. Der DAG-Visualisierer zeigt die Codezeile, aus der die Phase abgeleitet ist, sodass Sie Probleme wieder im Code verfolgen können.

  • Klicken Sie auf den Tab „Executors“ (Ausführer), um Informationen zu den Treiber- und Executor-Knoten der Spark-Anwendung aufzurufen.

    Wichtige Informationen auf dieser Seite umfassen die Anzahl der Kerne und die Anzahl der Aufgaben, die auf jedem Executor ausgeführt wurden.

Tez-Weboberfläche

Tez ist die Standardausführungs-Engine für Hive und Pig in Dataproc. Wenn Sie einen Hive-Job in einem Dataproc-Jobcluster einreichen, wird eine Tez-Anwendung gestartet (siehe Apache Hive in Dataproc verwenden ).

Wenn Sie Yarn Timeline Service v2 konfiguriert und die Eigenschaft dataproc:yarn.atsv2.bigtable.instance beim Erstellen der PHS- und Dataproc-Jobcluster festgelegt haben, schreibt YARN generierte Zeitachsendaten für Hive- und Pig-Jobs in die angegebene Bigtable-Instanz, um sie abzurufen und auf der Tez-Weboberfläche anzuzeigen, die auf dem PHS-Server ausgeführt wird.

YARN Application Timeline V2-Weboberfläche

Wenn Sie Yarn Timeline Service v2 konfiguriert und die Eigenschaft dataproc:yarn.atsv2.bigtable.instance beim Erstellen der PHS- und Dataproc-Jobcluster festgelegt haben, schreibt YARN generierte Jobzeitachsendaten in die angegebene Bigtable-Instanz zum Abrufen und Anzeigen auf der Weboberfläche des YARN Application Timeline Service, die auf dem PHS-Server ausgeführt wird. Dataproc-Jobs werden in der Weboberfläche auf dem Tab Ablaufaktivität aufgeführt.

Yarn Timeline Service v2 konfigurieren

Um Yarn Timeline Service v2 zu konfigurieren, richten Sie eine Bigtable-Instanz ein und prüfen Sie gegebenenfalls die Rollen des Dienstkontos:

  1. Bigtable-Instanz erstellen

  2. Prüfen Sie gegebenenfalls die Rollen des Dienstkontos. Das standardmäßige VM-Dienstkonto, das von Dataproc-Cluster-VMs verwendet wird, hat die erforderlichen Berechtigungen zum Erstellen und Konfigurieren der Bigtable-Instanz für den YARN-Zeitplandienst. Wenn Sie Ihren Job oder PHS-Cluster mit einem benutzerdefinierten VM-Dienstkonto erstellen, muss das Konto entweder die Rolle Bigtable Administrator oder Bigtable User haben.

Erforderliches Tabellenschema

Für die Unterstützung von Dataproc-PHS für den YARN Timeline Service v2 ist ein bestimmtes Schema erforderlich, das in der Bigtable-Instanz erstellt wird. Dataproc erstellt das erforderliche Schema, wenn ein Jobcluster oder PHS-Cluster erstellt wird, bei dem die dataproc:yarn.atsv2.bigtable.instance-Eigenschaft so festgelegt ist, dass sie auf die Bigtable-Instanz verweist.

Das folgende Schema ist für die Bigtable-Instanz erforderlich:

Tabellen Spaltenfamilien
prod.timelineservice.application c,i,m
prod.timelineservice.app_flow m
prod.timelineservice.entity c,i,m
prod.timelineservice.flowactivity i
prod.timelineservice.flowrun i
prod.timelineservice.subapplication c,i,m

Bigtable-Speicherbereinigung

Sie können die Bigtable-Speicherbereinigung basierend auf dem Alter für ATSv2-Tabellen konfigurieren:

  • Installieren Sie cbt, einschließlich der Erstellung der .cbrtc file.

  • Erstellen Sie die altersbasierte Richtlinie für die automatische Speicherbereinigung für ATSv2:

export NUMBER_OF_DAYS = number \
cbt setgcpolicy prod.timelineservice.application c maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.application i maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.application m maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.app_flow m maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.entity c maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.entity i maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.entity m maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.flowactivity i maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.flowrun i maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.subapplication c maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.subapplication i maxage=${NUMBER_OF_DAYS} \
cbt setgcpolicy prod.timelineservice.subapplication m maxage=${NUMBER_OF_DAYS}

Hinweise:

NUMBER_OF_DAYS: Die maximale Anzahl von Tagen ist 30d.