■ はじめに

https://dk521123.hatenablog.com/entry/2019/11/14/221126
https://dk521123.hatenablog.com/entry/2021/04/03/004254

の続き。
今回は、PySpark で CSV を扱う。

補足

エスケープされた区切り文字が含んだデータを扱う場合は
以下の関連記事を参照のこと。

PySpark ～エスケープされた区切り文字が含んだデータを扱う～
https://dk521123.hatenablog.com/entry/2020/07/09/000832

【１】関連するAPI
　１）読み込み
　２）書き込み
【２】補足
　１）読み込み時でファイルがなかった場合の挙動
【３】サンプル
　例１：CSVファイルの読み込み
　例２：CSVファイルの書き込み
　例３：distinct 重複値の削除

【１】関連するAPI

１）読み込み

spark.read.csv()
https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=read%20csv

２）書き込み

data_frame.write.mode('書き込みモード').csv()
https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrameWriter

* mode : 書き込みモード（ファイルがあった場合の振る舞いを指定）
  + append : 追記するモード
  + overwrite : 上書きするモード
  + ignore : 無視するモード（何もしない？）
  + error or errorifexists : 例外を発生させるモード（デフォルト）

* それ以外のオプションについては、以下の関連記事を参照のこと

PySpark ～ CSV / Read/Writeのプロパティ～
https://dk521123.hatenablog.com/entry/2020/07/30/195226

【２】補足

１）読み込み時でファイルがなかった場合の挙動

ファイルが存在しない場合（アクセス権がない場合も）、以下の例外が発生する
=====
pyspark.sql.utils.AnalysisException: 'Path does not exist: file:/C:/xxx/xxx/xxxx.csv;'
=====

サンプル

from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.utils import AnalysisException


spark_context = SparkContext()
spark = SparkSession(spark_context)

try:
  data_frame = spark.read.csv("dummy.csv", header=False)
  data_frame.show()
except AnalysisException as ex:
  print("AnalysisException = " + str(ex))
  raise ex
except Exception as ex:
  print("Exception = " + str(ex))
  raise ex

【３】サンプル

例１：CSVファイルの読み込み

from pyspark import SparkContext
from pyspark.sql import SparkSession

def main():
  spark_context = SparkContext()
  spark = SparkSession(spark_context)

  # ★ここが違う点★
  data_frame = spark.read.csv("hello.csv", header=True)
  data_frame.show()

if __name__ == '__main__':
  main()

入力ファイル：hello.csv

id,name,remarks
001,Mike,-
002,Tom,Hello
003,Smith,World
004,Nick,!!

出力結果

+---+-----+-------+
| id| name|remarks|
+---+-----+-------+
|001| Mike|      -|
|002|  Tom|  Hello|
|003|Smith|  World|
|004| Nick|     !!|
+---+-----+-------+

例２：CSVファイルの書き込み

from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType
from pyspark.sql.types import StructField
from pyspark.sql.types import StringType

spark_context = SparkContext()
spark = SparkSession(spark_context)

data_frame = spark.read.csv(
  "./hello_without_header.csv")
# StructFiled([項目名], [データ型], [Nullを許容するか（True：許容する）])
schema = StructType([
  StructField('id', StringType(), True),
  StructField('name', StringType(), True)
])
data_frame = spark.createDataFrame(
  data_frame.rdd, schema)
# 書き込み
data_frame.write.mode('overwrite').csv(
  "./out",
  header=True,
  quoteAll=True)

data_frame.show()

print("Done")

入力ファイル：hello_without_header.csv

1,Mike
2,Tom
3,Naomi

出力ファイル：out/part-xxxxx-xxxxx-xxxx-xxxx-xxx-xxxxxxxxxxx-xxxx.csv

"id","name"
"1","Mike"
"2","Tom"
"3","Naomi"

例３：distinct 重複値の削除

from pyspark import SparkContext
from pyspark.sql import SparkSession

def main():
  spark_context = SparkContext()
  spark = SparkSession(spark_context)

  data_frame = spark.read.csv("hello.csv", header=True)
  # 重複値の削除
  count_for_id = data_frame.select("id").distinct().count()
  print(count_for_id)

if __name__ == '__main__':
  main()

入力ファイル：hello.csv

id,name,remarks
001,Mike,-
002,Tom,Hello
003,Smith,World
001,Mike,!!
004,Nick,!!
004,Nick,!!
003,Smith,!!
003,Smith,!!
005,Ken,!!

出力結果

参考文献

https://fisproject.jp/2018/04/methods-of-the-pyspark-sql-dataframe-class/
https://qiita.com/taka4sato/items/4ab2cf9e941599f1c0ca
CSVの書き込み
https://qiita.com/paulxll/items/1c0833782cd4e1de86e2
今後役に立ちそうなサイト
https://www.atmarkit.co.jp/ait/articles/0902/27/news129.html
https://www.atmarkit.co.jp/ait/articles/0903/09/news094.html

プログラムの超個人的なメモ

Memo for Programming.

【分散処理】PySpark ～ CSV / 基本編～