SparkによるAirbnbデータセットのクレンジング

2024/4/12に翔泳社より**Apache Spark徹底入門**を出版します！

<a href="https://www.amazon.co.jp/dp/4798182281/"><img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/87910d7c-8a63-c198-8e82-216ea7a8ee07.jpeg" width=300></a>

書籍の[サンプルノートブック](https://github.com/databricks/LearningSparkV2/blob/master/notebooks/LearningSparkv2.dbc)をウォークスルーしていきます。`Python/Chapter10/10-1 Data Cleansing`となります。

翻訳ノートブックのリポジトリはこちら。

https://github.com/taka-yayoi/SparkSample/tree/main

ノートブックはこちら

https://github.com/taka-yayoi/SparkSample/blob/main/Chapter10/10-1%20Data%20Cleansing.py

いくつかの探索的データ分析やクレンジングからスタートします。[Inside Airbnb](http://insideairbnb.com/get-the-data.html)のSF Airbnbレンタルデータセットを使用します。

SF Airbnbデータセットをロードしましょう(オプションの動作を確認したい場合にはコメントアウトしてください)。

```py
filePath = "/databricks-datasets/learning-spark-v2/sf-airbnb/sf-airbnb.csv"

rawDF = spark.read.csv(filePath, header="true", inferSchema="true", multiLine="true", escape='"')

display(rawDF)
```
![Screenshot 2024-03-27 at 17.38.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f36f7f2d-c91d-7772-1343-645a54d89fab.png)

```py
rawDF.columns
```
```
['id',
 'listing_url',
 'scrape_id',
 'last_scraped',
 'name',
 'summary',
 'space',
 'description',
 'experiences_offered',
 'neighborhood_overview',
 'notes',
 'transit',
 'access',
 'interaction',
 'house_rules',
 'thumbnail_url',
 'medium_url',
 'picture_url',
 'xl_picture_url',
 'host_id',
 'host_url',
 'host_name',
 'host_since',
 'host_location',
 'host_about',
 'host_response_time',
 'host_response_rate',
 'host_acceptance_rate',
 'host_is_superhost',
 'host_thumbnail_url',
 'host_picture_url',
 'host_neighbourhood',
 'host_listings_count',
 'host_total_listings_count',
 'host_verifications',
 'host_has_profile_pic',
 'host_identity_verified',
 'street',
 'neighbourhood',
 'neighbourhood_cleansed',
 'neighbourhood_group_cleansed',
 'city',
 'state',
 'zipcode',
 'market',
 'smart_location',
 'country_code',
 'country',
 'latitude',
 'longitude',
 'is_location_exact',
 'property_type',
 'room_type',
 'accommodates',
 'bathrooms',
 'bedrooms',
 'beds',
 'bed_type',
 'amenities',
 'square_feet',
 'price',
 'weekly_price',
 'monthly_price',
 'security_deposit',
 'cleaning_fee',
 'guests_included',
 'extra_people',
 'minimum_nights',
 'maximum_nights',
 'minimum_minimum_nights',
 'maximum_minimum_nights',
 'minimum_maximum_nights',
 'maximum_maximum_nights',
 'minimum_nights_avg_ntm',
 'maximum_nights_avg_ntm',
 'calendar_updated',
 'has_availability',
 'availability_30',
 'availability_60',
 'availability_90',
 'availability_365',
 'calendar_last_scraped',
 'number_of_reviews',
 'number_of_reviews_ltm',
 'first_review',
 'last_review',
 'review_scores_rating',
 'review_scores_accuracy',
 'review_scores_cleanliness',
 'review_scores_checkin',
 'review_scores_communication',
 'review_scores_location',
 'review_scores_value',
 'requires_license',
 'license',
 'jurisdiction_names',
 'instant_bookable',
 'is_business_travel_ready',
 'cancellation_policy',
 'require_guest_profile_picture',
 'require_guest_phone_verification',
 'calculated_host_listings_count',
 'calculated_host_listings_count_entire_homes',
 'calculated_host_listings_count_private_rooms',
 'calculated_host_listings_count_shared_rooms',
 'reviews_per_month']
 ```

シンプルにするために、このデータセットの特定のカラムのみを保持します。あとで、特徴量選択について言及します。

```py
columnsToKeep = [
  "host_is_superhost",
  "cancellation_policy",
  "instant_bookable",
  "host_total_listings_count",
  "neighbourhood_cleansed",
  "latitude",
  "longitude",
  "property_type",
  "room_type",
  "accommodates",
  "bathrooms",
  "bedrooms",
  "beds",
  "bed_type",
  "minimum_nights",
  "number_of_reviews",
  "review_scores_rating",
  "review_scores_accuracy",
  "review_scores_cleanliness",
  "review_scores_checkin",
  "review_scores_communication",
  "review_scores_location",
  "review_scores_value",
  "price"]

baseDF = rawDF.select(columnsToKeep)
baseDF.cache().count()
display(baseDF)
```
![Screenshot 2024-03-27 at 17.39.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/66d12328-ef04-7add-8a3a-9c8741f72b85.png)

# データ型の修正

上のスキーマを確認してみます。`price`フィールドが文字列として判定されていることがわかります。我々のタスクでは、これを数値型のフィールド(double型)にする必要があります。

修正しましょう。

```py
from pyspark.sql.functions import col, translate

fixedPriceDF = baseDF.withColumn("price", translate(col("price"), "$,", "").cast("double"))

display(fixedPriceDF)
```
![Screenshot 2024-03-27 at 17.39.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cb3971b2-4c0b-b6a1-887b-fcf71b38affa.png)

# サマリー統計情報

2つのオプションがあります:
* describe
* summary (describe + IQR)

```py
display(fixedPriceDF.describe())
```
![Screenshot 2024-03-27 at 17.40.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/639e2f03-89fd-ecb5-b1ed-1e5aeead3eab.png)

```py
display(fixedPriceDF.summary())
```
![Screenshot 2024-03-27 at 17.40.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3b69008e-ec77-2404-a44b-0892ac6c8bcb.png)

# Null値

Null値を取り扱うための数多くの手法があります。時には、nullは実際には予測しようとする事柄のキーのインジケータとなることがあります(例: フォームの特定の割合を記入しない場合、承認される確率が減少する)。

nullを取り扱う方法として:
* nullを含むすべてのレコードを削除
* 数値型:
  * mean/median/zeroなどで補完
* カテゴリー型:
  * モードで置換
  * nullに対する特殊なカテゴリーを作成
* 欠損地を挿入するために設計されたALSのようなテクニックを使う

**カテゴリー特徴量/数値特徴量に何かしらの補完テクニックを用いる場合、当該フィールドが保管されたことを示す追加フィールドを含めるべきです(なぜこれが必要なのかを考えてみてください)**

カテゴリー特徴量`host_is_superhost`にいくつかnullが含まれています。これらのカラムのいずれかがnullである行を除外しましょう。

SparkMLのImputer(この後カバーします)は、カテゴリー特徴量の補完をサポートしていませんので、この時点ではこれがもっともシンプルなアプローチとなります。

```py
noNullsDF = fixedPriceDF.na.drop(subset=["host_is_superhost"])
```

# 補完: Doubleへのキャスト

SparkMLの`Imputer`は、すべてのフィールドがdouble型である必要があります [Python](https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.feature.Imputer)/[Scala](https://spark.apache.org/docs/latest/api/scala/#org.apache.spark.ml.feature.Imputer)。すべてのintegerフィールドをdoubleにキャストしましょう。

```py
from pyspark.sql.functions import col
from pyspark.sql.types import IntegerType

integerColumns = [x.name for x in baseDF.schema.fields if x.dataType == IntegerType()]
doublesDF = noNullsDF

for c in integerColumns:
  doublesDF = doublesDF.withColumn(c, col(c).cast("double"))

columns = "\n - ".join(integerColumns)
print(f"Columns converted from Integer to Double:\n - {columns}")
```
```
Columns converted from Integer to Double:
 - host_total_listings_count
 - accommodates
 - bedrooms
 - beds
 - minimum_nights
 - number_of_reviews
 - review_scores_rating
 - review_scores_accuracy
 - review_scores_cleanliness
 - review_scores_checkin
 - review_scores_communication
 - review_scores_location
 - review_scores_value
```

値を補完したかどうかを示すダミー変数を追加します。

```py
from pyspark.sql.functions import when

imputeCols = [
  "bedrooms",
  "bathrooms",
  "beds", 
  "review_scores_rating",
  "review_scores_accuracy",
  "review_scores_cleanliness",
  "review_scores_checkin",
  "review_scores_communication",
  "review_scores_location",
  "review_scores_value"
]

for c in imputeCols:
  doublesDF = doublesDF.withColumn(c + "_na", when(col(c).isNull(), 1.0).otherwise(0.0))
```
```py
display(doublesDF.describe())
```
![Screenshot 2024-03-27 at 17.42.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1ac97529-3443-73ed-c3e5-cd2497311f97.png)

```py
from pyspark.ml.feature import Imputer

imputer = Imputer(strategy="median", inputCols=imputeCols, outputCols=imputeCols)

imputedDF = imputer.fit(doublesDF).transform(doublesDF)
```

## 外れ値を排除

`price`カラムの *min* と *max* の値を見てみましょう:

```py
display(imputedDF.select("price").describe())
```
![Screenshot 2024-03-27 at 17.43.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8e30fce0-8460-644f-d3e4-5744f8755f54.png)

いくつか非常に高価な物件があります。しかし、これらに対して何をすべきかを決めるのはデータサイエンティストの仕事です。しかし、「無料」のAirbnbはフィルタリングします。

*price*がゼロである物件がいくつあるのかを見てみましょう。

```py
imputedDF.filter(col("price") == 0).count()
```
```
1
```

厳密に正の*price*を持つ行のみを保持します。

```py
posPricesDF = imputedDF.filter(col("price") > 0)
```

*minimum_nights*カラムの *min* と *max* を見てみましょう:

```py
display(posPricesDF.select("minimum_nights").describe())
```
![Screenshot 2024-03-27 at 17.45.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d249f3bc-17e1-eebb-25e6-4edb166fa61f.png)

```py
display(posPricesDF
  .groupBy("minimum_nights").count()
  .orderBy(col("count").desc(), col("minimum_nights"))
)
```
![Screenshot 2024-03-27 at 17.45.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b03a3e9d-fe0d-0f3b-184f-f1c124e78ea1.png)

1年という最小滞在期間は合理的な制限のように見えます。*minimum_nights*が365を上回るレコードを除外しましょう:

```py
cleanDF = posPricesDF.filter(col("minimum_nights") <= 365)

display(cleanDF)
```
![Screenshot 2024-03-27 at 17.45.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7eb87f6a-962f-5efe-4c68-b42fc522da7e.png)

OK、データが綺麗になりました。これを用いてモデルの構築をスタートできるように、このデータフレームをファイルに保存しましょう。

```py
outputPath = "/tmp/sf-airbnb/sf-airbnb-clean.parquet"

cleanDF.write.mode("overwrite").parquet(outputPath)
```

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)