AI Data Platform(AIDP) にオープンデータを取り込んで東京都62市区町村を分析してみてみた

Last updated at 2026-10-21Posted at 2026-10-21

■ はじめに

前回は、Oracle AI Data Platform（AIDP）を作成し、外部Databaseの表へSQLを実行してみてみました。今回は、Databaseの外にあるデータをAIDPへ取り込み、SQLで比較できる形に整えるところから試してみます。

使用するのは、e-Statで公開されている国勢調査のExcelファイルです。人口・世帯・年齢のデータを使って、「東京都ではどの地域の人口が増えているのか」「単独世帯の割合や高齢者割合は、地域によってどのくらい違うのか」を調べます。

ただ、公開されているExcelをそのままJOINすればよい、というわけではありません。表題や注記、合計行、旧市町村の行を見分け、比較する地域と指標の意味を揃える必要があります。今回は東京都62市区町村を対象に、原本の読込みからParquetへの保存、Spark SQLによる分析、品質検査、Workflowによる一連の処理の実行までをまとめます。

令和7年（2025年）のデータも公開されていますが、本稿では、まず令和2年（2020年）のデータで処理の流れと検査方法を確認します。この土台を作ってから、続編で2025年データとの比較やAIによる分析へ進める予定です。

ということで、今回はAIDPにオープンデータを取り込んで、東京都62市区町村の分析とWorkflowによる自動実行をしてみてみます。

■ 今回やってみること

2020年10月1日時点の国勢調査を使い、次のような地域の違いを比較できました。割合は小数第2位まで表示しています。

中央区の人口増加率：19.83%。2015年から2020年の増加率で、分母は2020年の境域に組み替えた2015年人口です。
新宿区の単独世帯割合：67.80%。単独世帯数を一般世帯数で割った割合です。
檜原村の65歳以上の割合：53.03%。65歳以上の人数を年齢が分かる人の合計で割り、年齢不詳を分母から除いています。

到達点は「東京都62市区町村の分析データを原本から作り、同じ処理を再実行できること」です。処理の全体像は次のとおりです。

e-Statの原本をObject Storageに置き、3つのNotebookが順番に読み書きします。Notebook間のデータは、すべて実行IDごとのディレクトリruns/〈run_id〉/配下のファイルで受け渡します。

Notebook	担当する処理	主な利用機能
`nb01_ingest_census`	Excelの表頭・注記を処理し、人口・世帯・年齢の3つのデータを抽出する	Python、pandas、openpyxl、明示的なSpark型、Parquet出力
`nb02_transform_census`	3つを地域コードで結合し、割合を計算・検査して保存する	Spark DataFrame、JOIN、Spark SQL、品質検査
`nb03_analyze_census`	保存結果を再読込みし、ランキング・グラフを作成してCSVへ出力する	Spark SQL、一時ビュー、可視化、CSV出力・再読込み

Workflowでは、この3つを順番に実行します。データはVolume経由のParquetで受け渡し、別NotebookのPython変数や一時ビューが残ることには依存しません。出力は実行ID（run_id）ごとに、次の構成で保存します。

census_workの出力構成

census_work（tokyo-census/work/）
└── runs/
    └── {run_id}/
        ├── source_manifest/          … 原本のサイズ・SHA-256・東京都合計（nb01）
        ├── stage/
        │   ├── population/           … 人口の中間Parquet（nb01）
        │   ├── households/           … 世帯の中間Parquet（nb01）
        │   └── age/                  … 年齢の中間Parquet（nb01）
        ├── curated/
        │   └── tokyo_comparison/     … 分析用Parquet（nb02）
        ├── validation/               … 検査記録 status=PASS（nb02）
        └── exports/
            └── tokyo_comparison_csv/ … 受渡し用CSV（nb03）

Oracle Databaseでの近いイメージ	今回使うもの
外部ファイルへのアクセス口	Volume
SQL Developerのワークシート＋説明文＋結果表示	Notebook
SQL実行エンジン／処理途中の問合せ結果	Spark／DataFrame
ステージング表	今回の3つの中間Parquet
SELECT・JOIN・GROUP BY	Spark SQL／DataFrame操作
データの保存	Parquet。Oracleの表・索引・制約とは異なる
DBMS_SCHEDULERのジョブ・チェーン	WorkflowのJobとTask

今回は小規模な機能検証です。62行の結果からSparkの性能優位性は判断しません。Excelの読込みはdriver上のPython処理で、後半のJOIN・計算・保存をSparkで行います。

■ 使用するオープンデータ

統計の基準日は2020年10月1日です。人口増減率は、2020年の市区町村境域に組み替えられた2015年人口との5年間の比較です。現在人口・最新人口・年率として説明しません。

表	利用する項目	本稿での保存名	公式ページ
表1-1	2020人口、2015組替人口、総世帯数、一般世帯数	`population_1_1.xlsx`	e-Stat 表1-1
表9-1	単独世帯数、照合用の一般世帯数	`households_9_1.xlsx`	e-Stat 表9-1
表2-7	年齢3区分、年齢不詳、公表割合	`age_2_7.xlsx`	e-Stat 表2-7

全国のExcel原本を取得し、Notebookで東京都を抽出します。利用するのは「人口等基本集計」の上記表です。人口集中地区版や不詳補完済みの参考表とは区別します。

出典表記：総務省統計局「令和2年国勢調査人口等基本集計」（政府統計の総合窓口 e-Stat）を加工して作成。
利用条件はe-Stat利用規約を参照してください。

■ 事前準備

● AIDPの既存環境を確認

AIDPインスタンスとWorkspaceは、前回の記事で作成済みの環境を使用します。今回は既存のWorkspaceを選び、その中にNotebook・Compute・Workflowを準備します。本稿ではPrivate Networkを使用しません。画面名や利用可能な機能は実環境で確認します。

準備するもの	確認内容
Workspace	作成済みのWorkspaceを使用する。本稿の画面では`AIDP_OpenData_PoC`。別名の場合はご自身のWorkspaceを選ぶ
Spark Compute	使用するWorkspace内に`Census_PoC_Cluster`を用意する。Python／Sparkの実際の版は後で出力する
標準CatalogとSchema	本稿で使用した`blog_catalog.opendata`。未作成の場合は次の手順で用意する
Object Storage	検証用バケットと、原本・出力用の別プレフィックス
権限	バケット／オブジェクトへの必要なアクセスと、AIDP内のVolume・Notebook・Compute・Workflowの権限

本編はObject StorageのデータをSparkで扱います。外部Databaseへの接続やAIエージェントの作成は、ここでの実行条件に含めていません。

1) 作成済みのWorkspaceを選ぶ

AIDPの左メニューからWorkspacesを開き、本稿で使用するWorkspaceを選択します。以降のNotebook・Compute・Workflowは、同じWorkspace内で操作します。

■ CatalogとSchemaを用意する

本稿では標準Catalogをblog_catalog、Schemaをopendataとします。同名のCatalog・Schemaを作成済みの場合は、そのまま使用します。

1) Master catalog画面
Master catalogからCatalogの作成画面を開きます。

2) Create catalog in Master catalog画面
Catalog nameにblog_catalog、Catalog typeにStandard catalogを指定し、Descriptionには後述の「Descriptionをまとめて設定する」の値を入力して作成します。保存先Compartmentを変更する場合は、対象Compartmentも指定します。

3) Create catalog in Master catalog完了画面

4) opendata Schemaを用意する

作成したblog_catalogを開き、Schemaタブから［Create Schema］を選択します。Nameをopendata、Descriptionには同じ一覧の値を入力して作成します。標準Catalogに自動作成されるdefault Schemaとは別に用意します。公式：Schema

Schemaの作成には、対象CatalogのCREATE_SCHEMAまたはADMIN権限を確認します。公式：Permissions Model

■ 原本を用意する

1) 原本Excelを取得

上記3つの公式ページからEXCELをダウンロードし、表の保存名に揃えます。

ターミナルで取得する場合の例です。これは手元のPCで実行するbashコマンドで、AIDPのPythonセルには貼り付けません。curlが必要です。新しい作業フォルダで実行します。

実行コマンド

mkdir -p aidp-opendata/raw/census
cd aidp-opendata
curl --fail --location --retry 2 --connect-timeout 30 --max-time 180 \
  'https://www.e-stat.go.jp/stat-search/file-download?statInfId=000032142402&fileKind=0' \
  --output raw/census/population_1_1.xlsx
curl --fail --location --retry 2 --connect-timeout 30 --max-time 180 \
  'https://www.e-stat.go.jp/stat-search/file-download?statInfId=000032142495&fileKind=0' \
  --output raw/census/households_9_1.xlsx
curl --fail --location --retry 2 --connect-timeout 30 --max-time 180 \
  'https://www.e-stat.go.jp/stat-search/file-download?statInfId=000032142410&fileKind=0' \
  --output raw/census/age_2_7.xlsx
ls -lh raw/census

実行コマンド出力結果

% ls -lh raw/census
    total 27400
    -rw-r--r--  1 user      staff   8.1M Oct  8 19:16 age_2_7.xlsx
    -rw-r--r--  1 user      staff   4.5M Oct  8 19:16 households_9_1.xlsx
    -rw-r--r--  1 user      staff   775K Oct  8 19:16 population_1_1.xlsx

固定した公開ファイルを取得する方式です。e-Stat APIのアプリケーションIDは使いません。ネットワーク制限で取得できない場合は、公式ページからブラウザで取得します。

2) Object Storageへアップロード

OCIコンソールから対象バケットを開き、ObjectsのUploadで3件を配置します。バケットを公開する必要はありません。

オブジェクト名	内容
`tokyo-census/raw/population_1_1.xlsx`	人口原本
`tokyo-census/raw/households_9_1.xlsx`	世帯原本
`tokyo-census/raw/age_2_7.xlsx`	年齢原本
`tokyo-census/work/`配下	今後AIDPが書き込む中間・最終結果

Object Storageの「フォルダ」はオブジェクト名のプレフィックスです。出力側が選択画面に現れない場合は、コンソールでフォルダ／プレフィックスを作成するか、空の_keep.txtをtokyo-census/work/_keep.txtとして置きます。

■ Volumeを作成する

Volumeは、ファイルにAIDPの名前と権限を付けてアクセスする入口です。今回はObject Storageの既存位置を参照するExternal Volumeを使います。公式：Volumes

既存のCatalog・Schema・Volumeを使用する場合は、原本のアップロードや同じ保存範囲へのVolume登録を重複して行わず、この節のパスと権限の確認から進めます。

● 入力・出力用Volumeの登録

Create Volumeを選択できない場合は、標準Catalogの対象Schemaを開いていることと、そのSchemaに対するCREATE_VOLUMEまたはADMIN権限を確認します。OCI IAMのread buckets／inspect objectsと、AIDP内のVolume作成権限は別の設定です。

作成後は、Notebook／Workflowの実行ユーザーについて、入力VolumeのREAD、出力VolumeのWRITE（読込みを含む）、またはそれらを含むADMIN権限を確認します。SchemaとVolumeの権限は、それぞれのPermissionsタブで確認します。公式：Permissions Model

標準Catalog内の対象Schemaを開き、VolumesからCreate Volumeを選択します。
Externalを選び、Compartment・Bucket・Folderを指定します。

Volume名の例	対応するObject Storageの範囲	用途
`census_raw`	`tokyo-census/raw/`	原本読込み
`census_work`	`tokyo-census/work/`	中間・分析結果の保存

別のExternal Volume／External Tableと保存範囲が重ならないようにします。バケット全体を登録したVolumeの下に、さらに同じバケットの一部を登録する構成は避けます。

・ census_raw Volume登録

1) Volumes画面
標準Catalog内の対象Schemaを開き、Volumes画面から、[+]をクリック

2) Create Volume画面
Externalを選び、Compartment・Bucket・Folderを指定し、[Create]をクリック

3) Create Volume完了

4) Volume内ファイル確認
Bucket内のtokyo-census/raw/プレフィックスに配置した3つのファイルが見えることを確認します。

・ census_work Volume登録

1) 出力用Volumeを登録
同じSchema内にcensus_workを作成し、Folderをtokyo-census/work/にします。

2) パスを確認

VolumeのCopy path、またはNotebook左側のオブジェクトのメニューで実際のパスを取得します。本稿では次のPOSIX形式を使用します。

変数名	今回の値。別環境では実際のパスへ置き換える
`raw_volume`	`/Volumes/blog_catalog/opendata/census_raw`
`work_volume`	`/Volumes/blog_catalog/opendata/census_work`

この形式のVolumeパスはPythonとSparkの両方から利用できます。公式：Workspaces / Work with Files

・ census_raw Volumeパス確認

Storage location

oci://Bucket-AIDP@<namespace>/tokyo-census/raw/

・ census_work Volumeパス確認

Storage location

oci://Bucket-AIDP@<namespace>/tokyo-census/work/

■ Compute・Notebook・ライブラリを準備する

● 使用するWorkspaceにComputeを作成する

AIDP_OpenData_PoC内で分析用のComputeを準備します。

1) Compute画面
Compute → Create Clusterを開く
左メニューで本稿に使用するWorkspaceが選択されていることを確認し、その配下のComputeからCreate Clusterを開きます。

2) Create Cluster画面
Clusterの設定を入力する

項目	今回の設定例・選び方
Name	`Census_PoC_Cluster`
Description	後述のDescription一覧を参照
Runtime version	`Spark 3.5.0`
Configuration	`Custom`
Driver	AMD、2 OCPU、Memory 16 GB、Block volume 200 GB
Worker（1台当たり）	AMD、2 OCPU、Memory 16 GB、Block volume 200 GB
Worker数	Autoscale、最小1／最大10
Run duration	Idle timeout、120分

上表は今回の作成画面の設定値で、62行の処理に必要な最小構成を示すものではありません。検証規模に合わせてWorker上限とIdle timeoutを選びます。

Createをクリックし、利用可能な状態になったことを確認します。QuickstartでもWorker数とIdle timeoutを確認してから作成します。公式：Manage Compute

Compute作成の確認結果

Workspace名：AIDP_OpenData_PoC
Cluster名：Census_PoC_Cluster
Runtime version：Spark 3.5.0
Driver／Worker：各AMD・2 OCPU・Memory 16 GB
Worker数：Autoscale、最小1／最大10（作成画面の設定）
Idle timeout：120分
作成後の状態：Active（Notebook接続・実行時に確認）

3) Create Cluster完了

● Notebookを3つ揃える

同じWorkspaceのCreate → Notebookから、次の3つを作成します。

Notebook	担当する処理
`nb01_ingest_census`	原本Excelの読込み・62市区町村の抽出・中間Parquetの保存
`nb02_transform_census`	Sparkによる結合・指標計算・品質検査・分析用Parquetの保存
`nb03_analyze_census`	Spark SQLによるランキング・グラフ・CSVの出力と再読込み

以下はNotebook 01を作成する例です。Notebook 02・03も同じ操作で作成し、上の表の名前へ変更します。

1) 使用するWorkspaceからNotebookを作成する

AIDP_OpenData_PoCのContent画面で、右上のCreate → Notebookを選択します。

今回の操作ではName・Descriptionを入力する設定画面は表示されず、Untitled_1791544475139340.ipynbというNotebookが自動作成され、そのまま編集画面が開きました。末尾の数字は自動生成された値なので、読者の環境では異なります。

公式資料にはName・Descriptionを入力して作成する手順も記載されていますが、本稿では今回確認できた操作経路を使用します。公式：Notebooks / Create a Notebook

2) Notebook名を変更する

編集画面上部のUntitled_...というNotebook名をクリックし、nb01_ingest_censusへ変更して確定します。今回の画面では.ipynbは入力欄の外に表示されるため、入力するのは拡張子を除いた名前です。

画面上部と左下のActivityの名前がnb01_ingest_censusに変わることを確認します。Notebook名を直接クリックして変更する方法は公式資料にも記載されています。公式：Notebooks / Rename a Notebook

Nameは英字から始めます。nb01_のように英字と番号を組み合わせ、Descriptionには後述の一覧の値を設定します。

3) 保存を確認する

Pythonセルに以下を入力します。ここではまだ実行しません。

保存確認用セル（この段階では実行しない）

print("AIDP notebook save test")

自動保存が完了し、Auto-save failed等のエラーが表示されないことを確認します。公式資料では自動保存は2分ごとです。保存完了を確認した後に別の画面へ移動し、同じNotebookを開いてセルの内容が残っていることを確認します。公式：Notebooks / Auto-save

Notebook保存の確認結果

Workspace名：AIDP_OpenData_PoC
Notebook名：nb01_ingest_census.ipynb
Untitledからの名前変更：完了
保存完了表示：Last saved at 10/9/26, 8:24:28 PM
保存したセル：print("AIDP notebook save test")
保存時のエラー：なし

保存確認用セルは、共通パラメータセルを配置する際に置き換えます。

各Notebookを開き、画面上部のClusterメニュー、またはActions → Attach an existing clusterからCensus_PoC_Clusterを選びます。Cluster名と接続状態を確認してからセルを実行します。公式：Notebooks

・ nb01_ingest_censusをCensus_PoC_Clusterへアタッチ

1) nb01_ingest_census画面
Notebookを開き、画面上部の [Cluster] > [Attach existing cluster] からCensus_PoC_Clusterを選択するとアタッチが開始します。

2) nb01_ingest_census アタッチ to Census_PoC_Cluster完了
アタッチ完了すると Cluster: Census_PoC_Cluster (Active) になります。

・ nb02_transform_censusをCensus_PoC_Clusterへアタッチ

同様に nb02_transform_census を Census_PoC_Clusterへアタッチします。

・ nb03_analyze_censusをCensus_PoC_Clusterへアタッチ

同様に nb03_analyze_census を Census_PoC_Clusterへアタッチします。

接続後の最初の実行確認

nb01_ingest_censusで、先ほどの保存確認用セルを実行します。

実行コマンド

print("AIDP notebook save test")

実行コマンド出力結果

AIDP notebook save test

照合用の期待値はAIDP notebook save testです。確認後、このセルを本編の共通パラメータセルに置き換えます。

以下の実行ブロックは、特記したもの以外はPythonセルで実行します。SQLもspark.sql(...)としてPythonから呼び出すので、セルの言語はPythonのままで構いません。
コードブロックの表示名は「実行コマンド」、続く出力は「実行コマンド出力結果」に統一しています。Notebookにはコードの中身だけを貼り付けます。

● ライブラリを確認

まずnb01_ingest_censusで、以下の環境確認用セルを実行します。次の出力は、追加ライブラリを入れる前の記録です。

実行コマンド

import sys
import importlib.metadata
print("Python:", sys.version)
print("Spark:", spark.version)
for package in ("pandas", "openpyxl", "matplotlib"):
    try:
        print(package, importlib.metadata.version(package))
    except importlib.metadata.PackageNotFoundError:
        print(package, "NOT INSTALLED")

実行コマンド出力結果

Python: 3.11.13 (main, Apr 27 2026, 16:44:16) [GCC 8.5.0 20210514 (Red Hat 8.5.0-28.0.1)]
Spark: 3.5.0
pandas 2.2.3
openpyxl NOT INSTALLED
matplotlib NOT INSTALLED

原本の読込みにはpandasとopenpyxl、グラフにはmatplotlibを使います。SparkはAIDPのRuntimeが提供するものを使い、pysparkを別途pipで上書きしません。

不足している場合はCompute → 対象Cluster → Library → Install Libraryから、必要なライブラリを指定したrequirements.txtを登録します。以下は、今回のWorkflow実行まで成功した構成です。requirements.txtの内容として保存します。

requirements.txt

numpy==1.26.4
pandas==2.2.3
openpyxl==3.1.5
matplotlib==3.10.1

古いrequirements.txtを登録済みの場合は、今回追加した対象LibraryをUninstallしてから、この内容で入れ替えます。インストール完了後にClusterをRestartし、Activeへ戻ってから以下の確認セルを実行します。クラスタのLibraryは、そのクラスタを使用するNotebookとWorkflowの両方に適用されます。公式：Manage Libraries

・ NumPyとpandasの組み合わせを確認する

途中の環境ではNumPy 2.4.6が追加ライブラリ側、pandas 2.2.3が標準環境側から読み込まれ、pandasの行抽出でRecursionErrorになりました。2行だけのDataFrameでも再現しています。

上記のライブラリ構成へ入れ替えてClusterを再起動したところ、NumPy 1.26.4とpandas 2.2.3の両方が追加ライブラリ側から読み込まれ、確認セルとWorkflowが成功しました。これは今回の環境で確認できた復旧結果です。NumPyの版だけが原因だったか、読み込み状態も影響したかまでは確定していません。

実行コマンド

import numpy as np
import pandas as pd

print("NumPy:", np.__version__, np.__file__)
print("pandas:", pd.__version__, pd.__file__)
print("NumPy確認:", np.array([False, True]).any())

test = pd.DataFrame({"E": ["13_東京都", "12_千葉県"]})
print(test.loc[test["E"] == "13_東京都"])

実行コマンド出力結果

NumPy: 1.26.4 /aidp/libraries/python/amd/numpy/__init__.py
pandas: 2.2.3 /aidp/libraries/python/amd/pandas/__init__.py
NumPy確認: True
        E
0  13_東京都

・導入後に4つのライブラリの実際の版を記録する

ライブラリ導入・Clusterの再起動が完了したら、Census_PoC_Clusterへ接続したnb01_ingest_censusで、次の確認セルを実行します。4つのライブラリを実際にimportし、指定した版・読み込まれた版・読込み元のパスをまとめて出力します。importlib.metadata.version()によるインストール情報に加え、ここではimportしたモジュールの__version__と__file__を記録します。

実行コマンド（ライブラリ導入後の環境記録）

import sys
import json
import importlib
from datetime import datetime, timezone, timedelta

requested_versions = {
    "numpy": "1.26.4",
    "pandas": "2.2.3",
    "openpyxl": "3.1.5",
    "matplotlib": "3.10.1"
}

environment_record = {
    "checked_at": datetime.now(timezone(timedelta(hours=9))).isoformat(),
    "python": sys.version,
    "spark": spark.version,
    "libraries": {}
}

for package, requested_version in requested_versions.items():
    module = importlib.import_module(package)
    loaded_version = str(module.__version__)
    environment_record["libraries"][package] = {
        "requested_version": requested_version,
        "loaded_version": loaded_version,
        "loaded_from": module.__file__,
        "matches_requested": loaded_version == requested_version
    }

print(json.dumps(environment_record, ensure_ascii=False, indent=2))

導入後の環境記録（JSON・実行画面）

実行コマンド出力結果

{
  "checked_at": "2026-10-12T08:27:54.533891+09:00",
  "python": "3.11.13 (main, Apr 27 2026, 16:44:16) [GCC 8.5.0 20210514 (Red Hat 8.5.0-28.0.1)]",
  "spark": "3.5.0",
  "libraries": {
    "numpy": {
      "requested_version": "1.26.4",
      "loaded_version": "1.26.4",
      "loaded_from": "/aidp/libraries/python/amd/numpy/__init__.py",
      "matches_requested": true
    },
    "pandas": {
      "requested_version": "2.2.3",
      "loaded_version": "2.2.3",
      "loaded_from": "/aidp/libraries/python/amd/pandas/__init__.py",
      "matches_requested": true
    },
    "openpyxl": {
      "requested_version": "3.1.5",
      "loaded_version": "3.1.5",
      "loaded_from": "/aidp/libraries/python/amd/openpyxl/__init__.py",
      "matches_requested": true
    },
    "matplotlib": {
      "requested_version": "3.10.1",
      "loaded_version": "3.10.1",
      "loaded_from": "/aidp/libraries/python/amd/matplotlib/__init__.py",
      "matches_requested": true
    }
  }
}

上の確認セルで、4つとも指定した版と一致することを確認しました。requirements.txtの指定値と、実際にimportしたモジュールの版を分けて記録しています。

ライブラリ	requirements.txtの指定版	実際に読み込まれた版
NumPy	`1.26.4`	`1.26.4`
pandas	`2.2.3`	`2.2.3`
openpyxl	`3.1.5`	`3.1.5`
matplotlib	`3.10.1`	`3.10.1`

4つすべてmatches_requested: trueで、読込み元は/aidp/libraries/python/amd/配下でした。Pythonは3.11.13、Sparkは3.5.0です。前のNOT INSTALLEDの出力は導入前の記録で、ここに掲載したJSONと画面が導入後の実測記録です。

Clusterの再起動後は変数や一時ビューを作り直す必要があるため、Notebookの共通パラメータセルから順に実行します。すでに書き込んだ出力を作り直す場合は、次節のルールで新しいrun_idを指定します。

● Descriptionをまとめて設定する

続編でAIにデータや処理の意味を伝える際に参照できるよう、各オブジェクトにDescriptionを設定します。作成時に入力できるものはその場で設定し、省略した場合は以下の場所から編集します。説明文はAIへ渡すメタデータとして利用する予定です。

対象	Descriptionの入力例	設定・編集箇所
Workspace `AIDP_OpenData_PoC`	東京都の国勢調査データを使用し、Sparkによる加工・分析とWorkflowによる自動実行を検証する。	Workspaces一覧の［⋯］→［Edit］
Catalog `blog_catalog`	国勢調査などのオープンデータを取り込み、加工・分析に使用するCatalog。	Master catalog一覧の［⋯］→［Edit description］
Schema `opendata`	国勢調査オープンデータの原本と加工結果を管理するSchema。	Catalog内のSchema一覧の［⋯］→［Edit Description］
Volume `census_raw`	国勢調査のExcel原本を参照する入力用Volume。	Schema内のVolumes一覧の［⋯］→［Edit Description］
Volume `census_work`	国勢調査の中間データ、分析用Parquet、検査記録、CSVを実行IDごとに保存する出力用Volume。	Schema内のVolumes一覧の［⋯］→［Edit Description］
Compute `Census_PoC_Cluster`	国勢調査データの読込み・加工・Spark SQL分析・Workflow実行に使用する検証用Compute。	Compute一覧の［⋯］→［Edit］
Notebook `nb01_ingest_census`	国勢調査の原本ExcelをVolumeから読み込み、東京都の市区町村データを抽出・整形して、後続処理で使用する3つの中間データを作成する。	Notebook名の下にあるDescriptionの編集欄
Notebook `nb02_transform_census`	中間データをSparkで読み込み、地域コードで結合する。人口増減率や単独世帯割合を計算し、62市区町村の件数・重複・欠損・合計値を検査して、分析用データをParquet形式で保存する。	Notebook名の下にあるDescriptionの編集欄
Notebook `nb03_analyze_census`	加工済みParquetを読み込み、東京都62市区町村の人口、人口増減率、単独世帯割合、年齢別割合を分析する。ランキングとグラフを作成し、分析用CSVを出力・再読込みする。	Notebook名の下にあるDescriptionの編集欄
Job `tokyo_census_pipeline`	国勢調査Excelから東京都62市区町村の分析データを作成	Create JobのDescription欄

入力後は画面の保存操作で確定し、各オブジェクトに説明が表示されることを確認します。詳しい処理内容や注意点は、NotebookのMarkdownセルにも記載できます。公式：Workspaces／Standard Catalogs／Schema／Volumes／Manage Compute

● 3つのNotebookに共通のパラメータセルを置く

このセルを3つのNotebookの冒頭に同じ内容で配置します。 raw_volumeとwork_volumeは実際のパスに変更します。手動実行では、3つとも同じrun_idを指定してください。

run_idは今回の実行を識別する値です。出力先を実行単位で分け、原本や過去の成功結果を上書きしません。手動実行では、3つとも同じIDに揃えます。次に最初から実行する場合は、3つとも未使用のIDへ変更します。

実行コマンド

import re
import json
from pathlib import Path

# 実際のCatalog・Schema・Volume名に変更してください
aidputils.widgets.text(
    name="raw_volume",
    defaultValue="/Volumes/blog_catalog/opendata/census_raw",
    label="入力Volume"
)

aidputils.widgets.text(
    name="work_volume",
    defaultValue="/Volumes/blog_catalog/opendata/census_work",
    label="出力Volume"
)

# 3つのNotebookで同じ値を使います
aidputils.widgets.text(
    name="run_id",
    defaultValue="manual_20261011_02",
    label="実行ID"
)

aidputils.widgets.text(
    name="expected_rows",
    defaultValue="62",
    label="期待する市区町村数"
)

RAW = aidputils.widgets.get("raw_volume").rstrip("/")
WORK = aidputils.widgets.get("work_volume").rstrip("/")
RUN_ID = aidputils.widgets.get("run_id")
EXPECTED_ROWS = int(aidputils.widgets.get("expected_rows"))

# 設定値のチェック
if not RAW.startswith("/Volumes/") or not WORK.startswith("/Volumes/"):
    raise ValueError("実在する /Volumes/... のPOSIXパスを設定してください")

if RAW == WORK or RAW.startswith(WORK + "/") or WORK.startswith(RAW + "/"):
    raise ValueError("入力・出力Volumeのパスを分離してください")

if (
    not re.fullmatch(r"[A-Za-z0-9_.-]{1,256}", RUN_ID)
    or RUN_ID in (".", "..")
):
    raise ValueError("run_idが未設定・未展開、または使用できない形式です")

RUN_ROOT = f"{WORK}/runs/{RUN_ID}"

print(json.dumps(
    {
        "raw": RAW,
        "run_root": RUN_ROOT,
        "expected_rows": EXPECTED_ROWS
    },
    ensure_ascii=False,
    indent=2
))

実行コマンド出力結果

{
  "raw": "/Volumes/blog_catalog/opendata/census_raw",
  "run_root": "/Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02",
  "expected_rows": 62
}

WidgetsはNotebookの入力欄です。Workflowでは同名のTask parameterから値を渡せます。画面に以前の値が残っている場合は、右側の「実行ID」欄を変更し、共通セルを再実行してRUN_IDとRUN_ROOTを更新します。Widgetの値と共通セルの出力が一致していることを確認します。公式：Notebooks / Widgets

■ Notebook 01：原本を読み込み、3つの中間データを作る

このNotebookでは、入力Volumeに置いた3つのExcel原本から、東京都62市区町村の人口・世帯・年齢データを取り出します。Excelに含まれる表題・注記・合計行を除き、Pythonで行と列を持つ表形式のデータ（DataFrame）へ整えます。

ここで作る「中間データ」は、次のNotebookで結合する前の3つの表です。人口はstage/population、世帯はstage/households、年齢はstage/ageへParquet形式で保存し、使用した原本の確認記録もsource_manifestへ保存します。

ここからはnb01_ingest_censusの共通セルの後へ、以下のセルを順に追加します。

● 原本の存在・サイズ・ハッシュを確認

入力Volumeに人口・世帯・年齢のExcelファイルがそろっていることを確認し、各ファイルのサイズとSHA-256を計算します。SHA-256はファイル内容から求める値で、同じ名前の原本が訂正された場合にも、前に使用した入力と内容が一致するかを確認できます。この結果をSOURCE_MANIFESTに保持し、後の保存処理で原本の確認記録として書き出します。

実行コマンド

import hashlib
FILES = ("population_1_1.xlsx", "households_9_1.xlsx", "age_2_7.xlsx")
SOURCE_MANIFEST = []
for name in FILES:
    path = Path(RAW) / name
    if not path.is_file():
        raise FileNotFoundError(path)
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    item = {"file": name, "bytes": path.stat().st_size, "sha256": digest.hexdigest()}
    SOURCE_MANIFEST.append(item)
    print(json.dumps(item, ensure_ascii=False))

実行コマンド出力結果

{"file": "population_1_1.xlsx", "bytes": 794110, "sha256": "91ba30c83013cf2f6efb9fcec8c3496be99b77ab0b800f97441648e9108215ab"}
{"file": "households_9_1.xlsx", "bytes": 4697251, "sha256": "343c7a2797071cb63215bcaee640e2b43c6b212efb3b31dd379cd84206981f25"}
{"file": "age_2_7.xlsx", "bytes": 8535914, "sha256": "114f2343dfaaa048614b0283830070b2048cae7da67a38df96c83039c6d0ac75"}

2026-09-29にローカル確認した原本のサイズは、表1-1が794,110 bytes、表9-1が4,697,251 bytes、表2-7が8,535,914 bytesでした。再取得で変わっていた場合は、更新・訂正と表の構造を確認します。

● Excelの表頭を見てみる

Excelのどの行からデータが始まり、どの列に地域コードや人口が入っているかを確認します。ここでは人口表の先頭18行を表示し、表題・注記・見出しと、市区町村のデータ行を見分けます。この確認をもとに、後述の関数で読み込む行・列と、東京都の市区町村を選ぶ条件を決めます。

実行コマンド

import openpyxl
with (Path(RAW) / "population_1_1.xlsx").open("rb") as stream:
    book = openpyxl.load_workbook(stream, read_only=True, data_only=True)
    sheet = book["b01_01"]
    for index, row in enumerate(sheet.iter_rows(min_row=1, max_row=18, values_only=True), 1):
        print(index, row[:8])
    book.close()

実行コマンド出力結果

1 ('令和２年国勢調査\u3000人口等基本集計', None, None, None, None, None, None, None)
2 ('第１－１表\u3000男女別人口，世帯の種類別世帯数及び世帯人員並びに2015年（平成27年）の人口（組替），2015年（平成27年）の世帯数（組替），5年間の人口増減数，5年間の人口増減率，5年間の世帯増減数，5年間の世帯増減率，人口性比，面積（参考）及び人口密度－全国，都道府県，市区町村（2000年（平成12年）市区町村含む）', None, None, None, None, None, None, None)
3 (None, None, None, None, None, None, None, None)
4 ('1) 「2015年（平成27年）の人口（組替）」及び「2015年（平成27年）の世帯数（組替）」は， 令和２年10月１日現在の市区町村の境域に基づいて組み替えた平成27年の人口及び世帯数を示す。', None, None, None, None, None, None, None)
5 ('2) 「5年間の人口増減数（率）」及び「5年間の世帯増減数（率）」については，それぞれ「2015年（平成27年）の人口（組替）」及び「2015年（平成27年）の世帯数（組替）」と比較したもの。', None, None, None, None, None, None, None)
6 ('3) 「人口性比」とは，女性100人に対する男性の数を示す。', None, None, None, None, None, None, None)
7 ('4) 地域名において，「旧：○○○」と表記されている項目は「2000年（平成12年）市区町村」を示す。「2000年（平成12年）市区町村」については，令和２年10月１日現在の市区町村の境域に基づいて平成12年10月１日時点の市区町村に組み替えた，令和２年10月１日現在の結果数値を表章している。', None, None, None, None, None, None, None)
8 ('5) 「面積（参考）」については，国土交通省国土地理院「令和２年全国都道府県市区町村別面積調（10月１日時点）」による。ただし，「2000年（平成12年）市区町村」については「平成12年全国都道府県市区町村別面積調」による。また，平成12年の境界未定地域の面積については，総務省統計局において面積を算出している。（面積に関する主な留意事項は「令和２年国勢調査\u3000面積及び人口密度に関する留意事項」を参照）', None, None, None, None, None, None, None)
9 ('6) 「人口密度」については，国勢調査施行規則第１条に基づいて調査対象外とした地域の面積を除いて算出。（詳細は「令和２年国勢調査\u3000面積及び人口密度に関する留意事項」を参照）', None, None, None, None, None, None, None)
10 (None, None, None, None, None, None, None, '人口')
11 (None, None, None, None, None, None, None, '男女')
12 (None, None, None, None, None, None, None, 1)
13 (None, None, None, None, None, None, None, '0_総数')
14 (None, None, None, None, None, None, None, '人')
15 ('地域識別コード', '2000年_都道府県', '2000年_地域コード', '2000年地域', '2020年_都道府県', '2020年_地域コード', '地域名', ' ')
16 ('a', '00_全国', '00000', '2000', '00_全国', '00000', '0001_全国', 126146099)
17 ('a', '01_北海道', '01000', '2000', '01_北海道', '01000', '0002_北海道', 5224614)
18 ('1', '01_北海道', '01100', '2000', '01_北海道', '01100', '0003_札幌市', 1973395)

今回の抽出ルールです。

原本	データ開始行	東京都の選択	使用する地域識別コード
表1-1	16行目	E列=`13_東京都`	A列=`0`,`2`,`3`
表9-1	11行目	B列=`13_東京都`	A列=`0`,`2`,`3`
表2-7	13行目	G列=`13_東京都`、国籍総数、男女総数	C列=`0`,`2`,`3`

これにより23特別区・26市・13町村を選択します。東京都合計・特別区部合計・旧市町村は除外します。西東京市と旧田無市・旧保谷市には同じ2020年地域コードがあるため、無条件の重複削除で済ませないことがポイントです。

● 原本を読む関数を定義

3つのExcelを読み、必要な行・列を取り出す処理を、read_census_tablesという関数にまとめます。原本ごとに表頭と列位置を確認し、東京都の対象地域を選んで、地域コード・人数・世帯数などの列を持つ表へ変換する手順を定義します。実際の読込みは、次の節でこの関数を呼び出して行います。

地域コードは文字列、人数・世帯数は整数として扱います。期待件数は共通セルのexpected_rowsを引数で受け取り、人口・世帯・年齢の3表と、照合用の東京都合計を返します。

国勢調査の公式ユーザーズガイド「利用上の注意」では、-を「該当数値がないもの」としています。本稿のコードでは、年齢不詳欄のこの記号を変換処理上いったんNULLとして保存します。このNULLは、年齢不詳の人数そのものが分からないことや、秘匿されていることを示すものではありません。後続のNotebook 02では、対象6地域について、年齢3区分の合計が総人口と一致することも検査します。原表の記号の意味と、今回のコードでの保存方法を分けて扱います。

実行コマンド

from pathlib import Path
import openpyxl
import pandas as pd


def read_census_tables(raw_dir, expected_rows):
    """保存済み2020年国勢調査Excel 3件から東京都の3表を抽出し、元表の東京都合計とあわせて返す。"""
    raw_dir = Path(raw_dir)
    marks = {"", "-", "－", "…", "...", "X", "x", "秘匿", "不詳"}

    def require(condition, message):
        if not bool(condition):
            raise ValueError(message)

    def read(filename, sheet_name, start, headers):
        with open(raw_dir / filename, "rb") as stream:
            book = openpyxl.load_workbook(stream, read_only=True, data_only=True)
            try:
                sheet = book[sheet_name]
                rows = list(sheet.values)
            finally:
                book.close()
        # 列位置を使う前に、元ファイルの表・ヘッダーが想定どおりか確認。
        for cell, expected in headers.items():
            column, row = openpyxl.utils.cell.coordinate_from_string(cell)
            actual = rows[row - 1][openpyxl.utils.column_index_from_string(column) - 1]
            require(actual == expected, f"{filename}: {cell} のヘッダー不一致: {actual!r}")
        names = [openpyxl.utils.get_column_letter(i + 1) for i in range(len(rows[0]))]
        frame = pd.DataFrame(rows[start - 1:], columns=names, dtype=object)
        frame["source_row"] = range(start, start + len(frame))
        return frame

    def number(series, dtype="Int64"):
        clean = series.map(lambda x: None if x is None or
                           (isinstance(x, str) and x.strip() in marks) else x)
        # 未知の文字列はエラー。欠損マーカーをゼロで埋めない。
        return pd.to_numeric(clean, errors="raise").astype(dtype)

    def join(left, right):
        require(left.area_code.notna().all() and right.area_code.notna().all(), "地域コードがNULL")
        require(left.area_code.is_unique and right.area_code.is_unique, "地域コードが重複")
        merged = left.merge(right, on="area_code", how="outer",
                            validate="one_to_one", indicator=True)
        unmatched = merged.loc[merged._merge != "both", "area_code"].tolist()
        require(not unmatched, f"結合できない地域コード: {unmatched}")
        require(len(merged) == expected_rows, f"結合後の行数が{expected_rows}ではない: {len(merged)}")
        return merged.drop(columns="_merge")

    # 0=特別区等、2=市、3=町村。東京都・特別区部の合計や旧市町村を除く。
    pop = read("population_1_1.xlsx", "b01_01", 16, {
        "A15": "地域識別コード", "F15": "2020年_地域コード", "G15": "地域名",
        "H10": "人口", "H13": "0_総数", "K10": "2015年（平成27年）の人口（組替）",
        "M10": "5年間の人口増減率", "Q13": "0_総数", "R13": "1_一般世帯"})
    tokyo = pop.loc[pop.E == "13_東京都"]
    p = tokyo.loc[tokyo.A.isin(["0", "2", "3"])]
    out = pd.DataFrame({
        "area_code": p.F.astype("string"),
        "area_name": p.G.str.split("_", n=1).str[1],
        "population_2020": number(p.H), "population_2015": number(p.K),
        "households_2020": number(p.Q), "general_households_2020": number(p.R),
        "population_change_pct_source": number(p.M, "Float64"),
        "year": 2020, "reference_date": "2020-10-01",
        "population_source_row": p.source_row})
    require(len(out) == expected_rows and out.area_code.is_unique, "人口表の行数または主キーが不正")
    require(out.area_code.str.fullmatch(r"13\d{3}").all(), "地域コード形式が不正")
    require(out.area_name.notna().all(), "地域名が欠損")

    hh = read("households_9_1.xlsx", "b09_01", 11, {
        "A10": "地域識別コード", "C10": "地域名", "E10": "世帯の家族類型",
        "F5": "一般世帯数", "F8": "0_総数"})
    h = hh.loc[(hh.B == "13_東京都") & hh.A.isin(["0", "2", "3"])].copy()
    h["area_code"] = h.C.str.split("_", n=1).str[0].astype("string")
    require(not h.duplicated(["area_code", "E"]).any(), "世帯表の地域・家族類型が重複")
    singles, totals = h.loc[h.E == "3_単独世帯"], h.loc[h.E == "0_総数"]
    household = join(pd.DataFrame({"area_code": singles.area_code,
        "single_households_2020": number(singles.F), "single_source_row": singles.source_row}),
        pd.DataFrame({"area_code": totals.area_code, "general_households_check": number(totals.F)}))

    age = read("age_2_7.xlsx", "b02_07", 13, {
        "A12": "国籍総数か日本人", "B12": "男女", "C12": "地域識別コード",
        "H12": "2020年_地域コード", "J10": "00_総数", "AF10": "22_年齢「不詳」",
        "AG10": "R1_（再掲）15歳未満", "AH10": "R2_（再掲）15～64歳",
        "AI10": "R3_（再掲）65歳以上", "AO10": "3_65歳以上"})
    a = age.loc[(age.G == "13_東京都") & age.C.isin(["0", "2", "3"])
                & (age.A == "0_国籍総数") & (age.B == "0_総数")]
    ages = pd.DataFrame({"area_code": a.H.astype("string"),
        "age_population_total_2020": number(a.J), "age_under15_2020": number(a.AG),
        "age_15_64_2020": number(a.AH), "age_65plus_2020": number(a.AI),
        "age_unknown_2020": number(a.AF),
        "age_65plus_share_known_source_pct": number(a.AO, "Float64"),
        "age_source_row": a.source_row})
    control = tokyo.loc[tokyo.F == "13000"]
    single_control = hh.loc[(hh.C == "13000_東京都") & (hh.E == "3_単独世帯")]
    require(len(control) == 1 and len(single_control) == 1, "東京都合計行が不正")
    c = control.iloc[0]
    tokyo_totals = {
        "population_2020": int(c.H), "population_2015": int(c.K),
        "households_2020": int(c.Q), "general_households_2020": int(c.R),
        "single_households_2020": int(single_control.iloc[0].F)}
    result = []
    for frame in (out, household, ages):
        require(len(frame) == expected_rows and frame.area_code.is_unique, "抽出結果の行数または主キーが不正")
        require(frame.area_code.str.fullmatch(r"13\d{3}").all(), "地域コード形式が不正")
        require(set(frame.area_code) == set(out.area_code), "3表の地域コードが不一致")
        result.append(frame.sort_values("area_code").reset_index(drop=True))
    return (*result, tokyo_totals)

print("read_census_tables: ready")

実行コマンド出力結果

read_census_tables: ready

● 3つのデータを抽出する

先ほど定義した関数を呼び出し、Excelから人口・世帯・年齢の3つのDataFrameを作ります。population_pd、households_pd、age_pdには、それぞれ同じ62市区町村のデータが入り、TOKYO_TOTALSには後で照合する東京都合計が入ります。出力のrowsとunique_codesがどちらも62であることを確認します。関数定義セルを実行してから、この抽出セルを実行します。

実行コマンド

population_pd, households_pd, age_pd, TOKYO_TOTALS = read_census_tables(RAW, EXPECTED_ROWS)
for name, frame in (("population", population_pd),
                    ("households", households_pd), ("age", age_pd)):
    print(name, "rows=", len(frame), "unique_codes=", frame.area_code.nunique())
    print(frame.head(3).to_string(index=False))
print("source_controls:", TOKYO_TOTALS)

実行コマンド出力結果

population rows= 62 unique_codes= 62
area_code area_name  population_2020  population_2015  households_2020  general_households_2020  population_change_pct_source  year reference_date  population_source_row
    13101      千代田区            66680            58406            37011                    36963                      14.16635  2020     2020-10-01                   1229
    13102       中央区           169179           141183            92533                    92487                      19.82958  2020     2020-10-01                   1230
    13103        港区           260486           243283           146160                   146084                       7.07119  2020     2020-10-01                   1231
households rows= 62 unique_codes= 62
area_code  single_households_2020  single_source_row  general_households_check
    13101                   21076              26962                     36963
    13102                   48646              27002                     92487
    13103                   83099              27042                    146084
age rows= 62 unique_codes= 62
area_code  age_population_total_2020  age_under15_2020  age_15_64_2020  age_65plus_2020  age_unknown_2020  age_65plus_share_known_source_pct  age_source_row
    13101                      66680              8976           46517            10852               335                           16.35692            1226
    13102                     169179             23086          121217            24683               193                           14.60654            1227
    13103                     260486             34714          174886            42546              8340                           16.87356            1228
source_controls: {'population_2020': 14047594, 'population_2015': 13515272, 'households_2020': 7227180, 'general_households_2020': 7216650, 'single_households_2020': 3625810}

● Spark DataFrameへ変換し、Parquetで保存

Pythonで作った3つの表を、次のNotebookでSparkが読み込める形にして保存します。地域コード・人数・割合の型を明示してSpark DataFrameへ変換し、実行IDごとのstage/配下へ人口・世帯・年齢のParquetを書き出します。さらに、原本のサイズ・ハッシュと東京都合計をsource_manifestに保存し、どの入力から作ったデータかを後続処理で確認できるようにします。

地域コードを数値に推論させず、NULLを含む整数も型を明示します。62行の小規模データなので、Pythonからレコードを渡す方法を使います。

Parquetは列単位でデータを保存するファイル形式です。Sparkが書き込む先は、単一ファイルではなく複数のpartファイルを含むディレクトリになります。

実行コマンド

import pandas as pd
from pyspark.sql import types as T

def to_spark(frame):
    strings = {"area_code", "area_name", "reference_date"}
    fields = []
    for col in frame.columns:
        kind = T.StringType() if col in strings else (
            T.DoubleType() if "pct" in col else T.LongType())
        fields.append(T.StructField(col, kind, True))
    schema = T.StructType(fields)
    records = []
    for row in frame.itertuples(index=False, name=None):
        record = []
        for value, field in zip(row, schema.fields):
            if pd.isna(value):
                record.append(None)
            elif isinstance(field.dataType, T.StringType):
                record.append(str(value))
            elif isinstance(field.dataType, T.DoubleType):
                record.append(float(value))
            else:
                record.append(int(value))
        records.append(tuple(record))
    return spark.createDataFrame(records, schema)

for name, frame in (("population", population_pd),
                    ("households", households_pd), ("age", age_pd)):
    sdf = to_spark(frame)
    target = f"{RUN_ROOT}/stage/{name}"
    sdf.write.mode("errorifexists").parquet(target)
    sdf.printSchema()
    print(name, "saved_rows=", spark.read.parquet(target).count(), "path=", target)

manifest = {"run_id": RUN_ID, "sources": SOURCE_MANIFEST,
            "controls": TOKYO_TOTALS,
            "python_version": __import__("sys").version, "spark_version": spark.version}
spark.createDataFrame([(json.dumps(manifest, ensure_ascii=False),)], ["value"]) \
    .coalesce(1).write.mode("errorifexists").text(f"{RUN_ROOT}/source_manifest")
print("nb01_ingest_census: completed")

実行コマンド出力結果（抜粋）

root
 |-- area_code: string (nullable = true)
 |-- area_name: string (nullable = true)
 |-- population_2020: long (nullable = true)
 |-- population_2015: long (nullable = true)
 |-- households_2020: long (nullable = true)
 |-- general_households_2020: long (nullable = true)
 |-- population_change_pct_source: double (nullable = true)
 |-- year: long (nullable = true)
 |-- reference_date: string (nullable = true)
 |-- population_source_row: long (nullable = true)


population saved_rows= 62 path= /Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02/stage/population

root
 |-- area_code: string (nullable = true)
 |-- single_households_2020: long (nullable = true)
 |-- single_source_row: long (nullable = true)
 |-- general_households_check: long (nullable = true)


households saved_rows= 62 path= /Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02/stage/households

root
 |-- area_code: string (nullable = true)
 |-- age_population_total_2020: long (nullable = true)
 |-- age_under15_2020: long (nullable = true)
 |-- age_15_64_2020: long (nullable = true)
 |-- age_65plus_2020: long (nullable = true)
 |-- age_unknown_2020: long (nullable = true)
 |-- age_65plus_share_known_source_pct: double (nullable = true)
 |-- age_source_row: long (nullable = true)


age saved_rows= 62 path= /Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02/stage/age

nb01_ingest_census: completed

既存パスへの書込みはエラーにしています。同じNotebookを再実行して「path already exists」になった場合は、原本を削除せず、3つのNotebookのrun_idを新しくして最初から実行します。

画面の実行IDは撮影時のものです。

■ Notebook 02：Sparkで結合・計算・品質検査する

このNotebookでは、Notebook 01が保存した人口・世帯・年齢の3表を地域コードで結合し、1市区町村につき1行の分析用データを作ります。人数・世帯数から比較用の割合を計算し、件数・合計・欠損・公表値との一致を検査してから保存します。

出力はcurated/tokyo_comparisonの分析用Parquetと、validationの検査記録です。Notebook 03はこの検査記録を確認し、合格したデータを分析に使います。

nb02_transform_censusに共通パラメータセルを置き、その後へ以下を追加します。Notebook 01と同じrun_idで実行します。

共通パラメータセル（nb02_transform_census：クリックして展開）

実行コマンド

import re
import json
from pathlib import Path

# 実際のCatalog・Schema・Volume名に変更してください
aidputils.widgets.text(
    name="raw_volume",
    defaultValue="/Volumes/blog_catalog/opendata/census_raw",
    label="入力Volume"
)

aidputils.widgets.text(
    name="work_volume",
    defaultValue="/Volumes/blog_catalog/opendata/census_work",
    label="出力Volume"
)

# 3つのNotebookで同じ値を使います
aidputils.widgets.text(
    name="run_id",
    defaultValue="manual_20261011_02",
    label="実行ID"
)

aidputils.widgets.text(
    name="expected_rows",
    defaultValue="62",
    label="期待する市区町村数"
)

RAW = aidputils.widgets.get("raw_volume").rstrip("/")
WORK = aidputils.widgets.get("work_volume").rstrip("/")
RUN_ID = aidputils.widgets.get("run_id")
EXPECTED_ROWS = int(aidputils.widgets.get("expected_rows"))

# 設定値のチェック
if not RAW.startswith("/Volumes/") or not WORK.startswith("/Volumes/"):
    raise ValueError("実在する /Volumes/... のPOSIXパスを設定してください")

if RAW == WORK or RAW.startswith(WORK + "/") or WORK.startswith(RAW + "/"):
    raise ValueError("入力・出力Volumeのパスを分離してください")

if (
    not re.fullmatch(r"[A-Za-z0-9_.-]{1,256}", RUN_ID)
    or RUN_ID in (".", "..")
):
    raise ValueError("run_idが未設定・未展開、または使用できない形式です")

RUN_ROOT = f"{WORK}/runs/{RUN_ID}"

print(json.dumps(
    {
        "raw": RAW,
        "run_root": RUN_ROOT,
        "expected_rows": EXPECTED_ROWS
    },
    ensure_ascii=False,
    indent=2
))

共通セルの出力で、run_rootがNotebook 01と同じであることを確認します。Widgetに旧値が残る場合の確認箇所は後述します。

● 中間Parquetを読み込む

Notebook 01が保存した3表と原本の確認記録を、同じ実行IDの出力先から読み込みます。各表が62行で、地域コードに欠損・重複・形式の違いがなく、3表に同じ地域がそろっていることを検査します。keys: PASSを確認してから結合へ進むことで、JOINによる行の欠落や増加を防ぎます。

実行コマンド

from pyspark.sql import functions as F

p = spark.read.parquet(f"{RUN_ROOT}/stage/population")
h = spark.read.parquet(f"{RUN_ROOT}/stage/households")
a = spark.read.parquet(f"{RUN_ROOT}/stage/age")
manifest_text = spark.read.text(f"{RUN_ROOT}/source_manifest").first()[0]
manifest = json.loads(manifest_text)
if manifest["run_id"] != RUN_ID:
    raise ValueError("別runの入力台帳です")

def reject_if_rows(frame, message):
    if frame.limit(1).count() > 0:
        raise ValueError(message)

for name, df in (("population", p), ("households", h), ("age", a)):
    n = df.count()
    print(name, "rows=", n)
    if n != EXPECTED_ROWS:
        raise ValueError(f"{name}: expected_rows={EXPECTED_ROWS}, actual_rows={n}")
    reject_if_rows(df.filter(F.col("area_code").isNull() |
                            ~F.col("area_code").rlike(r"^13[0-9]{3}$")), "地域コード形式が不正")
    reject_if_rows(df.groupBy("area_code").count().filter("count <> 1"), "地域コード重複")
for other in (h, a):
    reject_if_rows(p.select("area_code").join(other.select("area_code"), "area_code", "left_anti"),
                   "人口表にしかないコードがあります")
    reject_if_rows(other.select("area_code").join(p.select("area_code"), "area_code", "left_anti"),
                   "人口表にないコードがあります")
print("keys: PASS")

実行コマンド出力結果

population rows= 62
households rows= 62
age rows= 62
keys: PASS

left_antiは左側にだけ存在する行を返します。Oracle SQLのNOT EXISTSを使った未結合チェックに近い処理です。JOINする前にキーを検査し、行の欠落・増加を防ぎます。

● 地域コードでJOINし、元表同士を照合

area_codeを共通のキーとして、人口・世帯・年齢の3表を横に結合します。たとえば中央区の人口と世帯数、年齢別人数が、同じ地域コードの1行にまとまります。結合後も62行であることに加え、別々の原本にある一般世帯数や総人口が一致することを確認し、地域や列の取り違えを検知します。

実行コマンド

joined = p.join(h, "area_code", "inner").join(a, "area_code", "inner")
if joined.count() != EXPECTED_ROWS:
    raise ValueError("JOIN後の件数が変わりました")
reject_if_rows(joined.filter(~F.col("general_households_2020").eqNullSafe(
    F.col("general_households_check"))), "表1-1と表9-1の一般世帯数が不一致")
reject_if_rows(joined.filter(~F.col("population_2020").eqNullSafe(
    F.col("age_population_total_2020"))), "人口表と年齢表の人口が不一致")
joined = joined.drop("general_households_check", "age_population_total_2020")
joined.createOrReplaceTempView("tokyo_joined")
print("joined_rows:", joined.count())
joined.orderBy("area_code").show(5, truncate=False)

実行コマンド出力結果

joined_rows: 62

+---------+---------+---------------+---------------+---------------+-----------------------+----------------------------+----+--------------+---------------------+----------------------+-----------------+----------------+--------------+---------------+----------------+---------------------------------+--------------+
|area_code|area_name|population_2020|population_2015|households_2020|general_households_2020|population_change_pct_source|year|reference_date|population_source_row|single_households_2020|single_source_row|age_under15_2020|age_15_64_2020|age_65plus_2020|age_unknown_2020|age_65plus_share_known_source_pct|age_source_row|
+---------+---------+---------------+---------------+---------------+-----------------------+----------------------------+----+--------------+---------------------+----------------------+-----------------+----------------+--------------+---------------+----------------+---------------------------------+--------------+
|13101    |千代田区 |66680          |58406          |37011          |36963                  |14.16635                    |2020|2020-10-01    |1229                 |21076                 |26962            |8976            |46517         |10852          |335             |16.35692                         |1226          |
|13102    |中央区   |169179         |141183         |92533          |92487                  |19.82958                    |2020|2020-10-01    |1230                 |48646                 |27002            |23086           |121217        |24683          |193             |14.60654                         |1227          |
|13103    |港区     |260486         |243283         |146160         |146084                 |7.07119                     |2020|2020-10-01    |1231                 |83099                 |27042            |34714           |174886        |42546          |8340            |16.87356                         |1228          |
|13104    |新宿区   |349385         |333560         |222800         |222461                 |4.74427                     |2020|2020-10-01    |1232                 |150831                |27082            |29202           |223593        |63185          |33405           |19.99652                         |1229          |
|13105    |文京区   |240069         |219724         |133661         |133564                 |9.25934                     |2020|2020-10-01    |1233                 |77316                 |27122            |28065           |152818        |42020          |17166           |18.85125                         |1230          |
+---------+---------+---------------+---------------+---------------+-----------------------+----------------------------+----+--------------+---------------------+----------------------+-----------------+----------------+--------------+---------------+----------------+---------------------------------+--------------+
only showing top 5 rows

ここで初めて人口・世帯・年齢が一つの表になります。DataFrameを一時ビューtokyo_joinedとして登録すると、同じSpark SessionからSQLで参照できます。このビューは永続的なCatalogの表ではありません。

● Spark SQLで指標を計算

結合した人数・世帯数から、人口増減率、単独世帯割合、高齢者割合、年齢不詳割合の4つの指標を計算します。比較に使う割合の列を加えることで、人口規模が異なる地域も同じ指標で並べられるようになります。人口増減率は2015年人口を基準とする増減、年齢不詳割合は総人口に占める年齢不詳の割合です。

単独世帯割合の分母は一般世帯数、高齢者割合の分母は年齢が分かる人の合計です。総世帯数や総人口を無条件に分母へ代用しません。

実行コマンド

result = spark.sql("""
SELECT *,
  CASE WHEN population_2015 > 0 THEN
    100.0 * (population_2020 - population_2015) / population_2015
  END AS population_change_pct,
  CASE WHEN general_households_2020 > 0 THEN
    100.0 * single_households_2020 / general_households_2020
  END AS single_households_pct,
  CASE WHEN age_under15_2020 + age_15_64_2020 + age_65plus_2020 > 0 THEN
    100.0 * age_65plus_2020 /
    (age_under15_2020 + age_15_64_2020 + age_65plus_2020)
  END AS age_65plus_share_known_pct,
  CASE WHEN population_2020 > 0 THEN
    100.0 * age_unknown_2020 / population_2020
  END AS age_unknown_pct
FROM tokyo_joined
""")
for c in ("population_change_pct", "single_households_pct",
          "age_65plus_share_known_pct", "age_unknown_pct"):
    result = result.withColumn(c, F.col(c).cast("double"))
result.select("area_code", "area_name", "population_change_pct",
              "single_households_pct", "age_unknown_pct") \
      .orderBy("area_code").show(5, truncate=False)

実行コマンド出力結果

+---------+---------+---------------------+---------------------+----------------+
|area_code|area_name|population_change_pct|single_households_pct|age_unknown_pct |
+---------+---------+---------------------+---------------------+----------------+
|13101    |千代田区 |14.16635277197548    |57.01918134350567    |0.50239952009598|
|13102    |中央区   |19.82958288179172    |52.59766237417151    |0.11408035276246|
|13103    |港区     |7.07118869793615     |56.88439527942827    |3.2017075773746 |
|13104    |新宿区   |4.74427389375225     |67.80109772049931    |9.56108590809565|
|13105    |文京区   |9.25934354007755     |57.88685573957054    |7.15044424727891|
+---------+---------+---------------------+---------------------+----------------+
only showing top 5 rows

joinやselectは主に処理内容を定義します。show()・count()・書込みなどが実行のきっかけになる、というSparkの遅延評価もここで体験できます。

● 件数・合計・欠損・公表値を検査

計算結果を保存する前に、原本から必要なデータを正しく取り出せたかを検査します。人口・世帯数の合計を元表の東京都合計と固定した期待値へ照合し、欠損・負の値、世帯数の大小関係、年齢内訳の合計、原表の記号に由来するNULLの地域も確認します。人口増減率と高齢者割合は公表値と比較し、すべての検査を通過した場合にvalidation: PASSを出力します。

実行コマンド

FIXED_TOTALS = {
    "population_2020": 14047594, "population_2015": 13515272,
    "households_2020": 7227180, "general_households_2020": 7216650,
    "single_households_2020": 3625810,
}
if manifest["controls"] != FIXED_TOTALS:
    raise ValueError("原本が今回の固定snapshotと異なります。訂正・版を確認してください")
for col, expected in FIXED_TOTALS.items():
    reject_if_rows(result.filter(F.col(col).isNull() | (F.col(col) < 0)), f"{col}: 欠損・負値")
    total = result.agg(F.sum(col)).first()[0]
    print(col, "sum=", total, "expected=", expected)
    if total != expected:
        raise ValueError(f"{col}: 合計不一致")
reject_if_rows(result.filter(F.col("general_households_2020") > F.col("households_2020")),
               "一般世帯が総世帯を超えています")
reject_if_rows(result.filter(F.col("single_households_2020") > F.col("general_households_2020")),
               "単独世帯が一般世帯を超えています")
for c in ("age_under15_2020", "age_15_64_2020", "age_65plus_2020"):
    reject_if_rows(result.filter(F.col(c).isNull() | (F.col(c) < 0)), f"{c}: 欠損・負値")
known = F.col("age_under15_2020") + F.col("age_15_64_2020") + F.col("age_65plus_2020")
reject_if_rows(result.filter(F.col("age_unknown_2020").isNotNull() &
    ((known + F.col("age_unknown_2020")) != F.col("population_2020"))), "年齢内訳不一致")
# 年齢不詳をNULLにした地域は、既知年齢の合計が総人口と一致することを確認
reject_if_rows(result.filter(F.col("age_unknown_2020").isNull() &
    (known != F.col("population_2020"))), "年齢不詳NULL地域で既知年齢の合計が総人口と不一致")
unknown_codes = {r.area_code for r in result.filter("age_unknown_2020 IS NULL").select("area_code").collect()}
if unknown_codes != {"13361", "13362", "13363", "13364", "13382", "13402"}:
    raise ValueError("原表記号に由来するNULLの地域が変わりました")
for calculated, source in (("population_change_pct", "population_change_pct_source"),
                           ("age_65plus_share_known_pct", "age_65plus_share_known_source_pct")):
    reject_if_rows(result.filter(F.col(calculated).isNull() | F.col(source).isNull() |
        (F.abs(F.col(calculated) - F.col(source)) > 0.0000051)), "公表割合との照合に失敗")
print("age_unknown NULL codes:", sorted(unknown_codes))
print("validation: PASS")

実行コマンド出力結果

population_2020 sum= 14047594 expected= 14047594
population_2015 sum= 13515272 expected= 13515272
households_2020 sum= 7227180 expected= 7227180
general_households_2020 sum= 7216650 expected= 7216650
single_households_2020 sum= 3625810 expected= 3625810
age_unknown NULL codes: ['13361', '13362', '13363', '13364', '13382', '13402']
validation: PASS

照合に使った期待値です。上のAIDP実行結果で合計5項目の一致と、年齢不詳をNULLにした地域コード6件を確認できました。年齢不詳をNULLにした6地域では、既知年齢（15歳未満・15〜64歳・65歳以上）の合計が総人口と一致することも確認できました。これは今回の原本の数値照合結果で、原表の記号やNULLを一律にゼロへ置き換える規則にはしていません。

指標	期待値
市区町村数	62
2020人口合計	14,047,594人
2015組替人口合計	13,515,272人
総世帯数	7,227,180世帯
一般世帯数	7,216,650世帯
単独世帯数	3,625,810世帯
年齢不詳欄の「-」を変換処理でNULLとして保存した自治体数	6

これらは今回の固定2020年データ用です。別年度や別地域へ展開する際は、期待値・コード集合・表のレイアウトを更新します。

● 分析用Parquetと検査記録を保存

検査を通過した62市区町村の分析用データを、curated/tokyo_comparisonへ保存します。保存したParquetを読み直し、件数と各行の値が保存前と一致することを確認してから、validationへPASS・実行ID・件数・保存先などの検査記録を書き出します。この記録をNotebook 03の開始時に確認し、検査完了したデータを分析へ渡します。

実行コマンド

TARGET = f"{RUN_ROOT}/curated/tokyo_comparison"
result.write.mode("errorifexists").parquet(TARGET)
saved = spark.read.parquet(TARGET).select(*result.columns)
if saved.count() != EXPECTED_ROWS:
    raise ValueError("保存後の件数不一致")
if result.exceptAll(saved).limit(1).count() or saved.exceptAll(result).limit(1).count():
    raise ValueError("保存後の値が変化しました")

quality = {"status": "PASS", "run_id": RUN_ID, "rows": saved.count(),
           "totals": FIXED_TOTALS, "age_unknown_nulls": len(unknown_codes),
           "target": TARGET, "spark_version": spark.version}
spark.createDataFrame([(json.dumps(quality, ensure_ascii=False),)], ["value"]) \
    .coalesce(1).write.mode("errorifexists").text(f"{RUN_ROOT}/validation")
print(json.dumps(quality, ensure_ascii=False, indent=2))
print("nb02_transform_census: completed")

実行コマンド出力結果

{
  "status": "PASS",
  "run_id": "manual_20261011_02",
  "rows": 62,
  "totals": {
    "population_2020": 14047594,
    "population_2015": 13515272,
    "households_2020": 7227180,
    "general_households_2020": 7216650,
    "single_households_2020": 3625810
  },
  "age_unknown_nulls": 6,
  "target": "/Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02/curated/tokyo_comparison",
  "spark_version": "3.5.0"
}
nb02_transform_census: completed

この例はrunごとに別のディレクトリへ保存する簡単な方式です。複数ファイルをまとめた原子的な公開処理や、本番用の同時更新制御を実装したものではありません。
また、Volumeで管理する保存範囲に同じパスのExternal Tableを重ねて登録しません。本稿はParquetの再読込み＋一時ビューで分析します。永続Catalog表やDelta／Icebergの管理は別の検証として追加できます。

■ Notebook 03：Spark SQLで地域の特徴を分析する

このNotebookでは、検査済みの62市区町村のデータをSpark SQLで比較し、人口・人口増減率・単独世帯割合・高齢者割合のランキングを作ります。人口上位10件はグラフにし、全62行は他のツールでも扱えるCSVとして保存・再読込みします。

nb03_analyze_censusでも、Notebook 01と同じ共通パラメータセルを先頭に置いて実行します。入力はNotebook 02が保存したParquetです。

共通パラメータセル（nb03_analyze_census：クリックして展開）

実行コマンド

import re
import json
from pathlib import Path

# 実際のCatalog・Schema・Volume名に変更してください
aidputils.widgets.text(
    name="raw_volume",
    defaultValue="/Volumes/blog_catalog/opendata/census_raw",
    label="入力Volume"
)

aidputils.widgets.text(
    name="work_volume",
    defaultValue="/Volumes/blog_catalog/opendata/census_work",
    label="出力Volume"
)

# 3つのNotebookで同じ値を使います
aidputils.widgets.text(
    name="run_id",
    defaultValue="manual_20261011_02",
    label="実行ID"
)

aidputils.widgets.text(
    name="expected_rows",
    defaultValue="62",
    label="期待する市区町村数"
)

RAW = aidputils.widgets.get("raw_volume").rstrip("/")
WORK = aidputils.widgets.get("work_volume").rstrip("/")
RUN_ID = aidputils.widgets.get("run_id")
EXPECTED_ROWS = int(aidputils.widgets.get("expected_rows"))

# 設定値のチェック
if not RAW.startswith("/Volumes/") or not WORK.startswith("/Volumes/"):
    raise ValueError("実在する /Volumes/... のPOSIXパスを設定してください")

if RAW == WORK or RAW.startswith(WORK + "/") or WORK.startswith(RAW + "/"):
    raise ValueError("入力・出力Volumeのパスを分離してください")

if (
    not re.fullmatch(r"[A-Za-z0-9_.-]{1,256}", RUN_ID)
    or RUN_ID in (".", "..")
):
    raise ValueError("run_idが未設定・未展開、または使用できない形式です")

RUN_ROOT = f"{WORK}/runs/{RUN_ID}"

print(json.dumps(
    {
        "raw": RAW,
        "run_root": RUN_ROOT,
        "expected_rows": EXPECTED_ROWS
    },
    ensure_ascii=False,
    indent=2
))

実行コマンド出力結果

{
  "raw": "/Volumes/blog_catalog/opendata/census_raw",
  "run_root": "/Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02",
  "expected_rows": 62
}

run_rootがNotebook 01・02と同じであることを確認します。Parquetの読込みは次のセルで行います。

● 検査結果を確認し、一時ビューを作る

Notebook 02の検査記録を読み、PASSであること、実行IDと件数が一致することを確認して、記録された保存先のParquetを読み込みます。そのDataFrameにtokyo_comparisonという一時ビュー名を付け、以降のSQLでFROM tokyo_comparisonとして参照できるようにします。一時ビューは、このNotebookでSQLを実行するための参照名です。

実行コマンド

quality = json.loads(spark.read.text(f"{RUN_ROOT}/validation").first()[0])
if quality["status"] != "PASS" or quality["run_id"] != RUN_ID or quality["rows"] != EXPECTED_ROWS:
    raise ValueError("分析可能な検査完了データではありません")
tokyo = spark.read.parquet(quality["target"])
if tokyo.count() != EXPECTED_ROWS:
    raise ValueError("分析対象の件数が変わりました")
tokyo.createOrReplaceTempView("tokyo_comparison")
print("run_id:", RUN_ID)
print("rows:", tokyo.count())
tokyo.printSchema()

実行コマンド出力結果

run_id: manual_20261011_02

rows: 62
root
 |-- area_code: string (nullable = true)
 |-- area_name: string (nullable = true)
 |-- population_2020: long (nullable = true)
 |-- population_2015: long (nullable = true)
 |-- households_2020: long (nullable = true)
 |-- general_households_2020: long (nullable = true)
 |-- population_change_pct_source: double (nullable = true)
 |-- year: long (nullable = true)
 |-- reference_date: string (nullable = true)
 |-- population_source_row: long (nullable = true)
 |-- single_households_2020: long (nullable = true)
 |-- single_source_row: long (nullable = true)
 |-- age_under15_2020: long (nullable = true)
 |-- age_15_64_2020: long (nullable = true)
 |-- age_65plus_2020: long (nullable = true)
 |-- age_unknown_2020: long (nullable = true)
 |-- age_65plus_share_known_source_pct: double (nullable = true)
 |-- age_source_row: long (nullable = true)
 |-- population_change_pct: double (nullable = true)
 |-- single_households_pct: double (nullable = true)
 |-- age_65plus_share_known_pct: double (nullable = true)
 |-- age_unknown_pct: double (nullable = true)

一時ビューをこのNotebook内で作り直すので、Notebook 02のSpark Sessionに依存しません。以下のSQLはOracle Databaseではなく、AIDPのSparkで実行されます。

● 人口の多い市区町村を調べる

2020年の人口が多い順に並べ、上位10市区町村の地域コード・地域名・人口・総世帯数を表示します。まず地域ごとの人口規模を把握し、後の人口増減率や世帯割合のランキングと見比べるための基準にします。

実行コマンド

population_top10 = spark.sql("""
SELECT area_code, area_name, population_2020, households_2020
FROM tokyo_comparison
ORDER BY population_2020 DESC, area_code
LIMIT 10
""")
population_top10.show(truncate=False)

実行コマンド出力結果

+---------+---------+---------------+---------------+
|area_code|area_name|population_2020|households_2020|
+---------+---------+---------------+---------------+
|13112    |世田谷区 |943664         |492065         |
|13120    |練馬区   |752608         |374842         |
|13111    |大田区   |748081         |400164         |
|13123    |江戸川区 |697932         |333200         |
|13121    |足立区   |695043         |345346         |
|13115    |杉並区   |591108         |336339         |
|13119    |板橋区   |584483         |314446         |
|13201    |八王子市 |579355         |267020         |
|13108    |江東区   |524310         |264278         |
|13122    |葛飾区   |453093         |215948         |
+---------+---------+---------------+---------------+

照合用の上位3件は、世田谷区943,664人、練馬区752,608人、大田区748,081人です。人口の規模を比較できますが、これだけでサービス需要や事業効果を推定することはできません。

● 2015年から2020年に人口が増えた地域を調べる

2015年から2020年の人口増減率が高い順に、上位10市区町村を表示します。2015年人口と2020年人口、増減率を並べることで、人口の多さに加えて、5年間でどの程度増えたかを比較できます。比較には、2020年の境域に組み替えた2015年人口を使います。

実行コマンド

spark.sql("""
SELECT area_code, area_name, population_2015, population_2020,
       ROUND(population_change_pct, 2) AS population_change_pct
FROM tokyo_comparison
ORDER BY population_change_pct DESC, area_code
LIMIT 10
""").show(truncate=False)

実行コマンド出力結果: 5年間の人口増減率ランキング

+---------+---------+---------------+---------------+---------------------+
|area_code|area_name|population_2015|population_2020|population_change_pct|
+---------+---------+---------------+---------------+---------------------+
|13102    |中央区   |141183         |169179         |19.83                |
|13101    |千代田区 |58406          |66680          |14.17                |
|13105    |文京区   |219724         |240069         |9.26                 |
|13109    |品川区   |386855         |422488         |9.21                 |
|13113    |渋谷区   |224533         |243883         |8.62                 |
|13103    |港区     |243283         |260486         |7.07                 |
|13106    |台東区   |198073         |211444         |6.75                 |
|13225    |稲城市   |87636          |93151          |6.29                 |
|13107    |墨田区   |256274         |272085         |6.17                 |
|13208    |調布市   |229061         |242614         |5.92                 |
+---------+---------+---------------+---------------+---------------------+

照合用の上位は中央区19.83%、千代田区14.17%、文京区9.26%です。異なる時点の人口を同じ境域に揃えて比較しています。

● 単独世帯の割合と世帯数を比較する

一般世帯に占める単独世帯の割合が高い順に、上位10市区町村を表示します。割合だけでなく単独世帯数と一般世帯数も並べ、単独世帯の数が多い地域と、世帯全体に占める比率が高い地域を読み分けます。

実行コマンド

spark.sql("""
SELECT area_code, area_name, single_households_2020, general_households_2020,
       ROUND(single_households_pct, 2) AS single_households_pct
FROM tokyo_comparison
ORDER BY single_households_pct DESC, area_code
LIMIT 10
""").show(truncate=False)

実行コマンド出力結果: 単独世帯数・一般世帯数・割合のランキング

+---------+---------+----------------------+-----------------------+---------------------+
|area_code|area_name|single_households_2020|general_households_2020|single_households_pct|
+---------+---------+----------------------+-----------------------+---------------------+
|13402    |青ヶ島村 |90                    |118                    |76.27                |
|13104    |新宿区   |150831                |222461                 |67.8                 |
|13113    |渋谷区   |96707                 |149856                 |64.53                |
|13382    |御蔵島村 |120                   |187                    |64.17                |
|13116    |豊島区   |117608                |183687                 |64.03                |
|13362    |利島村   |121                   |193                    |62.69                |
|13114    |中野区   |129649                |207944                 |62.35                |
|13381    |三宅村   |819                   |1376                   |59.52                |
|13421    |小笠原村 |867                   |1457                   |59.51                |
|13106    |台東区   |71778                 |121557                 |59.05                |
+---------+---------+----------------------+-----------------------+---------------------+

青ヶ島村は90／118世帯で76.27%、新宿区は150,831／222,461世帯で67.80%です。割合が高いことと、世帯数が多いことは分けて読みます。

● 年齢不詳の影響も確認する

年齢が分かる人を分母にした高齢者割合の上位10市区町村を表示し、年齢不詳の人数と総人口に占める割合を併記します。高齢者割合を比べる際に、その計算から除かれる年齢不詳がどの程度あるかも確認するためです。原表の「-」をNULLへ変換した地域は、保存した値に合わせて年齢不詳の人数・割合をNULLで表示します。

実行コマンド

spark.sql("""
SELECT area_code, area_name, age_65plus_2020,
       ROUND(age_65plus_share_known_pct, 2) AS age_65plus_share_known_pct,
       age_unknown_2020, ROUND(age_unknown_pct, 2) AS age_unknown_pct
FROM tokyo_comparison
ORDER BY age_65plus_share_known_pct DESC, area_code
LIMIT 10
""").show(truncate=False)

実行コマンド出力結果: 高齢者割合と年齢不詳のランキング

+---------+----------+---------------+--------------------------+----------------+---------------+
|area_code|area_name |age_65plus_2020|age_65plus_share_known_pct|age_unknown_2020|age_unknown_pct|
+---------+----------+---------------+--------------------------+----------------+---------------+
|13307    |檜原村    |1058           |53.03                     |8               |0.4            |
|13308    |奥多摩町  |2410           |50.79                     |5               |0.11           |
|13401    |八丈町    |2801           |39.93                     |27              |0.38           |
|13363    |新島村    |967            |39.61                     |NULL            |NULL           |
|13381    |三宅村    |894            |39.38                     |3               |0.13           |
|13305    |日の出町  |6538           |38.76                     |90              |0.53           |
|13361    |大島町    |2713           |38.2                      |NULL            |NULL           |
|13205    |青梅市    |42492          |31.96                     |576             |0.43           |
|13364    |神津島村  |587            |31.64                     |NULL            |NULL           |
|13228    |あきる野市|23792          |31.23                     |3116            |3.93           |
+---------+----------+---------------+--------------------------+----------------+---------------+

Notebook 03の手動実行で、高齢者割合の上位10件と年齢不詳の人数・割合を確認できました。檜原村は53.03%、奥多摩町は50.79%です。この高齢者割合の分母は、年齢が分かる人の合計です。

あきる野市では、年齢不詳が3,116人、総人口に占める割合が3.93%でした。高齢者割合を比較するときは、このように年齢不詳の割合も併記します。

上位10件には新島村・大島町・神津島村のNULLも表示されています。これは、公式には「該当数値がないもの」を表す「-」を、本稿の変換処理でNULLとして保存したためです。年齢不詳の人数自体が分からないという意味ではありません。NULLとして保存した全6地域では、既知年齢の合計が総人口と一致することをNotebook 02で検査しています。高齢者割合は年齢3区分の合計を分母にするため、この保存方法でも計算でき、公表割合との照合も通過しています。

● グラフにする

人口上位10市区町村を棒グラフにし、表の数値だけでは見比べにくい人口規模の差を確認します。横軸は地域コード、縦軸は人口（人）です。Sparkの計算結果をcollect()で10件だけ取得し、Pythonのリストをmatplotlibへ渡して描画します。

日本語フォントがない環境でも表示できるよう、横軸は地域コードにしています。グラフ直下に市区町村名との対応表を載せます。

実行コマンド

import matplotlib.pyplot as plt

# 上位10件をPythonのリストとして取得する
chart_rows = spark.sql("""
SELECT area_code, population_2020
FROM tokyo_comparison
ORDER BY population_2020 DESC, area_code
LIMIT 10
""").collect()

area_codes = [row["area_code"] for row in chart_rows]
populations = [int(row["population_2020"]) for row in chart_rows]
print("chart_rows:", len(chart_rows))

fig, ax = plt.subplots(figsize=(10, 5))
ax.bar(area_codes, populations, color="#218C8D")
ax.set_title("Tokyo municipalities: population top 10 (2020 Census)")
ax.set_xlabel("Municipality code")
ax.set_ylabel("Population (persons)")
ax.ticklabel_format(axis="y", style="plain")
ax.tick_params(axis="x", rotation=45)
fig.tight_layout()
plt.show()
plt.close(fig)

実行コマンド出力結果

chart_rows: 10

出典：総務省統計局「令和2年国勢調査人口等基本集計」（e-Stat）を加工して作成。

横軸の地域コードと市区町村名は、左から次の順番で対応しています。

地域コード	市区町村名
`13112`	世田谷区
`13120`	練馬区
`13111`	大田区
`13123`	江戸川区
`13121`	足立区
`13115`	杉並区
`13119`	板橋区
`13201`	八王子市
`13108`	江東区
`13122`	葛飾区

グラフセルは可視化用です。ライブラリ導入前にデータ処理だけを確認する場合は、このセルを追加せずに進め、後から追加して再実行できます。

● 分析用CSVを出力する

ランキング表示に使った分析用データの全62行を、他のツールへ受け渡せるCSVとして保存します。保存先は実行IDごとのexports/tokyo_comparison_csvです。coalesce(1)は今回の小規模な出力を1つのpartファイルにまとめるために使います。大規模データで常に1ファイルへまとめる設計にはしません。

実行コマンド

csv_output = f"{RUN_ROOT}/exports/tokyo_comparison_csv"
tokyo.orderBy("area_code").coalesce(1).write.mode("errorifexists") \
    .option("header", True).option("encoding", "UTF-8").csv(csv_output)
print("CSV directory:", csv_output)
print("rows:", tokyo.count())
print("nb03_analyze_census: completed")

実行コマンド出力結果

CSV directory: /Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02/exports/tokyo_comparison_csv

rows: 62
nb03_analyze_census: completed

Object Storage上では、この配下のpart-...csvが実データです。_SUCCESSはデータではありません。ダウンロード先でCSVを読む際も、area_codeは文字列として扱います。

CSVはBOMなしのUTF-8で出力されるため、日本語版Excelでダブルクリックして開くと地域名が文字化けすることがあります。Excelで開く場合は、［データ］→［テキストまたはCSVから］でファイルの元の形式にUTF-8を指定し、area_code列をテキストとして取り込みます。

● 分析用CSV出力確認

書き出したCSVを、保存前と同じスキーマを指定してSparkで読み直し、件数が62行であることと、選択した5列の先頭10行を確認します。さらに、実際に読み込んだファイル名と保存先を表示し、Volumeの画面とObject Storageでも同じファイルを確認します。この手順で、CSVの出力と再読込みを確かめます。

ここでのCSV read: PASSは、この確認セルを通過したことを示します。Parquet保存時に行ったexceptAllによる双方向比較とは異なり、CSVの保存前後の全行・全列の値比較は、本稿では実施していません。

1) Notebookで保存したCSVを読み直す
nb03_analyze_censusのCSV出力セルの後へ追加します。RUN_ROOTを使うため、手動実行でもWorkflowでも、その実行で生成したCSVを読み直せます。

実行コマンド

csv_dir = f"{RUN_ROOT}/exports/tokyo_comparison_csv"

# 保存前と同じ型を指定して読み込む
csv_check = (
    spark.read
    .schema(tokyo.schema)
    .option("header", True)
    .option("encoding", "UTF-8")
    .option("mode", "FAILFAST")
    .csv(f"{csv_dir}/part-*.csv")
)

n = csv_check.count()
print("CSV rows:", n)
if n != EXPECTED_ROWS:
    raise ValueError(f"CSV件数不一致: {n}")

csv_check.select(
    "area_code", "area_name", "population_2020",
    "population_change_pct", "single_households_pct"
).orderBy("area_code").show(10, truncate=False)

print("CSV read: PASS")
print("CSV directory:", csv_dir)
print("files:", csv_check.inputFiles())

実行コマンド出力結果（CSV再読込み）

CSV rows: 62
+---------+---------+---------------+---------------------+---------------------+
|area_code|area_name|population_2020|population_change_pct|single_households_pct|
+---------+---------+---------------+---------------------+---------------------+
|13101    |千代田区 |66680          |14.16635277197548    |57.01918134350567    |
|13102    |中央区   |169179         |19.82958288179172    |52.59766237417151    |
|13103    |港区     |260486         |7.07118869793615     |56.88439527942827    |
|13104    |新宿区   |349385         |4.74427389375225     |67.80109772049931    |
|13105    |文京区   |240069         |9.25934354007755     |57.88685573957054    |
|13106    |台東区   |211444         |6.75054146703488     |59.04884128433574    |
|13107    |墨田区   |272085         |6.16956850870553     |52.16710505531938    |
|13108    |江東区   |524310         |5.26009367427611     |46.75609876150558    |
|13109    |品川区   |422488         |9.2109446691913      |57.28436240508408    |
|13110    |目黒区   |288088         |3.76987414542075     |53.19516740569372    |
+---------+---------+---------------+---------------------+---------------------+
only showing top 10 rows

CSV read: PASS
CSV directory: /Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02/exports/tokyo_comparison_csv
files: ['file:/Volumes/blog_catalog/opendata/census_work/runs/manual_20261011_02/exports/tokyo_comparison_csv/part-00000-b1b4a5f0-c065-47a1-959a-1c2022599dc7-c000.csv']

同じ手動実行IDの出力先からCSVを読み直し、62行・CSV read: PASSと実ファイル名を確認できました。Workflowの実行結果は後述します。

2) CSVファイルの場所を確認する
今回のCSVは、次の順番で開いた場所にあります。
blog_catalog → opendata → Volumes → census_work
　→ runs → manual_20261011_02
　→ exports → tokyo_comparison_csv

画面の実行IDは撮影時のものです。

3) Object Storage CSVファイルの場所確認
Object StorageのBucket-AIDPでも、tokyo-census/work/runs/manual_20261011_02/exports/tokyo_comparison_csv/配下に同じCSVファイルがあることを確認しました。_SUCCESSとCSVが表示され、CSVのサイズは11.72 KiBでした。

■ Workflowで3つのNotebookを順番に実行する

ここからは、3つのNotebookを1つずつ手動操作していた処理を、1回の実行で最後まで進むWorkflowにまとめます。ingestで原本を取り込み、buildで結合・計算・検査、analyzeで分析・CSV出力を行う順番を設定します。Notebook間では保存したParquetと検査記録を受け渡し、前の処理が成功してから次の処理へ進みます。

Workflowでは、{{job.run_id}}が展開されたUUIDを出力先に使います。手動実行とは別の実行IDになり、3つのTaskが同じ出力先を共有します。

Notebookで動作確認できた処理を、Jobにまとめます。ここではJob＝一連の処理、Task＝各Notebookの実行です。

● Jobを作成

3つのNotebookをまとめて実行するために、tokyo_census_pipelineというJobを作ります。Jobは処理全体の実行単位で、この段階では名前・説明・同時実行数を設定します。どのNotebookをどの順番で動かすかは、次の節でTaskとして追加します。

AIDP_OpenData_PoC → Workflow → Jobs → Create Jobで、次のように設定します。

同時実行数は、初回の動作を追いやすいように1とします。公式：Configure Jobs

1) Workflow画面
AIDP_OpenData_PoC → Workflow → Jobsタブで [Create Job]をクリック

2) Create Job画面
次の項目を設定し、[Create]をクリック

項目	設定例
Name	`tokyo_census_pipeline`
Description	Description一覧の`tokyo_census_pipeline`の値
Max Concurrent Runs	`1`

今回のCreate Job画面にはLocation項目が表示されませんでした。選択中のAIDP_OpenData_PoCで作成し、各TaskのNotebook選択時に対象のパスを確認します。

3) Create Job完了

● 3つのTaskを追加

Jobの中に、Notebook 01・02・03を実行するTaskを1つずつ登録します。ingest → build → analyzeの依存関係を設定すると、前のTaskが成功した後に次のTaskを開始するため、中間データの保存や検査が終わる前に後続が読み込むことを防げます。各Taskで、実行するNotebookとComputeを指定します。

JobのAdd taskから、Task typeをNotebook、SourceをWorkspaceとして、同じWorkspace内の対象NotebookとCensus_PoC_Clusterを選びます。今回の設定はTimeout 30分、Retries 0、Retry on timeoutオフ、Streaming taskオフです。

Task name	Notebook	Depends on	Run if
`ingest`	`/Workspace/nb01_ingest_census.ipynb`	なし	設定不要（最初のTask）
`build`	`/Workspace/nb02_transform_census.ipynb`	`ingest`	`All dependencies have executed and succeeded`
`analyze`	`/Workspace/nb03_analyze_census.ipynb`	`build`	`All dependencies have executed and succeeded`

依存関係のあるTaskは、上表の「全依存タスクが成功した場合」を選びます。掲載したbuildの画面ではAt least one dependency has succeededを使用していますが、今回は依存先がingestの1つだけなので、どちらもingest成功時に実行されます。ingestは依存先がないためRun ifの設定は不要です。Taskの変更は自動保存されます。公式：Configure Tasks

● Task parameterを設定

各NotebookのWidgetへ、Workflow実行時の入力値を渡します。入力Volume、出力Volume、実行ID、期待件数を3つのTaskで揃えることで、同じ原本と同じ保存先を使って一連の処理を実行できます。run_idにはJob実行ごとのIDを渡し、実行するたびに出力先のディレクトリを分けます。

3つのTaskすべてに、次のキー・値を設定します。Jobレベルに同名パラメータを重ねず、Taskの設定へ揃えます。

Key	Value
`raw_volume`	`/Volumes/blog_catalog/opendata/census_raw`
`work_volume`	`/Volumes/blog_catalog/opendata/census_work`
`run_id`	`{{job.run_id}}`
`expected_rows`	`62`

blog_catalog.opendata.census_rawのようなドット区切りの名前ではなく、上表の/Volumes/...をValueに入力します。

{{job.run_id}}は手入力する固定IDではなく、AIDPが実行時に展開するシステムパラメータです。同じJob実行内では3つのTaskが同じ出力ディレクトリを共有し、次回のRun Nowや定期実行では別のIDになります。{{task.run_id}}ではTaskごとに値が変わるため、ここでは使いません。公式：Parameterization

・ ingest Task

・ build Task

・ analyze Task

● Run Nowで実行

設定したJobを手動で1回実行し、3つのTaskが指定した順番で成功するかを確認します。Taskのステータスに加え、原本抽出の62行、検査のPASS、CSVの62行と再読込み結果をそれぞれのログで確認します。これにより、個別に動いたNotebookが、保存済みデータを受け渡して連続実行できることを確かめます。

Run Nowをクリックし、Job Runsで対象の実行を開きます。各Taskのノードからログを開き、共通セルが出力したrun_rootが一致していることを確認します。

確認箇所	成功時に確認する内容
ingest	3種類の中間データが各62行
build	keysとvalidationがPASS、合計5項目一致、保存後の再読込み一致
analyze	ランキング、グラフ、CSV出力
実行履歴	3つが依存関係どおりに成功

画面の実行名と、Taskへ渡されたrun_idパラメータは別の表示です。出力パスにはUUID形式のrun_idが使われています。各Taskの詳細に表示されるTask run keyもTaskごとの別のIDなので、出力先にはbuildの検査記録やanalyzeのログにある共通のrun_idを使います。

・ Ingest Success

人口・世帯・年齢の3表それぞれでrows: 62、unique_codes: 62を確認できました。

・ Build Success

検査結果はstatus: PASS、rows: 62です。合計5項目と、年齢不詳をNULLにした6地域も確認できました。

・ Analyze Success

Workflow内のCSV確認セルでCSV rows: 62、CSV read: PASSを確認できました。

● Workflow実行で生成したCSVを確認する

Workflowが出力したCSVを、Notebook 03の確認セルで保存前と同じ型を指定して読み直します。analyzeのログから、CSVが62行であること、実ファイル名と保存先の実行IDを確認し、buildの検査記録と同じ実行の成果物であることを照合します。手動実行時のCSVと区別し、このJobが作成したファイルを確認する手順です。

Workflow内のCSV再読込み結果（analyze出力の抜粋）

CSV rows: 62
CSV read: PASS
CSV directory: /Volumes/blog_catalog/opendata/census_work/runs/ab7cee16-f554-4e38-ac7b-ffb8de167101/exports/tokyo_comparison_csv
files: ['file:/Volumes/blog_catalog/opendata/census_work/runs/ab7cee16-f554-4e38-ac7b-ffb8de167101/exports/tokyo_comparison_csv/part-00000-43a6d48c-26dd-4d4a-8583-7f500667fa5e-c000.csv']

buildの検査記録と同じ実行IDの保存先から、1つのCSVファイルを読み込み、ヘッダーを除くデータが62行あることを確認できました。これで3タスクの成功、62市区町村の検査、CSVの保存・再読込みまで確認できました。

任意：別Notebookで特定のRunのCSVを確認する

特定の成功Runを後から確認する場合は、次の例を確認用の別Notebookで実行します。これは追加確認用のコード例で、上の実測ログはWorkflow内の確認セルの出力です。

実行コマンド（確認用。Workflowには含めない）

WORK = "/Volumes/blog_catalog/opendata/census_work"
check_run_id = "ab7cee16-f554-4e38-ac7b-ffb8de167101"  # 確認したいJob Runのrun_id

csv_dir = f"{WORK}/runs/{check_run_id}/exports/tokyo_comparison_csv"
csv_check = spark.read.option("header", True).csv(f"{csv_dir}/part-*.csv")
print("CSV directory:", csv_dir)
print("rows:", csv_check.count())
print("files:", csv_check.inputFiles())

今後の実行も検査する場合は、Notebook 03のRUN_ROOTを使ったCSV確認セルを使います。特定のUUIDを固定した一度限りの確認セルは、Workflowで実行するNotebookへ残さないようにします。

● 今後の検証：異常系テストと定期実行

ここまで確認した正常系に加えて、検査が失敗した場合に後続Taskが止まるかと、指定時刻にJobを実行できるかを確認するための手順です。異常系テストでは期待件数を変えて検査を失敗させ、定期実行ではScheduleを設定して実行履歴を確認します。どちらも本稿では未実施のため、以下は今後行う検証の要点です。

異常系テスト：検査が失敗したら後続が止まるか

build Taskだけexpected_rowsを61に変えてRun Nowします。入力データは壊さず、検査側の期待件数だけを変えます。
buildが件数不一致で失敗し、analyzeが実行されないことをJob Runsで確認します。後続の表示名（Skipped、Upstream failed等）は環境により異なります。
expected_rowsを62へ戻し、新しいRun Nowで全体を再実行します。本稿は既存出力を上書きしない設計なので、既存RunのRepairや自動retryは、同じrun_idの途中出力の扱いを決めてから追加します。

Scheduleによる定期実行

JobのDetailsでScheduleを追加します。最初はPausedで作成し、Time ZoneをAsia/Tokyo、Schedule TypeをCalendarにして、次回の予定時刻を確認してからActivateします。1回の動作を確認したらPauseします。
Run Asの実行ユーザーと権限を確認します。保存済み資格情報がなく組織ログインに依存している場合、ログイン期限切れで定期実行が失敗する可能性があります。継続運用ではCredential Storeや適切なサービスアカウントを含めて設定します。公式：Workflow Jobs / Run As
公式のKnown Issuesには、定期ジョブからObject Storageへ直接アクセスする制約が記載されています。本稿は公式の回避策であるExternal Volume経由を使います。公式：Known Issues

今回の国勢調査は固定した過去データなので、定期実行は自動実行の仕組みを確認することが目的です。

■ つまずいたときの確認箇所

作成や実行が想定どおりに進まない場合に、どこを確認するかを症状別にまとめます。エラーが出たNotebook・セル・保存先を確認し、該当する行からVolumeのパス、権限、Compute、ライブラリ、Widgetの値を切り分けます。

症状	確認すること
Notebook名でInvalid resource name	英字で始まる`nb01_ingest_census`等を指定する。先頭の数字や許可されない記号を確認する
Notebook保存で404 BucketNotFound	エラーに記載されたWorkspace用Bucketの存在、Region・Namespace・Compartment、AIDP自身のIAM権限を確認する
Create Volumeを選択できない	標準Catalogの対象Schemaと、Schemaの`CREATE_VOLUME`または`ADMIN`権限を確認する。OCI IAMの設定とは別にAIDP内権限が必要
WorkspaceでComputeが選べない	同じWorkspace内にClusterを用意したか、Notebookの接続先と利用権限を確認する
`/Volumes/...`が見つからない	実際のCatalog・Schema・Volume名、Copy path、パスの大文字小文字、ファイルの位置
403／権限エラー	Object StorageのIAMとAIDP内権限、WorkflowのRun As。手動とJobの実行者の差
pandasの行抽出で`RecursionError`	今回はNumPy 1.26.4・pandas 2.2.3等の構成へ入れ替えてClusterを再起動し解消。先に2行のDataFrameで動作と読み込み場所を確認する
グラフ作成でエラー	matplotlibの導入と再起動後の一時ビュー作成を確認。本文の成功コードは上位10件を`collect()`してPythonリストで描画する
Create Notebookで設定画面が出ない	今回は`Untitled_...`が自動作成された。編集画面上部の名前をクリックして変更し、Descriptionも後から編集する
Create JobにLocationがない	今回の画面には項目なし。対象Workspaceと各TaskのNotebookパスを確認する
Task parameterのVolumeパスで失敗	`blog_catalog.opendata.census_raw`ではなく`/Volumes/blog_catalog/opendata/census_raw`を指定する
openpyxl等をimportできない	対象ComputeのLibrary、インストール結果、再起動、Notebookに接続したCompute
元ファイルの表頭不一致	表番号、人口集中地区版等の取り違え、原本の訂正・レイアウト変更
NULL／型変換エラー	原表の記号と数値の区別。未知の値をゼロやNULLに黙って変換しない
62行にならない／JOINで増える	地域識別の条件、旧市町村、地域コードの一意性、未結合コード
path already exists	同じrun_idを再利用していないか。3つとも新しいrun_idへ変えて全体を再実行
SQLで一時ビューが見えない	同じNotebook・Spark Sessionでビューを作成したか。Notebook 03の読込みセルを実行
`{{job.run_id}}`が文字列のまま残る	Task parameterのキー・書式・対応機能。共通セルの出力を確認
グラフで日本語が文字化け	まず地域コードを軸にする本稿の例で確認。名称表示には日本語フォントを準備
手動実行は成功、定期実行は失敗	Volume経由か、Run Asの資格情報・権限・期限、Computeの起動状況

● Workspaceの保存先とデータ用Bucketを区別する

Notebook自体を保存する領域と、Excel・Parquet・CSVを置くデータ用Bucketは用途が異なります。BucketNotFoundが出たときは、まずどちらの保存処理で使われたBucketかを確認します。この区別により、確認するBucket名と権限を絞れます。

保存対象	この検証での扱い
Notebook等のWorkspaceファイル	AIDPが管理するWorkspace用の保存領域を使用する
原本Excel・加工済みParquet	利用者が用意した`Bucket-AIDP`の`tokyo-census/`配下をExternal Volume経由で利用する

検証途中では、当初使用していたWorkspaceでNotebook保存時にPutObjectの404 BucketNotFoundが発生し、エラーに記載されたBucketを一覧で確認できませんでした。削除の可能性はありますが、削除履歴の確認前なので原因は未確定です。その後、使用するWorkspaceを切り替えてNotebookの作成・名前変更・保存を確認し、セル実行も成功しました。

同じエラーが出た場合は、エラーに記載されたBucket名・Namespace・Regionを控え、Bucketの存在とAIDP自身のIAM権限を確認します。見つからない場合も参照権限やCompartmentの違いを切り分けます。既存Workspaceの復旧が必要な場合は、Workspace IDとエラー全文を添えてOracle Supportへ確認します。

● Widgetに以前の実行IDが残ったとき

共通セルのdefaultValueを編集しても、画面のWidgetには以前に入力した実行IDが残ることがあります。そのまま実行すると、意図した保存先ではなく古い実行のディレクトリを使うため、Widgetの値と共通セルが出力したrun_rootを確認します。以下はWidgetに旧値が残った例です。

Widgetに旧値が残った例

{
  "raw": "/Volumes/blog_catalog/opendata/census_raw",
  "run_root": "/Volumes/blog_catalog/opendata/census_work/runs/manual_20261009_01",
  "expected_rows": 62
}

右側の「実行ID」を実行に使う値へ変更し、共通セルから順に実行します。新しい手動実行IDを使う場合は、3つのNotebookで同じ値に揃え、Notebook 01の保存が完了してからNotebook 02を実行します。

■ やってみた結果

原本の読込みから、結合・指標計算・品質検査・分析・CSV出力までの確認結果をまとめます。個々のNotebookとWorkflowの両方について、62市区町村のデータが処理され、保存したファイルを読み直せたことを、実行ログと成果物で確認しています。

検証日は2026-10-12（JST）です。手動実行IDはmanual_20261011_02、Workflowの実行IDはab7cee16-f554-4e38-ac7b-ffb8de167101です。

Workflowの実行記録（開始・終了時刻と各Taskの所要時間）

Workflow実行記録

Job名：tokyo_census_pipeline
画面の実行名：tokyo_census_pipeline.job_1791740372746_44049043
run_idパラメータ：ab7cee16-f554-4e38-ac7b-ffb8de167101
開始：2026-10-12 02:39:32 JST
終了：2026-10-12 02:41:24 JST
全体：Success／1分51秒（画面表示）
ingest：Success／39秒／3表すべてrows=62、unique_codes=62
build：Success／53秒／status=PASS、rows=62、合計5項目一致、age_unknown_nulls=6
analyze：Success／16秒／CSV rows: 62、CSV read: PASS（Workflow内のログ）

検証項目	結果・確認したログ
Workspace	作成済みの`AIDP_OpenData_PoC`をNotebook・Compute・Workflowで使用
Notebookの名前変更・保存	`nb01_ingest_census`へ変更し、Last saved表示と保存したセルの実行を確認
Compute接続とライブラリ	同じClusterで4ライブラリをimportし、指定版との一致と読込み元を確認。版は導入後の環境記録表を参照
原本の読込み・抽出	External VolumeからExcel3件を読み込み、3表とも62行・地域コード62件
Sparkでの結合・品質検査	JOIN後62行、合計5項目一致、NULLの6地域の集合一致、既知年齢合計と総人口の一致、validation PASS
Parquet保存・再読込み	件数と`exceptAll`による双方向の値比較を通過
ランキングとグラフ	人口・人口増減率・単独世帯割合を表示。人口上位10件の棒グラフを確認
年齢別ランキング	檜原村53.03%、奥多摩町50.79%。年齢不詳割合とNULLも併記
Workflowの3タスク実行	ingest → build → analyzeがすべてSuccess。全体1分51秒（画面表示）
手動実行のCSV	CSV再読込み62行・PASS。Object Storage上の同名ファイルも確認
Workflow出力CSV	成功Runの保存先からCSV1ファイルを読み込み、62行・CSV read: PASSを確認。buildの検査記録とrun_idも一致
期待件数を61にする異常系テスト	未実施（「今後の検証」に手順の要点を記載）
Scheduleによる定期実行	未実施（「今後の検証」に手順の要点を記載）

手動実行とWorkflowの両方で、CSVの保存・再読込みまで確認できました。各Notebookは同じ実行IDのParquetと検査記録を通してデータを受け渡します。

● 分析から見えたこと

SQLのランキングから、人口規模、人口の増減、単独世帯割合、高齢者割合という異なる切り口で地域の特徴を読みます。割合を見る際は、元の人数・世帯数と分母、年齢不詳の扱いも合わせて確認します。以下は今回のデータから得られた代表的な結果です。

人口増減率：2015年から2020年の増加率の上位は、中央区（19.83%）・千代田区（14.17%）・文京区（9.26%）など都心の区が占めました。
単独世帯割合：新宿区（67.80%）・渋谷区・豊島区などの区部と、青ヶ島村・御蔵島村などの島しょ部が上位に並びました。島しょ部は一般世帯数が数百以下と小さいため、割合と世帯数を分けて読みます。
高齢者割合：檜原村（53.03%）・奥多摩町（50.79%）など西多摩地域と、八丈町・新島村などの島しょ部が上位でした。

いずれも2020年10月1日時点の国勢調査の値で、現在の人口や将来の動向を示すものではありません。

● AIDPを使ってみてみて

今回は、国勢調査のExcel原本から人口・世帯・年齢のデータを取り出し、東京都62市区町村を比較できる形へ整えました。地域コードや件数、東京都合計、公表割合を照合し、分析用ParquetとCSVを保存・再読込みするところまで確認できました。Workflowで3つのNotebookが順番に成功し、CSVの再読込みまで確認しています。

Oracle Databaseで表を設計するときも、主キーやデータ型、NULLの意味、集計の分母を意識します。今回も、ExcelやParquetを扱いながら、同じようにデータの意味を確認する工程が必要でした。SQLを書けることに加えて、SQLで正しく比較できる状態へ整えること。その工程をNotebookのコードと検査記録として残せたことが、今回のポイントです。

こうしたデータの準備は、次にAIで分析結果を説明するときにも関わってきます。「人口が増えた」と言うときにどの期間を比較したのか、「割合が高い」と言うときに何を分母にしたのか。AIへ渡すデータにも、その意味と出典を揃えておきたいところです。今回の検証を、そこへ進むための土台にしていきます。

● 続編の年度データ設計

続編では、令和7年（2025年）の国勢調査データを追加し、人口・単独世帯・高齢者割合がどのように変わったのかを比較します。年度ごとの表の構造や地域の境域、年齢不詳率の違いも確認しながら、同じ条件で比較できるデータを用意する予定です。

年度ごとの中間データは、year列と年なしの列名で縦持ちにし、そこから年度比較用の横持ちテーブルを作る方針です。

データ	列の例	用途
年度別の中間データ（縦持ち）	`year`、`area_code`、`population`	2020年と2025年を同じ処理で扱い、年度ごとに行を保存する
年度比較用テーブル（横持ち）	`area_code`、`population_2020`、`population_2025`	地域ごとの増減を計算する。年度は列名で表し、`year`列は持たない

表頭や地域の境域、公表割合の分母を年度ごとに確認してから比較します。この設計は続編で実装する予定です。
そのうえで、SQLで得た数値をもとに、AIによる地域分類や結果の解説を試してみます。元のデータと計算結果を追える状態を保ちながら、AIがどのような説明を返すのか。次回は、このデータを使ってAI分析までつなげてみてみたいと思います。

■ 参考

原本データの取得元と、この記事で使用したAIDPの操作・権限・ファイル形式を確認するための資料です。別の環境や年度へ展開する際は、該当する公式資料と原本の表頭を確認します。

■ 解説

■ おまけ

今回の内容を、ずんだもん達に紹介してもらう漫画も作ってみました。技術記事の補足として、少しでも楽しく読んでもらえたらうれしいです。

※ 本漫画は筆者による非公式の二次創作です。
※ 使用キャラクター：ずんだもん / 四国めたん / 春日部つむぎ
※ キャラクターの権利は各権利元に帰属します。
※ クレジット
　- ずんだもん / 四国めたん：東北ずん子・ずんだもんプロジェクト関連ガイドラインに基づいて利用
　- 春日部つむぎ：公式利用規約に基づいて利用

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

AI Data Platform(AIDP) にオープンデータを取り込んで東京都62市区町村を分析してみてみた

■ はじめに

■ 今回やってみること

■ 使用するオープンデータ

■ 事前準備

● AIDPの既存環境を確認

■ CatalogとSchemaを用意する

■ 原本を用意する

■ Volumeを作成する

● 入力・出力用Volumeの登録

・ census_raw Volume登録

・ census_work Volume登録

■ Compute・Notebook・ライブラリを準備する

● 使用するWorkspaceにComputeを作成する

● Notebookを3つ揃える

・ nb01_ingest_censusをCensus_PoC_Clusterへアタッチ

・ nb02_transform_censusをCensus_PoC_Clusterへアタッチ

・ nb03_analyze_censusをCensus_PoC_Clusterへアタッチ

● ライブラリを確認

・ NumPyとpandasの組み合わせを確認する

・ 導入後に4つのライブラリの実際の版を記録する

● Descriptionをまとめて設定する

● 3つのNotebookに共通のパラメータセルを置く

■ Notebook 01：原本を読み込み、3つの中間データを作る

● 原本の存在・サイズ・ハッシュを確認

● Excelの表頭を見てみる

● 原本を読む関数を定義

● 3つのデータを抽出する

● Spark DataFrameへ変換し、Parquetで保存

■ Notebook 02：Sparkで結合・計算・品質検査する

● 中間Parquetを読み込む

● 地域コードでJOINし、元表同士を照合

● Spark SQLで指標を計算

● 件数・合計・欠損・公表値を検査

● 分析用Parquetと検査記録を保存

■ Notebook 03：Spark SQLで地域の特徴を分析する

● 検査結果を確認し、一時ビューを作る

● 人口の多い市区町村を調べる

● 2015年から2020年に人口が増えた地域を調べる

● 単独世帯の割合と世帯数を比較する

● 年齢不詳の影響も確認する

● グラフにする

● 分析用CSVを出力する

● 分析用CSV出力確認

■ Workflowで3つのNotebookを順番に実行する

● Jobを作成

● 3つのTaskを追加

● Task parameterを設定

● Run Nowで実行

● Workflow実行で生成したCSVを確認する

● 今後の検証：異常系テストと定期実行

■ つまずいたときの確認箇所

● Workspaceの保存先とデータ用Bucketを区別する

● Widgetに以前の実行IDが残ったとき

■ やってみた結果

● 分析から見えたこと

● AIDPを使ってみてみて

● 続編の年度データ設計

■ 参考

■ 解説

■ おまけ

・導入後に4つのライブラリの実際の版を記録する