PythonでParquetを扱う #Python

始めに

私の所属する内製チームではユニケージからの移行を進めており、テキストファイルの大規模トランザクションデータをユニケージコマンド以外の方法でどう扱うかが課題になっております。

以前pandasでユニケージファイルの取り扱いを試してみたところ、100万行レベルになるとユニケージコマンドよりも圧倒的に遅いという結果になってしまいました。その後parquetに変換してはどうかという話があり、早速試してみることにしました。

環境

MacOSX 10.15.4
Python3.8.2

pysparkのインストール

pipenv install pyspark

まずpysparkでそのまま読み込み

実際に使われている89万行のデータを用意して先頭5行を表示します

from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder.appName('SparkSample').getOrCreate()

# ユニケージ形式なのでスペース区切りとして読み込む
df = spark.read.csv('./SAMPLE_DATA', header=True, sep=' ')
df.show(5)

これだけで30秒以上かかってます。思ったより遅いです。読み込み方を工夫すればもっと高速になるのかもしれません。

ユニケージ形式ファイルをparquet形式に変換

$ pyspark

(中略)

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.0.1
      /_/

Using Python version 3.8.2 (default, Aug 25 2020 09:23:57)
SparkSession available as 'spark'.
>>> df = spark.read.csv('./SAMPLE_DATA', header=True, sep=' ')
>>> df.write.save('./SAMPLE_DATA.parquet')

pyspark でpython環境が立ち上がるので再度読み込んでから、今度はparquet形式で保存します。

 $ ls SAMPLE_DATA.parquet/
_SUCCESS
part-00000-1a94dff0-2de3-4ee8-acf0-667643aff8b7-c000.snappy.parquet
part-00001-1a94dff0-2de3-4ee8-acf0-667643aff8b7-c000.snappy.parquet
part-00002-1a94dff0-2de3-4ee8-acf0-667643aff8b7-c000.snappy.parquet
part-00003-1a94dff0-2de3-4ee8-acf0-667643aff8b7-c000.snappy.parquet

フォルダが作成され、フォルダ配下にparquetファイルができています。

from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder.appName('SparkSample').getOrCreate()
df = spark.read.parquet('./SAMPLE_DATA.parquet')
df.show(5)

parquet形式で読み込みましたがさっきと同じように30秒以上かかりました。ファイル行数を10行にしても同等の時間がかかっており、pysparkの起動にかなり時間がかかるようです。

$ python sample.py 
20/12/24 00:50:56 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).

スクリプトの起動時のメッセージを見るとネイティブライブラリを使えていない…それで遅いのかもしれません。

調べたのですが簡単に解決しそうにないので、残念ですが飛ばします。

参照したり更新してみる

from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder.appName('SparkSample').getOrCreate()

df = spark.read.parquet('./SAMPLE_DATA.parquet')

# SQLで絞り込み
df.createOrReplaceTempView('sample_data')
df = spark.sql("select SCAN_CODE,BUNRUI2_CODE from sample_data where BUNRUI2_CODE = '29' limit 5").show()

SQLで参照できました。なおparquetの作成時に特にスキーマを設定していないので全てSTRING型になっています。数字データを扱うときは注意が必要です。

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *

spark = SparkSession.builder.appName('SparkSample').getOrCreate()

df = spark.read.parquet('./SAMPLE_DATA.parquet',)

# 条件分岐させて値を更新
new_df = df.withColumn('BUNRUI2_CODE', F.when(F.col('BUNRUI2_CODE') == '29','9999').otherwise(F.col('BUNRUI2_CODE')))

SQLで更新しようとしましたが、SparkSQLはUpdate文がない模様。新しい値の新しい列を作ることで更新できました。

new_df.write.save('./SAMPLE_DATA.new.parquet')

更新して保存。

parquet-tool のインストール

parquetファイルを閲覧するためにparquet-toolsをインストール。

$ brew install parquet-tools

$ parquet-tools cat --json SAMPLE_DATA.parquet/part-00000-1a94dff0-2de3-4ee8-acf0-667643aff8b7-c000.snappy.parquet

csv形式で確認できないのが辛いですね…。

終わりに

今回は初歩的な内容でした。今後はAWS Glueなどの組み合わせも視野に進めていきたいと思います。参照用途であれば、RDSやDynamoDBに格納するよりもparquetにしてS3に置けばコスト的にも好ましいです。

発表されたばかりの AWS Glue Elastic Viewsもユニケージ移行において有力なサービスとして社内で期待が集まっており、こちらも検証を進めていきたいと考えています。

PythonでParquetを扱う

始めに

環境