pysparkでFitbitの睡眠JSONをパースして平日と休日の平均睡眠時間を求めてみた

pysparkでFitbitの歩数JSONをパースして、平日と休日の平均歩数を求めてみました。

↓PySparkで歩数JSONを平日・休日別に集計

↓Pythonでの睡眠時間比較

確認環境
コード
実行結果例
2019年、2020年の月別、平日・休日別平均睡眠時間
参考にしました

確認環境

import sys
print(sys.version)
3.8.6 | packaged by conda-forge | (default, Dec 26 2020, 05:05:16) 
[GCC 9.3.0]

print(spark.version)
3.0.1

コード

文字列だった日付を日付型に変換したのち、月、曜日、平日、休日フラグ列を追加したテーブルを作成。それを集計することで月別、平日・休日別の平均睡眠時間をSQLで集計しました。なお、睡眠時間はmsで記録されていたので、hに変換しています。

from pyspark.sql import SparkSession
from pyspark.sql import functions

spark = SparkSession \
    .builder \
    .enableHiveSupport() \
    .getOrCreate()

# 4半期ごとのJSONファイルを一括でロード
df = spark.read.json('/tmp/sleep_2020*.json')

# 要素を行に分割したカラムを作成
col = functions.explode(functions.col('sleep'))

# 元のカラムの横にカラムを追加
df = df.withColumn('lines',col)

# 元のカラムを削除
df = df.drop('sleep')

# isMainSleepがTrueのレコードの日付、睡眠時間(ms)でDFを作成
df = df.select(functions.col('lines.dateOfSleep').alias('strDateOfSleep'),'lines.duration').where(df.lines.isMainSleep == 'true')

#  dateOfSleepをパースして日付型にしたカラムの追加,strDateOfSleepカラムの削除
df = df.withColumn("dateOfSleep", functions.to_timestamp("strDateOfSleep", 'yyyy-MM-dd'))
df = df.drop("strDateOfSleep")

# 睡眠時間(H)のカラムを追加して、msのカラムを削除
df = df.withColumn("sleepTime",functions.round(functions.col('duration')/(60*60*1000),2))
df = df.drop("duration")

# dateOfSleepを基に、曜日列、月列を追加
df = df.withColumn("dayOfWeek", functions.date_format("dateOfSleep", 'E'))
# 月はそのままだと文字列扱いなので数値にキャスト
df = df.withColumn("month", functions.date_format("dateOfSleep", "M").cast("integer"))

# dateOfSleepを基に平日と週末を区別する列を追加
df = df.withColumn("is_weekday", 
                   functions.when((functions.col("dayOfWeek") == "Sat") | (functions.col("dayOfWeek") == "Sun"), 0).otherwise(1))

# DFをSQLで操作するため、テーブル登録
df.registerTempTable("sleep")

# 月別、平日、週末別の平均睡眠時間を集計
sql = """
select
    month
   ,round(avg(sleepTime) filter(where is_weekday = 1), 1)as avg_weekday
   ,round(avg(sleepTime) filter(where is_weekday = 0), 1) as avg_weekend
from
    sleep
group by
    month
order by
    month
"""

df = spark.sql(sql)

# DFの表示
df.show()

実行結果例

# 集計元テーブル
+-------------------+---------+---------+-----+----------+
|        dateOfSleep|sleepTime|dayOfWeek|month|is_weekday|
+-------------------+---------+---------+-----+----------+
|2020-12-31 00:00:00|     9.17|      Thu|   12|         1|
|2020-12-30 00:00:00|     8.35|      Wed|   12|         1|
|2020-12-29 00:00:00|     6.82|      Tue|   12|         1|
|2020-12-28 00:00:00|     6.98|      Mon|   12|         1|
|2020-12-26 00:00:00|     6.08|      Sat|   12|         0|
|2020-12-25 00:00:00|     8.33|      Fri|   12|         1|
|2020-12-24 00:00:00|     6.77|      Thu|   12|         1|
...

# 集計結果テーブル
+-----+-----------+-----------+
|month|avg_weekday|avg_weekend|
+-----+-----------+-----------+
|    1|        6.2|        8.4|
|    2|        6.9|        8.8|
|    3|        7.1|        8.1|
|    4|        7.7|        8.1|
...

2019年、2020年の月別、平日・休日別平均睡眠時間

month	2019_weekday	2019_weekend	2020_weekday	2020_weekend
1	7.1	8.6	6.2	8.4
2	6.7	8.2	6.9	8.8
3	6.1	8	7.1	8.1
4	7.2	7.6	7.7	8.1
5	6.9	8.2	7.4	8.1
6	6.7	8.2	7.9	9.5
7	5.2	6.9	7.9	8.8
8	6.8	6.9	7.8	7.5
9	6.8	6.8	7.5	8.6
10	6.9	5.9	7.6	8
11	6.7	8.3	7.1	8.1
12	6.5	7.9	8.1	7.7