pandasの処理をPysparkで

#1.図
Apache Arrowのサイトに載っている図。
>[参考](https://arrow.apache.org/)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385220/24fc8c78-3597-764d-95c7-7001137892eb.png)

図をみるとpandasがApache Arrowを経由してsparkというものにつながっている。
調べてみるとarrrowはいちいちcsvなどでデータ処理ツール間を経由させずに高速にデータを動かすことができる中間体みたいなものを提供し、sparkは分散処理に強いツールということでした。
特許データは集まるとかなり重いので、これはぜひ使ってみたい。ということで試してみることに

#2.colaboratoryで実現
早速colaboratoryで実施できるようにしよう、ということで、[ここ](https://qiita.com/calderarie/items/1b50b358082f4c163d5f)を参照させてもらいつつ作成。

    # 各ライブラリインストール
    !pip install pyarrow
    !apt-get install openjdk-8-jdk-headless -qq > /dev/null
    !wget -q http://apache.osuosl.org/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz
    !tar xf spark-2.4.4-bin-hadoop2.7.tgz
    !pip install -q findspark

    # 環境変数設定
    import os
    os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
    os.environ["SPARK_HOME"] = "/content/spark-2.4.0-bin-hadoop2.7"

    # findsparkで環境設定
    import findspark
    findspark.init("/content/spark-2.4.4-bin-hadoop2.7")
    from pyspark.sql import SparkSession
    spark = SparkSession.builder.master("local[*]").getOrCreate()
    spark.conf.set("spark.sql.execution.arrow.enabled", "true")

↑これを実行しておけばたぶんcolaboratoryでsparkが使えて、かつpyarrowも使える。
　（spark.conf.set("spark.sql.execution.arrow.enabled", "true")の部分）

##　読み込み
    spark_df = spark.read.csv('~.csv', header=True,encoding="sjis")

encoding="sjis"を使えば日本語も大丈夫問題なく読み込んでくれる。cp932じゃないのね。
が、parquetでsaveしようとするとエラーが出るのでなんとかしないと。調べると列名が問題のよう。

    spark_df.write.save("test.parquet", format="parquet")

##少し試し
たとえば

    from pyspark.sql.functions import desc
    spark_df.groupby("発明の名称").count().sort(desc("count")).show()

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/385220/f15f7552-181b-6b08-16aa-1dedfa35f8e6.png)

おお、ちゃんと出てくれる！SQLを叩いた後の結果のようでpandasとは違うんだなあと実感。



#3.実験
[ここ](https://spark.apache.org/docs/latest/api/python/pyspark.sql.html)を見ると結構ある。pandasの操作とsparkの操作を対比。
pandasっぽくできる部分と気を付けないといけない部分に注意。

import pandas as pd
spark = SparkSession.builder.master("local[*]").getOrCreate()
として

|操作 |pandas  |spark  |
|---|---|---|
|csv読み込み|pd.read_csv()|spark.read.csv|
|excel読み込み|pd.read_excel()|無|
|表示させる  |df  |spark_df.show()<br>.show()をつけないと表示されないので焦る。|
|先頭5行|df.head(5)|spark_df.show(5)<br>substrの代わりに[0:4]もあり。|
|列追加(一部抽出で)|df['newcolname']=df['colname'].str[0,4]|spark_df.withColumn('newcolname', spark_df['colname'].substr(0,4)).show(5)|
|rename|df.rename(columns={~~:**})|spark_df.withColumnRenamed('~~', '**')|
|groupby  |df.groupby()  |    from pyspark.sql.functions import desc<br>spark_df.groupby('~').count().sort(desc("count")).show()|
|groupby⇒agg|df.groupby(FUN)|spark_df.groupby("A").agg(FUN).show()|
|列抽出|df[['A','B','D']]|spark_df.select('A','B','D').show()|
|sort|df.sort_values(by='A')|spark_df.sort("A")|
|crosstab|pd.crosstab(df['A'],df['B'])|spark_df.crosstab('A','B').show()|
|UDF|df.apply(FUN)|from pyspark.sql.functions import pandas_udf, PandasUDFType <br> :pandas_udf("id long, v double", PandasUDFType.GROUPED_MAP) <br> def normalize(pdf): <br> v = pdf.v <br> return pdf.assign(v=(v - v.mean()) / v.std()) <br> df.groupby("id").apply(normalize).show()  （作成中・・・） |
非常に多かったので、pandasで出来ることは大体実現できるのではという感触。
user defined function (UDF)も色々あったので追加しないと。
日付系もあるようなので、整理して後日追加予定。


#2020/5/7追加
pandasとpysparkの間を取り持つライブラリを発見。その名もコアラ。
pandaslikeな操作感でpysparkを操作できる！
[koalas](https://github.com/databricks/koalas)