SLS を使用した Spark ジョブのログ収集 - Container Service for Kubernetes

Container Service for Kubernetes (ACK) クラスターで Spark ジョブを実行すると、大量のログが異なる Pod に分散して生成されるため、ログ管理が複雑になります。Simple Log Service (SLS) は、ログの収集、処理、クエリ、分析、可視化、アラートのためのワンストップソリューションを提供します。このトピックでは、SLS を使用して ACK クラスターで実行される Spark ジョブのログを効率的に管理する方法を説明します。

前提条件

ack-spark-operator アドオンがデプロイされていること。
SLS プロジェクトが作成されていること。詳細については、「プロジェクトの管理」をご参照ください。
Logtail コンポーネントがインストールされていること。

概要

このトピックでは、Spark ジョブによって生成されるシステムログとアプリケーションログを管理するために SLS を設定する、以下の手順を説明します。

log4j JSON テンプレートレイアウトの依存関係を含む Spark コンテナイメージをビルドし、そのイメージをコンテナイメージレジストリにプッシュします。
ConfigMap を作成して Log4j2 を設定し、ログレベルを INFO に、ログ出力形式を JSONL に設定します。
AliyunLogConfig リソースを作成します。これにより、SLS は指定された Logstore に対応する Logtail 設定を自動的に作成し、Spark オペレーターが送信した Spark ジョブからログを収集します。
サンプルの Spark ジョブを作成して実行し、Pod のログが JSONL 形式であることを確認し、主要なフィールドの定義を確認します。
SLS コンソールにログインし、特定の時間範囲内の Spark ジョブのログをクエリおよび分析します。
(オプション) クリーンアップ：テストが完了したら、課金を避けるために作成した Spark ジョブやその他のリソースを削除します。

手順1：Spark コンテナイメージのビルド

次の Dockerfile (この例では Spark 3.5.3 を使用) を作成し、必要な依存関係を Spark のクラスパスに追加します。ビルドが完了したら、イメージをコンテナイメージレジストリにプッシュします。ログの収集と解析を簡素化するため、ログ出力には JSONL 形式を使用します。

ARG SPARK_IMAGE=<SPARK_IMAGE>  # <SPARK_IMAGE> をご自身の Spark ベースイメージに置き換えます。
FROM ${SPARK_IMAGE}
# log4j-layout-template-json の依存関係を追加します
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/logging/log4j/log4j-layout-template-json/2.24.1/log4j-layout-template-json-2.24.1.jar ${SPARK_HOME}/jars

手順2：Log4j2 ログの設定

次の内容で spark-log-conf.yaml という名前のファイルを作成します。この設定では、ログレベルが INFO に、出力形式が JSONL に設定されます。ログテンプレートとして、標準化されたログ形式である Elastic Common Schema (ECS) を使用します。設定の詳細については、「Log4j ログの収集」をご参照ください。

apiVersion: v1
kind: ConfigMap
metadata:
  name: spark-log-conf
  namespace: default
data:
  log4j2.properties: |
    # すべてをコンソールとファイルに記録するように設定します
    rootLogger.level = info
    rootLogger.appenderRefs = console, file
    rootLogger.appenderRef.console.ref = STDOUT
    rootLogger.appenderRef.file.ref = FileAppender
    appender.console.name = STDOUT
    appender.console.type = Console
    appender.console.layout.type = JsonTemplateLayout
    appender.console.layout.eventTemplateUri = classpath:EcsLayout.json
    appender.file.name = FileAppender
    appender.file.type = File
    appender.file.fileName = /opt/spark/logs/spark.log
    appender.file.layout.type = JsonTemplateLayout
    appender.file.layout.eventTemplateUri = classpath:EcsLayout.json

次のコマンドを実行して ConfigMap リソースを作成します。

kubectl apply -f spark-log-conf.yaml

予想される出力：

configmap/spark-log-conf created

手順3：Logtail 設定の作成

次の内容で aliyun-log-config.yaml という名前の AliyunLogConfig マニフェストファイルを作成します。<SLS_PROJECT> をご自身の SLS プロジェクトの名前に、<SLS_LOGSTORE> をご自身の Logstore の名前に置き換えてください。設定オプションの詳細については、「AliyunLogConfig を使用した Logtail 設定の管理」をご参照ください。

apiVersion: log.alibabacloud.com/v1alpha1
kind: AliyunLogConfig
metadata:
  name: spark
  namespace: default
spec:
  # (オプション) 宛先プロジェクトの名前。 デフォルト値：k8s-log-<Your_Cluster_ID>。
  project: <SLS_PROJECT>
  # Logstore の名前。 指定した Logstore が存在しない場合、SLS が自動的に作成します。
  logstore: <SLS_LOGSTORE>
  # Logtail 設定。
  logtailConfig:
    # Logtail 設定の名前。
    configName: spark
    # データソースのタイプ。 `file` はテキストログを示します。
    inputType: file
    # ログ入力の設定。
    inputDetail:
      # ログファイルが配置されているディレクトリ。
      logPath: /opt/spark/logs
      # ログファイルの名前。 ワイルドカード文字がサポートされています。
      filePattern: '*.log'
      # ログファイルのエンコーディング。
      fileEncoding: utf8
      # ログタイプ。
      logType: json_log
      localStorage: true
      key:
      - content
      logBeginRegex: .*
      logTimezone: ''
      discardNonUtf8: false
      discardUnmatch: true
      preserve: true
      preserveDepth: 0
      regex: (.*)
      outputType: LogService
      topicFormat: none
      adjustTimezone: false
      enableRawLog: false
      # コンテナからテキストログを収集します。
      dockerFile: true
      # 詳細設定。
      advanced:
        # コンテナメタデータのプレビュー。
        collect_containers_flag: true
        # Kubernetes 収集設定。
        k8s:
          # label による Pod のフィルタリング。
          IncludeK8sLabel:
            sparkoperator.k8s.io/launched-by-spark-operator: "true"
          # 名前によるコンテナのフィルタリング。
          K8sContainerRegex: "^spark-kubernetes-(driver|executor)$"
          # 追加のログタグ設定。
          ExternalK8sLabelTag:
            spark-app-name: spark-app-name
            spark-version: spark-version
            spark-role: spark-role
            spark-app-selector: spark-app-selector
            sparkoperator.k8s.io/submission-id: sparkoperator.k8s.io/submission-id
      # ログ処理プラグイン。
      plugin:
        processors:
        # ログ分割。
        - type: processor_split_log_string
          detail:
            SplitKey: content
            SplitSep: ''
        # JSON フィールド解析。
        - type: processor_json
          detail:
            ExpandArray: false
            ExpandConnector: ''
            ExpandDepth: 0
            IgnoreFirstConnector: false
            SourceKey: content
            KeepSource: false
            KeepSourceIfParseError: true
            NoKeyError: false
            UseSourceKeyAsPrefix: false
        # ログタイムスタンプ抽出。
        - type: processor_strptime
          detail:
            SourceKey: '@timestamp'
            Format: '%Y-%m-%dT%H:%M:%S.%fZ'
            KeepSource: false
            AdjustUTCOffset: true
            UTCOffset: 0
            AlarmIfFail: false

次のコマンドを実行して Logtail 設定を作成します。

kubectl apply -f aliyun-log-config.yaml

次の手順に従って、新しい Logstore と Logtail 設定を表示します。

Log Serviceコンソールにログインします。
[プロジェクト] セクションで、管理するプロジェクトをクリックします。
[ログストレージ] > [ログストア] タブで、ターゲットログストアの前にある [>] アイコンをクリックし、[データ収集] > [Logtail 構成] を選択します。

[Logtail 設定] ページに、既存の Logtail 設定のリストが表示されます。リストには、1 つのマシングループに関連付けられた spark という名前の設定が含まれています。
ターゲット Logtail 構成をクリックして、詳細を表示します。

手順4：サンプルの Spark ジョブの送信

次の内容で spark-pi.yaml という名前の SparkApplication マニフェストファイルを作成します。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pi
  namespace: default
spec:
  type: Scala
  mode: cluster
  image: <SPARK_IMAGE>
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.3.jar
  arguments: 
  - "5000"
  sparkVersion: 3.5.3
  sparkConfigMap: spark-log-conf
  driver:
    cores: 1
    memory: 512m
    serviceAccount: spark-operator-spark
  executor:
    instances: 1
    cores: 1
    memory: 4g

次のコマンドを実行してジョブを送信します。

kubectl apply -f spark-pi.yaml

ジョブが完了したら、ドライバー Pod ログの最後の 10 行を表示します。

kubectl logs --tail=10 spark-pi-driver

予想される出力：

{"@timestamp":"2024-11-20T11:45:48.487Z","ecs.version":"1.2.0","log.level":"WARN","message":"Kubernetes client has been closed.","process.thread.name":"-937428334-pool-19-thread-1","log.logger":"org.apache.spark.scheduler.cluster.k8s.ExecutorPodsWatchSnapshotSource"}
{"@timestamp":"2024-11-20T11:45:48.585Z","ecs.version":"1.2.0","log.level":"INFO","message":"MapOutputTrackerMasterEndpoint stopped!","process.thread.name":"dispatcher-event-loop-7","log.logger":"org.apache.spark.MapOutputTrackerMasterEndpoint"}
{"@timestamp":"2024-11-20T11:45:48.592Z","ecs.version":"1.2.0","log.level":"INFO","message":"MemoryStore cleared","process.thread.name":"main","log.logger":"org.apache.spark.storage.memory.MemoryStore"}
{"@timestamp":"2024-11-20T11:45:48.592Z","ecs.version":"1.2.0","log.level":"INFO","message":"BlockManager stopped","process.thread.name":"main","log.logger":"org.apache.spark.storage.BlockManager"}
{"@timestamp":"2024-11-20T11:45:48.596Z","ecs.version":"1.2.0","log.level":"INFO","message":"BlockManagerMaster stopped","process.thread.name":"main","log.logger":"org.apache.spark.storage.BlockManagerMaster"}
{"@timestamp":"2024-11-20T11:45:48.598Z","ecs.version":"1.2.0","log.level":"INFO","message":"OutputCommitCoordinator stopped!","process.thread.name":"dispatcher-event-loop-1","log.logger":"org.apache.spark.scheduler.OutputCommitCoordinator$OutputCommitCoordinatorEndpoint"}
{"@timestamp":"2024-11-20T11:45:48.602Z","ecs.version":"1.2.0","log.level":"INFO","message":"Successfully stopped SparkContext","process.thread.name":"main","log.logger":"org.apache.spark.SparkContext"}
{"@timestamp":"2024-11-20T11:45:48.604Z","ecs.version":"1.2.0","log.level":"INFO","message":"Shutdown hook called","process.thread.name":"shutdown-hook-0","log.logger":"org.apache.spark.util.ShutdownHookManager"}
{"@timestamp":"2024-11-20T11:45:48.604Z","ecs.version":"1.2.0","log.level":"INFO","message":"Deleting directory /var/data/spark-f783cf2e-44db-452c-83c9-738f9c894ef9/spark-2caa5814-bd32-431c-a9f9-a32208b34fbb","process.thread.name":"shutdown-hook-0","log.logger":"org.apache.spark.util.ShutdownHookManager"}
{"@timestamp":"2024-11-20T11:45:48.606Z","ecs.version":"1.2.0","log.level":"INFO","message":"Deleting directory /tmp/spark-dacdfd95-f166-4b23-9312-af9052730417","process.thread.name":"shutdown-hook-0","log.logger":"org.apache.spark.util.ShutdownHookManager"}

ログは JSONL 形式で出力されます。各フィールドの説明は次のとおりです：

@timestamp：ログエントリが生成された時刻。
ecs.version：Elastic Common Schema (ECS) のバージョン。ECS はログデータの標準化されたスキーマです。
log.level：ログレベル。
message：ログメッセージ。
process.thread.name：ログエントリを生成したスレッドの名前。
log.logger：ログエントリを記録したロガーの名前。

手順5：Spark ログのクエリと分析

「ログのクエリと分析のクイックスタートガイド」を参照して、ジョブの実行時間範囲を指定し、ログが正常に収集されたことを確認できます。

SLS コンソールのクエリページで、spark Logstore を選択し、時間範囲を設定してからクエリを実行します。[生ログ] タブで、収集された Spark アプリケーションのログを表示できます。起動フェーズの WARN レベルの Unable to load native-hadoop library 警告や、Spark バージョン (3.5.3)、オペレーティングシステム、Java バージョンに関する INFO レベルのレコードなどのログエントリで、ログが正常に収集されていることを確認できます。

(オプション) 手順6：クリーンアップ

このチュートリアルが完了し、リソースが不要になった場合は、次のコマンドを実行して削除します。

次のコマンドを実行して Spark ジョブを削除します。

kubectl delete -f spark-pi.yaml

次のコマンドを実行して Logtail 設定を削除します。

kubectl delete -f aliyun-log-config.yaml

次のコマンドを実行して Log4j2 ログ設定を削除します。

kubectl delete -f spark-log-conf.yaml