PySparkとPandasを使用してRSSの頻出ワードを抽出

## はじめに
本記事では、Pythonを使って、データの加工と分析の流れを簡易的にハンズオンします。

お急ぎの方は次のGoogle Colaboratoryのリンクからコードを試すことができます。

https://colab.research.google.com/drive/1b6wCYm-y4GrKpMN0LQ_WarNx6Wiadk60?usp=sharing

## やりたいこと
技術評論社のRSSからSQLクエリを使って、最も使用頻度の高いワードを使った記事のタイトルとリンクを取得する。

**ステップ**
1. 技術評論社のRSS（XML形式）をJSON形式へ変換
2. JSONからRSSのtitleタグを取得し、タイトル一覧をtxtファイルに出力
2. タイトル一覧.txtから使用頻度の高いワードを抽出
2. JSONファイルとtitle情報からデータ分析・取得を行う


![スクリーンショット 2023-12-09 22.52.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1650886/1fca2ec2-6d80-b242-eadd-10e688dc7200.png)


### PandasとPySpark
今回使うPythonのパッケージにはPandasとPySparkがあります。コードの紹介の前に、まずは二つのパッケージの概要について説明しましょう。

#### Pandas
Pandasは、データ分析および機械学習タスクに最も広く使用されているオープンソースPythonパッケージです。Pandas を使用すると、「データクレンジング」「データ分析」「統計分析」など、時間のかかる反復的なタスクを簡単に行うことができます。

ちなみに”Pandas”という名前は「Panel Data」から来ているそうです。


#### PySpark
PySparkは「Apache Spark」のPython APIです。
Apache Sparkは、大規模なデータ処理のための統合分析エンジンです。
Apache Sparkの機能の一つに構造化データ処理用の「Spark SQL」があります。
Spark SQLを使うことで、SQLの書き方でデータ分析を行えるようになります。

![スクリーンショット 2023-12-09 22.25.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1650886/80a7b4b1-f9fe-bce5-942b-182dfe91a935.png)


## 1. 技術評論社のRSS（XML形式）をJSON形式へ変換
まずは、RSSがXML形式のため、JSON形式へ変換します。JSONにする理由は、ファイルが視覚的にわかりやすく、またkey：valueという非常に汎用的に使えるデータ形式だからです。

Pandasは元データがXMLでもCSVでもJSONでも、データを読み込むことができるため、開発者はデータ処理に注力できるのが魅力の一つです。

技術評論者のRSSのリンクは以下です。
https://gihyo.jp/feed/rss2

Colaboratory（以下Colab）のルートにリンクから取得したXMLファイル「feed.xml」を配置しています。次のコードを実行すると、`data/feed.json`が出力されます。

※`data/feed.json`は一時ファイルのため、時間経過で削除されます。
```py
import pandas as pd
import os

# 入力XMLファイルのパス
input_xml_path = "feed.xml"
# 出力JSONファイルのパス
output_json_path = "data/feed.json"

# ファイルの存在確認
if not os.path.exists(input_xml_path):
    print(f"ファイルが見つかりません: {input_xml_path}")
else:
    try:
        # XMLファイルの 'item' タグをDataFrameに変換
        df_items = pd.read_xml(input_xml_path, xpath=".//item", encoding="utf-8")
        # DataFrameをJSON形式の文字列に変換
        json_str = df_items.to_json(orient='records', force_ascii=False, date_format='iso')
        # JSON形式の文字列をファイルに書き出す
        with open(output_json_path, "w", encoding='utf-8') as f:
            f.write(json_str)
    except Exception as e:
        print(f"ファイルの読み込み中にエラーが発生しました: {e}")
```


#### コード解説

```py
# XMLファイルの 'item' タグをDataFrameに変換
df_items = pd.read_xml(input_xml_path, xpath=".//item", encoding="utf-8")
```

`pd.read_xml(ファイル名)`でXMLファイルを読み込んでいます。
オプションの`xpath=".//item"`は`<item>`タグの内側のみ読み込むという意味です。XMLファイルには余分なタグがいくつかあるため、このオプションを使用しています。

```py
# DataFrameをJSON形式の文字列に変換
json_str = df_items.to_json(orient='records', force_ascii=False, date_format='iso'
```

この行では、DataFrame（XML）の内容をJSON形式の文字列に変換しています。

`orient='records'` は、出力されるJSONの形式を指定しています。records は、DataFrame(XML)の各行を個別のJSONオブジェクトとして出力する形式を意味しています。

![スクリーンショット 2023-12-09 22.42.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1650886/4c74f130-5ddb-91da-3f17-4b2d4c6e5f9d.png)

`force_ascii=False` は、JSON文字列に非ASCII文字（日本語など）が含まれる場合に、それらをエスケープせずにそのまま出力するよう指定しています。このオプションを指定しないと、非ASCII文字はエスケープされた形（例えば \uXXXX など）で出力されます。

`date_format='iso'` は、DataFrame内の日付データのフォーマットをISO 8601標準にするよう指定しています。これにより、日付と時刻は YYYY-MM-DDTHH:MM:SS の形式で出力されます。

## 2. JSONからRSSのtitleタグを取得し、txtファイルに出力
タイトルから、最も使用頻度の高いワードを取得したいため、先ほど作成した`feed.json`からtitleキーのバリューを抜き出しタイトル一覧ファイルを作成します。

```py
# JSONファイルを読み込み、DataFrameに変換
df = pd.read_json("data/feed.json")

# 'title' 列のデータを取得
titles = df['title'].tolist()

# 'title' のデータをtxtファイルに保存
with open('data/title.txt', 'w', encoding='utf-8') as f:
    for title in titles:
        f.write(title + '\n')
```

![スクリーンショット 2023-12-09 22.47.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1650886/618f4944-60d0-b512-9a02-4f3fb08cffd9.png)



## 3. タイトル一覧.txtから使用頻度の高いワードを抽出
PySparkを使ってタイトル一覧から使用頻度の高いワードを抽出します。

ここで一つ謝罪があります。頻出ワードを取得する単語は、英語にする必要があります。しかし翻訳するコードを書く手間よりGoogle翻訳に突っ込んだ方が早かったので、タイトル一覧ファイルを翻訳した、英語版タイトル一覧`"title_en.txt"`をColabのルートディレクトリに配置しました。


```py
from pyspark.sql import SparkSession, Row
from pyspark.sql.functions import explode,split,col,lower

# ストップワードのリストを定義
stopwords = ['of', 'and', 'the', 'is', 'in', 'to', 'a', 'for', 'on', 'with', 'by', 'at', 'from', 'as', 'it', 'an', 'are', 'this', 'that', 'be', 'not', 'or', 'but', 'if', 'than', 'so', 'such', 'only', 'own', 'same', 'so', 'than', 'too', 'very', 's', 't', 'can', 'will', 'just', 'don', 'should', 'now', '-', 'no.', '/','using','new','about','how','2023']

spark = SparkSession.builder\
 .master("local")\
 .appName('word_count')\
 .getOrCreate()

df=spark.read.text("title_en.txt")

df_count = (
    df.withColumn('word', explode(split(lower(col('value')), ' ')))  # 単語を小文字に変換
    .filter(~col('word').isin(stopwords))  # ストップワードを除外
    .groupBy('word')
    .count()
    .sort('count', ascending=False)
)

df_count.show()
```

次のような結果になります。最頻単語は「ubuntu」でしたね。

![スクリーンショット 2023-12-09 22.54.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1650886/55aa1d47-a484-ebe2-c74f-fe6567a992af.png)

#### コード解説

```py
spark = SparkSession.builder\
 .master("local")\
 .appName('word_count')\
 .getOrCreate()
```

Apache Sparkの初期化をしています。

```py
df_count = (
    df.withColumn('word', explode(split(lower(col('value')), ' ')))  # 単語を小文字に変換
    .filter(~col('word').isin(stopwords))  # ストップワードを除外
    .groupBy('word')
    .count()
    .sort('count', ascending=False)
)
```
読み込んだ英語版タイトル一覧から、頻出単語をカウントしています。大文字と小文字で別カウントになってしまうため、pyspark.sqlの関数`lower()`を使って、全て小文字にしています。
また、ストップワードを除外しています。ストップワードとは除外ワードとも呼ばれ、テキスト検索処理に関連する内容をもたないワードのことです（"and"、"or"、"in" など）。

## 4. JSONファイルとtitle情報からデータ分析・取得を行う

最頻単語は「Ubuntu」であることを発見したため、「Ubuntu」が使われているタイトルの記事一覧を取得します。

```py
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# JSONファイルを読み込む
df = spark.read.json("data/feed.json")
# タイトルとリンクを取得
df.createOrReplaceTempView("temp_view")
ubuntu = spark.sql("SELECT title, link FROM temp_view WHERE title LIKE '%Ubuntu%'")
# 結果を表示する
ubuntu.show(truncate=False)
```

結果は次のようになります。
![スクリーンショット 2023-12-09 23.05.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1650886/48f0d1ae-1cc3-387a-201d-c843151a80a9.png)

#### コード解説

```py
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
```
Apach Sparkの新しいセッションを作成しています。


```py
# タイトルとリンクを取得
df.createOrReplaceTempView("temp_view")
ubuntu = spark.sql("SELECT title, link FROM temp_view WHERE title LIKE '%Ubuntu%'")
```

`createOrReplaceTempView`メソッドを呼び出すと、一時的にビューテーブルを作成することができ、Spark SQLでSQLライクな書き方でデータ取得を行うことができます。


## まとめ
Pandasを使ってデータ加工、PySparkを使って、データ分析を行いました。


### 備考
本記事は「WEB+DB PRESS Vol.128」の特集1演習で慣れる！データベース入門SQL操作，テーブル設計，分析用データ処理を参考にしています。
https://gihyo.jp/magazine/wdpress/archive/2022/vol128

### 参考
https://medium.com/@manojt2501/pyspark-using-different-spark-api-to-write-word-count-program-324378ee04c6

https://stackoverflow.com/questions/63323344/using-pyspark-in-google-colab