Education Data Lake Export のデータをスキーマ付きで読み込む

こんにちは。株式会社 エーティーエルシステムズ 中村です。

[Education Data Lake Export](https://learn.microsoft.com/en-us/schooldatasync/enable-education-data-lake-export) を用いることで、Microsoft 365 Education テナントに関する「名簿データ」と「アクティビティデータ」をデータレイク経由で取得することができます。    
データは CSV 形式で出力されますが、CSV データにはヘッダ行がなく、スキーマとして出力される Common Data Model (CDM) に沿ってデータを取得する必要があります。

この記事では、Education Data Lake Export で出力されたデータを Synapse Analytics において CDM をもとに読み込む方法について解説します。

## 環境

本記事では、以下の環境で動作確認をしています。

- [Azure Synapse Analytics](https://azure.microsoft.com/ja-jp/products/synapse-analytics)
- Apache Spark 3.3
- Python 3.10
- Common Data Model (jsonSchemaSemanticVersion) 1.1.0

## Common Data Model (CDM) とは

[Common Data Model](https://www.microsoft.com/ja-jp/power-platform/common-data-model) は、Microsoft などが提供するデータモデルです。アプリケーション間で相互に利用できるよう標準化されたデータモデルとなっています。

Common Data Model については、@yomatsum さんによる [Common Data Model を紐解く](https://qiita.com/yomatsum/items/5b022701d0f12a98d0a7) が分かりやすいです。

## PySpark でスキーマを付けて読み込むメソッド

詳細な情報は後述しますが、まずは結論から。次のようなメソッドを用意します。

```python
from pyspark.sql.types import *
from pyspark.sql import functions as F

def load_csv_with_cdm(source_csv_path, cdm_schema_path, entity_name):
      """
      CDMスキーマをもとに、CSVを読み込みヘッダー付きデータフレームを返す

      source_csv_path: 元データとするCSVファイルのパス
      cdm_schema_path: CDMスキーマファイルのパス
      entity_name: エンティティ名
      """
      # CDMのスキーマ
      cdm_schema = StructType([
          StructField("jsonSchemaSemanticVersion", StringType(), True),
          StructField("imports", ArrayType(StructType([
                  StructField('corpusPath',StringType(), True)
          ])), True),
          StructField("definitions", ArrayType(StructType([
              StructField("entityName", StringType(), True),
              StructField("hasAttributes", ArrayType(StructType([
                  StructField("name", StringType(), True),
                  StructField("dataType", StructType([
                      StructField("dataTypeReference", StringType(), True)
                  ]), True),
                  StructField("dataFormat", StringType(), True),
                  StructField("description", StringType(), True)
              ])), True)
          ])), True)
      ])

      # スキーマを付けて、CDM JSONを読み込む
      df_schema = spark.read.load(cdm_schema_path, format="json", multiline=True, schema=cdm_schema)

      # エンティティ名に合致するスキーマのみとする
      df_schema = df_schema.withColumn("definition", F.explode("definitions"))
      df_schema = df_schema.withColumn("entityName", F.col("definition.entityName"))
      df_schema = df_schema.filter(F.col("entityName") == entity_name)
      df_schema = df_schema.limit(1)

      # 属性の名前とデータ型を attributeName と attributeType として取り出す
      # データ型は、dataType.dataTypeReference か dataFormat のいずれかで記載されている
      df_schema = df_schema.withColumn("attribute", F.explode("definition.hasAttributes"))
      df_schema = df_schema.withColumn("attributeName", F.col("attribute.name"))
      df_schema = df_schema.withColumn("attributeRawType",
          F.when(F.col("attribute.dataType.dataTypeReference").isNotNull(), F.col("attribute.dataType.dataTypeReference"))
          .when(F.col("attribute.dataFormat").isNotNull(), F.col("attribute.dataFormat"))
      )

      # [値名, データ型] の配列にする
      df_schema = df_schema.withColumn("attributeType", 
          F.when(F.lower("attributeRawType") == "datetime", F.lit("timestamp"))
          .when(F.lower("attributeRawType") == "int", F.lit("integer"))
          .when(F.lower("attributeRawType") == "int32", F.lit("integer"))
          .when(F.lower("attributeRawType") == "guid", F.lit("string"))
          .when(F.lower("attributeRawType") == "time", F.lit("string"))
          .otherwise(F.col("attributeRawType"))
      )
      df_schema = df_schema.select("attributeName", "attributeType")
      origin_schema = [[row.attributeName, row.attributeType] for row in df_schema.collect()]

      # PySparkで利用できるスキーマに変換する
      fields = []
      for col_schema in origin_schema:
          col_name = col_schema[0]
          col_dtype = col_schema[1]
          fields.append(StructField(col_name, globals()[col_dtype.lower().capitalize() + "Type"](), True))
      spark_schema = StructType(fields)

      # CDMをもとに作成したスキーマを用いて、CSVを読み込んでヘッダー付きデータフレームを作成
      df = spark.read.load(source_csv_path, format="csv", header=False, schema=spark_schema)

      return df
```

このメソッドは以下のように利用します。

```python
date = "yyyy-MM-dd" # 対象の日付に変更
roster_datetime = "yyyy-MM-ddThh-mm-ss" # 対象の日時に変更

activity_schema_version = "vX.Y.Z" # アクティビティスキーマのファイルバージョンに変更
activedirectory_schema_version = "vX.Y.Z" # ADスキーマのファイルバージョンに変更
roster_schema_version = "vX.Y.Z"  # Rosterスキーマのファイルバージョンに変更

activity_schema_path = "abfss://*****@***********.dfs.core.windows.net/M365/schema/activity." + activity_schema_version + ".cdm.json"
activedirectory_schema_path = "abfss://*****@***********.dfs.core.windows.net/M365/schema/activedirectory." + activedirectory_schema_version + ".cdm.json"
roster_schema_path = "abfss://*****@***********.dfs.core.windows.net/M365/schema/roster." + roster_schema_version + ".cdm.json"

# Activityの場合
activity_entity_name = "TechActivity"
activity_abfss_path = "abfss://*****@***********.dfs.core.windows.net/M365/activity/" + date + "/*.csv"

df_activity = load_csv_with_cdm(activity_abfss_path, activity_schema_path, activity_entity_name)
display(df_activity)

# Roster AadGroupの場合
aadgroup_entity_name = "AadGroup"
aadgroup_abfss_path = "abfss://*****@***********.dfs.core.windows.net/M365/roster/" + roster_datetime + "/" + aadgroup_entity_name + "/*.csv"

df_aadgroup = load_csv_with_cdm(aadgroup_abfss_path, aadgroup_schema_path, aadgroup_entity_name)
display(df_aadgroup)

# Roster Personの場合
person_entity_name = "Person"
person_abfss_path = "abfss://*****@***********.dfs.core.windows.net/M365/roster/" + roster_datetime + "/" + person_entity_name + "/*.csv"

df_person = load_csv_with_cdm(person_abfss_path, person_schema_path, person_entity_name)
display(df_person)
```

これを実行すると、Education Data Lake Export のデータをスキーマ付きで読み込むことができます。

## Education Data Lake Export におけるデータ構造

Education Data Lake Export でデータレイクに出力させると、以下のディレクトリ構造で格納されます。

```
M365
|   current.manifest.cdm.json
|
+--- activity
|   |   activity.manifest.cdm.json
|   |
|   \--- {yyyy-MM-dd}
|           ApplicationUsage.Part001.csv
|           ApplicationUsage.Part002.csv
|           ApplicationUsage.Part003.csv
+--- roster
|   |   current.roster.manifest.cdm.json
|   |
|   \--- {yyyy-MM-dd'T'hh-mm-ss}
|       |   snapshot.roster.manifest.cdm.json
|       |
|       +--- AadGroup
|       |       part-00000-*.csv
|       |
|       +--- AadGroupMembership
|       |       part-00000-*.csv
|       |
|       \--- AadUser
|               part-00000-*.csv
\--- schema
        activedirectory.{vX.Y.Z}.cdm.json
        activity.{vX.Y.Z}.cdm.json
        roster.{vX.Y.Z}.cdm.json
```

このうち、「名簿データ」が roster ディレクトリに格納されるデータ、「アクティビティデータ」が activity ディレクトリに格納されるデータとなります。  
それぞれ、「名簿データ」は `snapshot.roster.manifest.cdm.json`、「アクティビティデータ」は `activity.manifest.cdm.json` にデータ構造が記載されており、それらをもとにスキーマを選択します。

実際にこれらを読むと、以下の関連性を確認できます。

| エンティティディレクトリ | エンティティ名 | スキーマファイル名 |
| :- | :- | :- |
| activity | TechActivity | activity.{vX.Y.Z}.cdm.json |
| roster/AadGroup | AadGroup | activedirectory.{vX.Y.Z}.cdm.json |
| roster/AadGroupMembership | AadGroupMembership | activedirectory.{vX.Y.Z}.cdm.json |
| roster/AadUser | AadUser | activedirectory.{vX.Y.Z}.cdm.json |
| roster/AadUserPersonMapping | AadUserPersonMapping | roster.{vX.Y.Z}.cdm.json |
| roster/Course | Course | roster.{vX.Y.Z}.cdm.json |
| roster/CourseGradeLevel | CourseGradeLevel | roster.{vX.Y.Z}.cdm.json |
| roster/CourseSubject | CourseSubject | roster.{vX.Y.Z}.cdm.json |
| roster/Enrollment | Enrollment | roster.{vX.Y.Z}.cdm.json |
| roster/Organization | Organization | roster.{vX.Y.Z}.cdm.json |
| roster/Person | Person | roster.{vX.Y.Z}.cdm.json |
| roster/PersonDemographic | PersonDemographic | roster.{vX.Y.Z}.cdm.json |
| roster/PersonDemographicEthnicity | PersonDemographicEthnicity | roster.{vX.Y.Z}.cdm.json |
| roster/PersonDemographicPersonFlag | PersonDemographicPersonFlag | roster.{vX.Y.Z}.cdm.json |
| roster/PersonDemographicRace | PersonDemographicRace | roster.{vX.Y.Z}.cdm.json |
| roster/PersonEmailAddress | PersonEmailAddress | roster.{vX.Y.Z}.cdm.json |
| roster/PersonIdentifier | PersonIdentifier | roster.{vX.Y.Z}.cdm.json |
| roster/PersonOrganizationRole | PersonOrganizationRole | roster.{vX.Y.Z}.cdm.json |
| roster/PersonPhoneNumber | PersonPhoneNumber | roster.{vX.Y.Z}.cdm.json |
| roster/PersonRelationship | PersonRelationship | roster.{vX.Y.Z}.cdm.json |
| roster/RefDefinition | RefDefinition | roster.{vX.Y.Z}.cdm.json |
| roster/Section | Section | roster.{vX.Y.Z}.cdm.json |
| roster/SectionGradeLevel | SectionGradeLevel | roster.{vX.Y.Z}.cdm.json |
| roster/SectionSession | SectionSession | roster.{vX.Y.Z}.cdm.json |
| roster/SectionSubject | SectionSubject | roster.{vX.Y.Z}.cdm.json |
| roster/Session | Session | roster.{vX.Y.Z}.cdm.json |
| roster/SourceSystem | SourceSystem | roster.{vX.Y.Z}.cdm.json |

この関連性をもとに、load_csv_with_cdm メソッドの引数にエンティティ名やスキーマパスを渡し、読み込むという流れになります。

## さいごに

Education Data Lake Export で出力された CSV データを Common Data Model を用いてスキーマ付きで読み込むことができました。  
Education Data Lake Export は日本での利用者数が少なく、情報もあまり多くないため手探りで試し実装することが多いですが、その分うまくいったときは嬉しいものです。

この記事の内容が参考になったら「いいね」をお願いします:star: