技術的な話

pyspark-notebookでPySparkスクリプトを動かす

公式Dockerイメージの場合、Notebookを使用すればPySparkを実行出来るので問題無いです。

今回は裏でバッチ処理的なものを動作させたかったのでその際の手順を記載しています。

PySparkをイチからインストールして設定するのも手間なので意外と便利です。

使用したイメージ

jupyter/pyspark-notebook

Sparkのバージョンは3.3.0を使用しました。

https://hub.docker.com/r/jupyter/pyspark-notebook

各種設定

Dockerfile設定内容

必要なソフトウェアのインストールやPySparkログ設定ファイルや必要となるPythonライブラリをインストールしています。

log4j2.properties及びrequirements.txtは別途用意してください。

FROM jupyter/pyspark-notebook:spark-3.3.0

USER root

# Environments
ENV LANG=C.UTF-8
ENV TZ=Azia/Tokyo

# requirements install
RUN apt-get update && \
    apt-get install -y curl

# pyspark logging setting
COPY ./log4j.properties /usr/local/spark/conf/log4j2.properties

# clean
RUN apt-get clean && \
    rm -rf /var/lib/apt/lists/*

USER jovyan

# python requirements install
COPY ./requirements.txt .
RUN pip install -r requirements.txt

WORKDIR /home/jovyan/work

docker-compose設定内容

composeで諸々管理したいので以下ファイルの通り設定しています。

ポート番号は自由に変更してください。

起動時のコマンドも必要に応じて変更してください。

version: '3.7'

services:
  pyspark:
    container_name: pyspark
    build:
      context: .
      dockerfile: ./Dockerfile
    ports:
      - 8888:8888
    environment:
      - JUPYTER_ENABLE_LAB=yes
    restart: unless-stopped
    tty: true
    command: >
      bash -c "
      start-notebook.sh --NotebookApp.token=""
      "
    volumes:
      - type: bind
        source: ./app/
        target: /home/jovyan/work

ビルド及びコンテナ起動等は下記コマンドにて実行します。

# build
$ sudo docker-compose build

# launch container
$ sudo docker-compose up -d

# enter container
$ sudo docker-compose exec pyspark bash

PySparkのログ設定

デフォルトの状態ですとPySparkを実行するとデフォルトのログが大量表示されて確認がし辛いです。

なので設定を変更しておきます。

デフォルトのlog4j2設定ファイルはコンテナの下記に配置されていますのでコピーして使用します。

/usr/local/spark/conf/log4j2.properties.template

ファイル内容は下記の通りとなっています。

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

# Set everything to be logged to the console
# rootLogger.level = info
rootLogger.level = warn
rootLogger.appenderRef.stdout.ref = console

（以下省略）

rootLogger.level = warnに変更してinfoログが表示されないように設定しています。

また、Dockerfile内にてファイルをコンテナ内にコピーしています。

スクリプトを作成

バッチ処理用のPySpark向けスクリプトを作成します。

以下サンプルスクリプトになります。適当なsleepを実行する関数を噛ませて実行しています。

from time import sleep
from datetime import datetime

import pyspark
from pyspark.sql import SparkSession


def main() -> None:
    spark = SparkSession.builder.appName('pyspark.sample').getOrCreate()
    sparkContext=spark.sparkContext
    
    rdd = sparkContext.parallelize(range(10))
    mapped_rdd = rdd.map(lambda value: heavy_task(value))
    result = mapped_rdd.collect()

def heavy_task(num:int) -> None:
    print(f'{datetime.now()} -> {num}')
    sleep(1)

if __name__ == '__main__':
    main()

スクリプトを実行

通常のPythonスクリプトの要領でスクリプトを実行すると下記のようなエラーが表示されます。

PySparkが無いと言われます。まぁそうですね…。

(base) jovyan@25a3dbf9faa6:~/work$ python main.py 
Traceback (most recent call last):
  File "/home/jovyan/work/main.py", line 4, in <module>
    import pyspark
ModuleNotFoundError: No module named 'pyspark'

同じ要領でpysparkコマンド経由で実行すると下記エラーが表示されます。

Pythonスクリプトを実行するには./bin/spark-submit <python file>と言われます。

(base) jovyan@25a3dbf9faa6:~/work$ pyspark main.py 
Running python applications through 'pyspark' is not supported as of Spark 2.0.
Use ./bin/spark-submit <python file>
/usr/local/spark/bin/spark-class: line 96: CMD: bad array subscript

コマンドを変更することで無事に実行できました。

後はこのコマンドをcron等に登録すればバッチ処理化出来ます。

(base) jovyan@25a3dbf9faa6:~/work$ spark-submit main.py 
22/09/05 15:10:03 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2022-09-05 15:10:05.110540 -> 5
2022-09-05 15:10:05.115504 -> 0
2022-09-05 15:10:06.112646 -> 6
2022-09-05 15:10:06.117193 -> 1
2022-09-05 15:10:07.114014 -> 7
2022-09-05 15:10:07.118828 -> 2
2022-09-05 15:10:08.115484 -> 8
2022-09-05 15:10:08.120616 -> 3
2022-09-05 15:10:09.116871 -> 9
2022-09-05 15:10:09.122860 -> 4

-技術的な話
-, ,