【分散処理】PySpark ~ CSV / Read/Writeのプロパティ ~

■ はじめに

https://dk521123.hatenablog.com/entry/2019/11/24/225534
https://dk521123.hatenablog.com/entry/2020/07/09/000832
https://dk521123.hatenablog.com/entry/2020/11/23/224349

の続き。

 今回は、PySpark において、
CSV の Read/Writeのプロパティ について扱う。

目次

【1】CSVの読み込み
 1)読み込みモード:mode 
 2)エスケープについて:escape

【2】CSVの書き込み
 1)書き込みモード:mode
 2)区切り文字の指定:sep
 3)囲み文字:quote, quoteAll
 4)エスケープについて:escape, escapeQuotes
 5)ホワイトスペースに関する振る舞い:ignoreLeadingWhiteSpace, ignoreTrailingWhiteSpace

【1】CSVの読み込み

* spark.read.csv() を使う

spark.read.csv()
https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrameReader.csv.html

DataFrameReader.csv(
path, schema=None, sep=None, encoding=None,
quote=None, escape=None, comment=None,
header=None, inferSchema=None,
ignoreLeadingWhiteSpace=None, ignoreTrailingWhiteSpace=None,
nullValue=None, nanValue=None, positiveInf=None, negativeInf=None,
dateFormat=None, timestampFormat=None,
maxColumns=None, maxCharsPerColumn=None, maxMalformedLogPerPartition=None,
mode=None, columnNameOfCorruptRecord=None, multiLine=None,
charToEscapeQuoteEscaping=None, samplingRatio=None,
enforceSchema=None, emptyValue=None, locale=None,
lineSep=None, pathGlobFilter=None, recursiveFileLookup=None,
modifiedBefore=None, modifiedAfter=None, unescapedQuoteHandling=None)

1)読み込みモード:mode

* 以下、読み込みモード。

PERMISSIVE

とりあえず取り込むモード(デフォルト)(※)

※
ただし、取り込めても、count() などのメソッドを呼ぶと
例外が発生する可能性がある
例外については、以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2020/07/10/192404

DROPMALFORMED

データが破損していた場合、無視するモード

FAILFAST

例外を発生させるモード

2)エスケープについて:escape

* エスケープに使う文字を指定
* デフォルトは、「\」
 ⇒ Noneを指定しても「\」が指定されることに注意
  (エスケープしないとかの指定はできないっぽい)
* escape を使ったサンプルについては、以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2020/11/23/224349

【2】CSVの書き込み

* data_frame.write.mode('書き込みモード').csv() を使う

spark.write.csv()
https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrameWriter.csv.html

DataFrameWriter.csv(
path, mode=None, compression=None, sep=None,
quote=None, escape=None, header=None, nullValue=None,
escapeQuotes=None, quoteAll=None, dateFormat=None,
timestampFormat=None,
ignoreLeadingWhiteSpace=None, ignoreTrailingWhiteSpace=None,
charToEscapeQuoteEscaping=None, encoding=None,
emptyValue=None, lineSep=None)

1)書き込みモード:mode

* ファイルがあった場合の振る舞いを指定
* 以下、書き込みモード。

append

追記するモード

overwrite

上書きするモード

※ パーティション付のテーブルに対して、上書きする場合は注意が必要。
 => 詳細は、以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2021/07/07/093147

ignore

無視するモード(何もしない?)

error / errorifexists

例外を発生させるモード(デフォルト)

2)区切り文字の指定:sep

* 区切り文字(separator)の指定(一文字または文字列可能)
* default value = ","

3)囲み文字について

quote

* 囲み文字(一文字指定)の指定(default:")

quoteAll

* 全ての値を囲み文字で囲うかどうか(True/False)

4)エスケープについて

escape

* もし囲み文字があった場合のエスケープ文字(default:\)

escapeQuotes

* もし囲み文字があった場合、エスケープするかどうか(default:True)

5)ホワイトスペースに関する振る舞い

* 詳しくは、以下の関連記事を参照のこと。

ignoreLeadingWhiteSpace

* 前の White Space を書き込むかどうか(default:True)
 => True : 前の White Space を無視して、そのまま書き込む
 => False : 前の White Space をトリムする

ignoreTrailingWhiteSpace

* 後ろの White Space を書き込むかどうか(default:True)
 => True : 後ろの White Space を無視して、そのまま書き込む
 => False : 後ろの White Space をトリムする

関連記事

PySpark ~ 環境設定 ~
https://dk521123.hatenablog.com/entry/2019/11/14/221126
PySpark ~ CSV / 基本編 ~
https://dk521123.hatenablog.com/entry/2019/11/24/225534
PySpark ~ CSV / White Spaceの扱い ~
https://dk521123.hatenablog.com/entry/2021/04/29/075903
PySpark ~ CSV / Null・空文字・異常値の扱い ~
https://dk521123.hatenablog.com/entry/2021/06/01/142457
PySpark ~ CSV / escape ~
https://dk521123.hatenablog.com/entry/2020/11/23/224349
PySpark ~ パーティション単位で上書きするには ~
https://dk521123.hatenablog.com/entry/2021/07/07/093147
PySpark ~ 出力ファイル / 空ファイル対応, 1ファイルに纏める ~
https://dk521123.hatenablog.com/entry/2021/04/03/004254
PySpark ~ エスケープされた区切り文字が含んだデータを扱う ~
https://dk521123.hatenablog.com/entry/2020/07/09/000832
PySpark でエラー「Exception: It appears ...」が表示された
https://dk521123.hatenablog.com/entry/2020/07/10/192404