コンテンツにスキップ
Fusion ERP 実装ラボ

BICC 演習：増分抽出を設計・設定する

この演習で身につくこと

Section titled “この演習で身につくこと”
  • 要件から抽出対象（データストア）を選び、粒度・主キー・増分に使う日付列を整理できる
  • 初回全件と増分の方針（Initial Extract Date、Prune Time、スケジュール、全件に戻す条件）を決められる
  • 主キー抽出による削除検知の設計を説明できる
  • マニフェスト起点の取込処理（完全性の検証・冪等・実行順・リトライ）を設計し、動かして検証できる
  • 実機では、BICC コンソールで外部ストレージ・ジョブ・スケジュールを設定し、出力ファイルを確認できる

用語と仕組みは BICC によるデータ抽出 を参照してください。

前提・準備（Fusion 環境あり／なし）

Section titled “前提・準備（Fusion 環境あり／なし）”

シナリオ（架空）

Section titled “シナリオ（架空）”

サンプル商事株式会社の経理 IT 部門は、AP（買掛管理）のデータを OCI のデータレイクへ毎日連携し、仕入先別の支払状況を分析したいと考えています。要件は次のとおりです。

No 要件
R1 対象は、AP 請求書のヘッダーと配分、仕入先、GL 仕訳明細。分析に不要な列（銀行口座など）は取り出さない
R2 初回は 2024 年 4 月以降の全件を週末に取り込み、その後は毎日、業務時間外に増分を取り込む
R3 Fusion で削除された請求書は、翌日の朝までにデータレイク側にも反映する（履歴は残してよい）
R4 抽出が失敗したら、IT 運用チームへメールで知らせる
R5 取込は、ファイルの欠けや破損を検出でき、再実行しても二重に取り込まれない
R6 抽出ファイルは OCI Object Storage に置き、30 日で削除する
R7 抽出に使うユーザーは人と分け、必要最小限の権限にする

データ量の見込み（架空）は次のとおりです。ジョブの分け方や主キー抽出の頻度を決める材料にします。

対象 全件の見込み 1 日の増分の見込み
請求書ヘッダー 約 50 万行 約 2,000 行
請求書配分 約 150 万行 約 6,000 行
仕入先 約 3 万行 数十行
GL 仕訳明細 約 2,000 万行 約 8 万行

Fusion 環境ありルート

Section titled “Fusion 環境ありルート”
  • BICC コンソールを使えるロール（BIACM_ADMIN または BI Administrator）と、テスト環境。本番では行いません
  • OCI の Object Storage のバケットと、そのバケットに書き込める OCI ユーザー（API キーを登録できること）。結果の確認に OCI CLI を使うと便利です（任意）
  • 請求書が数件以上あるテスト環境。データが多い環境では、初回の全件抽出が長くなるため、Initial Extract Date を直近の日付にします

環境なしルート（机上演習とサンプルファイル）

Section titled “環境なしルート（机上演習とサンプルファイル）”
  • ステップ 1〜4 を、設計シートとして表に書きます（机上演習）。実機の画面に出てくる名前は、後の解答例で確認します
  • ステップ 7 の取込処理は、Python 3（標準ライブラリのみ）で動くサンプルを使います。BICC の出力を模した疑似ファイルを作って取り込みます

手順

Section titled “手順”

ステップ 1 抽出対象を選ぶ（設計シート 1）

Section titled “ステップ 1 抽出対象を選ぶ（設計シート 1）”

課題：要件 R1 から、抽出するデータを「請求書ヘッダー」「請求書配分」「仕入先」「GL 仕訳明細」の 4 つに整理し、次の項目を表（設計シート 1）にまとめてください。

  • データストアを探す検索語（英語）
  • 1 行が表すもの（粒度）と主キー
  • 増分の判定に使う日付列と、Initial Extract Date に使う作成日列
  • 取り出さない列
  • ジョブを分けるか（データ量の見込みから判断）

ヒント

  • データストア（PVO）の名前は推測で書かず、実機の Manage Offerings and Data Stores で検索して確定します。机上では「（実機で確認）」とし、検索語と選ぶ基準を書きます
  • 業務テーブルの主キーは 主要テーブルマップ で調べられます（例：請求書は INVOICE_ID）。PVO の列名は物理テーブルの列名と一致するとは限らないため、Select Columns で実際の列を確認します
  • 抽出する列は、Select Columns で必要なものだけにします。受け側で結合に使う ID 列は落とさないようにします
  • 増分の判定に使う日付列が定義されているかは、データストアの列の設定（Incremental Key Columns など）で確認します

ステップ 2 抽出方針を決める（設計シート 2）

Section titled “ステップ 2 抽出方針を決める（設計シート 2）”

課題：要件 R2 と R4 を満たす方針を、次の項目で表（設計シート 2）にまとめ、理由を 1 行ずつ添えてください。

  • Initial Extract Date
  • 初回の全件抽出の実行時期と、ジョブの分け方
  • 増分のスケジュール
  • Prune Time
  • Reset to Full Extract を使う条件
  • 通知

ヒント

  • Prune Time の既定は 1,440 分（24 時間）です。ガイドは、抽出頻度に合わせる（毎時なら 60、毎日なら 1,440、0 は重なりなし）としています。重なった分が重複して出力されることを前提にします
  • ジョブのタイムアウトは既定で 10 時間で、超えると失敗します。データ量の見込みと比べて、初回が収まるかを考えます
  • 時間単位の抽出は、Fusion のデータベースへの負荷から、ガイドでは推奨されていません
  • 繰り返しの選択肢は、ガイド（25C）では Simple（1 回）・Hourly・Weekly・Monthly・Yearly です。「毎日」の作り方は実機の画面で確認します

ステップ 3 削除の検知を設計する（設計シート 3）

Section titled “ステップ 3 削除の検知を設計する（設計シート 3）”

課題：要件 R3 を満たす方式を、次の項目で表（設計シート 3）にまとめてください。

  • ジョブ種別と、対象のデータストア
  • 実行の頻度と時刻
  • 受け側の処理（削除の持ち方、行が再び現れたときの扱い）
  • 取り込む順序

ヒント

  • 増分抽出は、変更・追加された行だけを返し、削除された行は返しません
  • 主キー抽出（ジョブ種別 Active Primary Key Extract）は、いま存在する主キーのすべてを .pecsv として出力します。全件分の主キーを読むため、対象と頻度は絞ります
  • 主キーの一覧にない行を、物理削除するか、削除フラグを立てるか（履歴・監査との兼ね合い）を決めます
  • 実行順に取り込まないと、削除と再出現の判定がずれます

ステップ 4 出力と自動化を設計する（設計シート 4）

Section titled “ステップ 4 出力と自動化を設計する（設計シート 4）”

課題：要件 R4〜R7 を満たす運用を、次の項目で表（設計シート 4）にまとめてください。

  • 格納先と保持期間
  • 取込の起動タイミングと、処理するファイルの探し方
  • 完全性の検証（欠け・破損）
  • 冪等性と実行順
  • 失敗時の動きと、再実行の方法
  • 監視する項目
  • 権限（誰が書き、誰が読むか）

ヒント

  • 出力先は UCM か OCI Object Storage です。UCM のファイルは、既定で 90 日で期限切れになります
  • マニフェストには、出力したファイル名と MD5 が書かれます。ジョブの抽出では MANIFEST_DATA_<JOB_ID>-SCHEDULE_<SCHEDULE ID>_REQUEST_<ESS_REQUEST_ID>.MF（主キー抽出は MANIFEST_PRIMARY_KEYS_…）という名前です
  • ステータス JSON には、状態（status）・エラー内容（errorMessage）・行数（rowCount）・所要時間が出ます
  • BICC の抽出が失敗したとき、そのデータストアの Last Extract Date は進みません。次の増分で何が起こるかを考えます

ステップ 5 BICC を設定する（環境ありルート）

Section titled “ステップ 5 BICC を設定する（環境ありルート）”
  1. ロール：実行ユーザーに BIACM_ADMIN（または BI Administrator）を付けます。UCM のファイルも扱うなら BICC_UCM_CONTENT_ADMIN も付けます。ロールの作り方は セキュリティ演習 と同じ考え方です
  2. コンソール：https://<your-pod>.fa.<region>.oraclecloud.com/biacm にサインインします
  3. 外部ストレージ：Configure External Storage の OCI Object Storage Connection タブで、Name・Host・Tenancy OCID・User OCID・Namespace・Bucket を入力します。Generate API Signing Key で鍵を作り（フィンガープリントが表示されます）、Export Public Key で公開鍵を保存して、OCI のユーザーの API キーとして登録します。最後に Test Connection で接続を確認します
  4. データストアと列：Manage Offerings and Data Stores で、設計シート 1 のデータストアを検索します。Select Columns で列を絞り、Export Metadata Definition で列定義を保存して設計書に添付します。必要なら Actions > Create Offering で、この演習用のオファリングにまとめます
  5. Initial Extract Date：Advanced Extract Configuration で、作成日列（Is Creation Date）の定義を確認し、Initial Extract Date を設定します。ジョブ側では Actions > Manage Initial Extract Date でも設定できます
  6. ジョブ：Manage Jobs で、設計シート 2 のとおりにジョブを作り、データストアを追加します。必要なデータストアだけ、Customize Job Data Stores で Query Filter や Support Chunking を設定します
  7. スケジュール：Manage Job Schedules で、ジョブ種別 Application Data Extract のスケジュールを作ります。最初は繰り返しを Simple（1 回）にして動作確認し、通知は Define Notification で失敗時のメールアドレスを指定します。確認できたら、毎日の増分と、設計シート 3 の Active Primary Key Extract のスケジュールを作ります

ステップ 6 抽出結果を確認する（環境ありルート）

Section titled “ステップ 6 抽出結果を確認する（環境ありルート）”
  1. 出力ファイル：バケットに、file_<データストア名>-batch<番号>-<日時>.zip、マニフェスト、ステータス JSON（EXTRACT_STATUS_…JSON）が出ていることを確認します。OCI コンソールか、OCI CLI の oci os object list を使います
  2. 完全性：マニフェストに書かれたファイルがすべて存在し、MD5 が一致することを確認します（手順とコードは BICC によるデータ抽出 の「コード例」）
  3. 件数：ステータス JSON の status が成功を示し、rowCount が Fusion の画面や BIP で数えた件数と合うことを確認します。所要時間（totalDurationInSec など）も記録します
  4. 増分：請求書を 1 件更新してから、もう一度抽出を実行します。更新した行が出力されること、前回と重なる行も含まれること（Prune Time の重なり）を確認します
  5. 削除：テスト環境で削除できる請求書を 1 件削除し、主キー抽出を実行します。出力された .pecsv にその主キーが含まれないことを確認します

ステップ 7 取込処理を動かして検証する

Section titled “ステップ 7 取込処理を動かして検証する”

取込処理を、マニフェスト起点で作ります。ここでは疑似ファイルで動かし、設計シート 4 の判断（完全性の検証・冪等・実行順・削除検知）がコードのどこに現れるかを確認します。

  1. 下の 2 つのスクリプトを、作業用フォルダに保存します
  2. 疑似ファイルを作る make_sample_extract.py と、取り込む ingest_extract.py を、下の順に実行します（Python 3.13 で動作確認）。やり直すときは、extract フォルダと dwh.db を削除します
make_sample_extract.py（疑似抽出ファイルの作成）を表示
make_sample_extract.py
"""学習用の疑似抽出ファイルを作る（実際の BICC 出力の完全な再現ではありません）。
使い方: python make_sample_extract.py 1   # 1 日目（全件）
        python make_sample_extract.py 2   # 2 日目（増分 + 主キー抽出）
"""
import csv
import hashlib
import io
import sys
import zipfile
from pathlib import Path


OUT = Path("extract")
OUT.mkdir(exist_ok=True)
STORE = "file_sampleam_invoiceheaderpvo"   # 架空のデータストア名（ピリオドをアンダースコアに置換した形）
created = []                               # 今回作ったファイル名


# 列：InvoiceId, InvoiceNum, InvoiceAmount, LastUpdateDate（設計シートの列定義）
DAY1 = [                                            # 1 日目：2026-09-21 02:00 の全件抽出
    (10001, "INV-2026-0001", 165000, "2026-09-10T10:00:00"),
    (10002, "INV-2026-0002", 44000, "2026-09-12T10:00:00"),
    (10003, "INV-2026-0003", 33000, "2026-09-15T10:00:00"),
    (10004, "INV-2026-0004", 11000, "2026-09-20T10:00:00"),
    (10005, "INV-2026-0005", 22000, "2026-09-18T10:00:00"),
]
DAY2_CHANGED = [                                    # 2 日目：2026-09-22 02:00 の増分抽出
    (10004, "INV-2026-0004", 11000, "2026-09-20T10:00:00"),   # 重なり（Prune Time 24 時間）で再抽出。内容は同じ
    (10003, "INV-2026-0003", 35000, "2026-09-21T09:00:00"),   # 金額が変更された
    (10006, "INV-2026-0006", 18000, "2026-09-21T15:00:00"),   # 新規
]
DAY2_ACTIVE_KEYS = [10001, 10002, 10003, 10004, 10006]        # 10005 は Fusion 側で削除された




def write_zip(name, member, rows, stamp):
    buf = io.StringIO(newline="")
    csv.writer(buf, quoting=csv.QUOTE_ALL).writerows(rows)   # すべての値を二重引用符で囲む
    info = zipfile.ZipInfo(member, date_time=stamp)
    info.compress_type = zipfile.ZIP_DEFLATED
    path = OUT / name
    with zipfile.ZipFile(path, "w") as z:
        z.writestr(info, buf.getvalue())
    created.append(name)
    return path




def write_manifest(name, files):
    lines = ["SOURCE_VERSION;sample"]                       # 先頭行はバージョン情報（内容は例）
    for f in files:
        lines.append(f"{f.name};{hashlib.md5(f.read_bytes()).hexdigest()}")
    (OUT / name).write_text("\n".join(lines) + "\n", encoding="utf-8")
    created.append(name)




day = sys.argv[1] if len(sys.argv) > 1 else "1"
if day == "1":
    z = write_zip(f"{STORE}-batch1001-20260921_020000.zip",
                  f"{STORE}-batch1001-20260921_020000.csv", DAY1, (2026, 9, 21, 2, 0, 0))
    write_manifest("MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9001.MF", [z])
else:
    z = write_zip(f"{STORE}-batch1002-20260922_020000.zip",
                  f"{STORE}-batch1002-20260922_020000.csv", DAY2_CHANGED, (2026, 9, 22, 2, 0, 0))
    p = write_zip(f"{STORE}-batch1003-20260922_030000.zip",
                  f"{STORE}-batch1003-20260922_030000.pecsv", [(k,) for k in DAY2_ACTIVE_KEYS],
                  (2026, 9, 22, 3, 0, 0))
    write_manifest("MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9002.MF", [z])
    write_manifest("MANIFEST_PRIMARY_KEYS_5001-SCHEDULE_7002_REQUEST_9003.MF", [p])
print("created:", *created, sep="\n  ")
ingest_extract.py（マニフェスト起点の取込）を表示
ingest_extract.py
"""マニフェストを起点に、疑似抽出ファイルを SQLite の DWH へ取り込む（冪等・削除検知つき）。"""
import csv
import hashlib
import io
import re
import sqlite3
import zipfile
from pathlib import Path


EXTRACT_DIR = Path("extract")     # ストレージからダウンロードしたファイルの置き場
COLUMNS = ["invoice_id", "invoice_num", "invoice_amount", "last_update_date"]   # 設計シートの列定義
HAS_HEADER = False                # CSV のヘッダー行の有無は実ファイルで確認して合わせる


db = sqlite3.connect("dwh.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS ap_invoice_header (
  invoice_id INTEGER PRIMARY KEY, invoice_num TEXT, invoice_amount INTEGER,
  last_update_date TEXT, is_deleted INTEGER DEFAULT 0);
CREATE TABLE IF NOT EXISTS processed_manifest (name TEXT PRIMARY KEY);
""")




def md5(path):
    return hashlib.md5(path.read_bytes()).hexdigest()




def request_id(path):
    """実行順は ESS 要求 ID の昇順。文字列ではなく数値で並べる。"""
    return int(re.search(r"REQUEST_(\d+)", path.name).group(1))




def read_manifest(path):
    lines = path.read_text(encoding="utf-8").splitlines()[1:]   # 先頭行はバージョン情報なので読み飛ばす
    return [line.split(";") for line in lines if line.strip()]




def rows_of(zf, member):
    with zf.open(member) as f:
        reader = csv.reader(io.TextIOWrapper(f, encoding="utf-8", newline=""))
        if HAS_HEADER:
            next(reader, None)
        yield from reader




def upsert(rows):
    """主キーで INSERT または UPDATE する。重複して届いた行（Prune Time の重なり）も同じ結果になる。"""
    db.executemany(
        f"INSERT INTO ap_invoice_header ({','.join(COLUMNS)}, is_deleted) VALUES (?,?,?,?,0) "
        "ON CONFLICT(invoice_id) DO UPDATE SET invoice_num=excluded.invoice_num, "
        "invoice_amount=excluded.invoice_amount, last_update_date=excluded.last_update_date, is_deleted=0",
        rows)




def mark_deleted(active_keys):
    """主キー抽出の一覧にない行を論理削除する（物理削除はしない）。"""
    db.execute("CREATE TEMP TABLE IF NOT EXISTS active_keys (invoice_id INTEGER PRIMARY KEY)")
    db.execute("DELETE FROM active_keys")
    db.executemany("INSERT OR IGNORE INTO active_keys VALUES (?)", [(int(k),) for k in active_keys])
    cur = db.execute("UPDATE ap_invoice_header SET is_deleted = 1 "
                     "WHERE is_deleted = 0 AND invoice_id NOT IN (SELECT invoice_id FROM active_keys)")
    print(f"  deleted flag set: {cur.rowcount} rows")




for mf in sorted(EXTRACT_DIR.glob("MANIFEST_*.MF"), key=request_id):
    if db.execute("SELECT 1 FROM processed_manifest WHERE name = ?", (mf.name,)).fetchone():
        print("skip (processed):", mf.name)
        continue
    entries = read_manifest(mf)
    for name, expected in entries:        # 1) 全ファイルの存在と MD5 を確認してから取り込む
        path = EXTRACT_DIR / name
        if not path.exists() or md5(path) != expected:
            raise SystemExit(f"NG: {name} が未到着か、MD5 が一致しません（{mf.name} は取り込まない）")
    print("load:", mf.name)
    for name, _ in entries:               # 2) 取り込み
        with zipfile.ZipFile(EXTRACT_DIR / name) as zf:
            for member in zf.namelist():
                if member.endswith(".pecsv"):          # 主キー抽出：存在する主キーの一覧
                    mark_deleted(r[0] for r in rows_of(zf, member))
                else:
                    upsert(rows_of(zf, member))
    db.execute("INSERT INTO processed_manifest VALUES (?)", (mf.name,))
    db.commit()                           # 3) マニフェスト単位でコミット（失敗したら同じマニフェストを再処理する）


for row in db.execute("SELECT * FROM ap_invoice_header ORDER BY invoice_id"):
    print(row)

1 日目（全件）と、再実行

$ python make_sample_extract.py 1
created:
  file_sampleam_invoiceheaderpvo-batch1001-20260921_020000.zip
  MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9001.MF
$ python ingest_extract.py
load: MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9001.MF
(10001, 'INV-2026-0001', 165000, '2026-09-10T10:00:00', 0)
(10002, 'INV-2026-0002', 44000, '2026-09-12T10:00:00', 0)
(10003, 'INV-2026-0003', 33000, '2026-09-15T10:00:00', 0)
(10004, 'INV-2026-0004', 11000, '2026-09-20T10:00:00', 0)
(10005, 'INV-2026-0005', 22000, '2026-09-18T10:00:00', 0)
$ python ingest_extract.py
skip (processed): MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9001.MF
(10001, 'INV-2026-0001', 165000, '2026-09-10T10:00:00', 0)
(10002, 'INV-2026-0002', 44000, '2026-09-12T10:00:00', 0)
(10003, 'INV-2026-0003', 33000, '2026-09-15T10:00:00', 0)
(10004, 'INV-2026-0004', 11000, '2026-09-20T10:00:00', 0)
(10005, 'INV-2026-0005', 22000, '2026-09-18T10:00:00', 0)

2 日目（増分と主キー抽出）

$ python make_sample_extract.py 2
created:
  file_sampleam_invoiceheaderpvo-batch1002-20260922_020000.zip
  file_sampleam_invoiceheaderpvo-batch1003-20260922_030000.zip
  MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9002.MF
  MANIFEST_PRIMARY_KEYS_5001-SCHEDULE_7002_REQUEST_9003.MF
$ python ingest_extract.py
skip (processed): MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9001.MF
load: MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9002.MF
load: MANIFEST_PRIMARY_KEYS_5001-SCHEDULE_7002_REQUEST_9003.MF
  deleted flag set: 1 rows
(10001, 'INV-2026-0001', 165000, '2026-09-10T10:00:00', 0)
(10002, 'INV-2026-0002', 44000, '2026-09-12T10:00:00', 0)
(10003, 'INV-2026-0003', 35000, '2026-09-21T09:00:00', 0)
(10004, 'INV-2026-0004', 11000, '2026-09-20T10:00:00', 0)
(10005, 'INV-2026-0005', 22000, '2026-09-18T10:00:00', 1)
(10006, 'INV-2026-0006', 18000, '2026-09-21T15:00:00', 0)

破損の実験：extract フォルダと dwh.db を削除し、1 日目のファイルを作り直してから、ZIP の末尾に 1 バイトを足して取り込みます。

$ python make_sample_extract.py 1
created:
  file_sampleam_invoiceheaderpvo-batch1001-20260921_020000.zip
  MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9001.MF
$ python -c "open('extract/file_sampleam_invoiceheaderpvo-batch1001-20260921_020000.zip','ab').write(b'x')"
$ python ingest_extract.py
NG: file_sampleam_invoiceheaderpvo-batch1001-20260921_020000.zip が未到着か、MD5 が一致しません（MANIFEST_DATA_5001-SCHEDULE_7001_REQUEST_9001.MF は取り込まない）

確認すること

  • 1 日目：5 行が入ります。再実行すると、処理済みのマニフェストはスキップされ、結果は変わりません
  • 2 日目：INV-2026-0003 の金額が 35,000 に更新され、INV-2026-0006 が追加されます。INV-2026-0004 は、前回の抽出と重なる範囲（Prune Time の 24 時間）にあるため再び届きますが、内容が同じなので結果は変わりません
  • 2 日目の主キー抽出：一覧にない INV-2026-0005 に is_deleted = 1（最後の列）が付きます。行は消さずに履歴を残しています
  • 破損の実験：MD5 が一致しないため、何も取り込まずに停止します

実機のファイルで試すときは、ステップ 6 で取得したファイルを extract フォルダに置き、COLUMNS・テーブル定義・HAS_HEADER を、実際のデータストアに合わせます。列の並びは、メタデータ（.mdcsv や Export Metadata Definition の出力）と実際の CSV を見比べて確認します。

確認ポイント

Section titled “確認ポイント”
  • 設計シート 1 の各行に、粒度・主キー・増分の日付列・取り出さない列が書かれている
  • Prune Time を決め、重複して届く前提で受け側を設計した
  • 削除検知の方式（ジョブ種別・頻度・受け側の処理）が決まっている
  • 取込が、マニフェストの全ファイルの存在と MD5 を確認してから始まる
  • 同じマニフェストを再実行しても、結果が変わらない
  • 取込の順序が、ESS 要求 ID の昇順（実行順）になっている
  • （実機）Test Connection が成功し、ZIP・マニフェスト・ステータス JSON がバケットに出た
  • （実機）ステータス JSON の rowCount が、画面で数えた件数と合った

解答例

Section titled “解答例”
設計シート 1（抽出対象）の解答例を表示
対象 検索語（実機で確認） 粒度・主キー 取り出さない列・備考 ジョブ
請求書ヘッダー Invoice（ヘッダー粒度のもの） 1 請求書 1 行・請求書 ID 仕入先の銀行口座など A
請求書配分 Invoice と Distribution 1 配分 1 行・配分 ID（親は請求書 ID と明細番号） 同上。行数が多いため別ジョブ B
仕入先 Supplier 1 仕入先 1 行・仕入先 ID 銀行口座・納税者番号などは対象外 A
GL 仕訳明細 Journal 1 仕訳明細 1 行・仕訳ヘッダー ID と行番号の組 最大のデータ量。別ジョブにし、Support Chunking を検討 C
  • 検索語は目安です。実機で候補を開き、Select Columns に、主キー・増分の判定に使う日付列・作成日列（Is Creation Date）があるかを確かめて確定します。確定した名前と列は、このシートに記録します
  • 増分の日付列と作成日列は、データストアごとに実機で確認します。見つからないデータストアは、全件抽出にするか、別のデータストアを探します
  • 主キーの確認は 主要テーブルマップ が出発点です（請求書 INVOICE_ID、配分 INVOICE_DISTRIBUTION_ID、仕入先 VENDOR_ID、仕訳明細 JE_HEADER_ID＋JE_LINE_NUM）
設計シート 2（抽出方針）の解答例を表示
項目 決定 理由
Initial Extract Date 2024-04-01 分析の対象期間の開始。全履歴は不要で、データ量と初回の所要時間を抑える。作成日列の定義が前提
初回の全件抽出 土曜の夜に、ジョブ A・B・C を別々に実行 10 時間のタイムアウトと、失敗の影響範囲を分ける。日曜のうちに取込と検証を終える
増分 毎日 02:00（Weekly で全曜日を指定。画面で確認） 業務時間外。時間単位の抽出は、データベースへの負荷から避ける
Prune Time 1,440（既定のまま） 毎日の抽出に合わせる。ヘッダーと明細のコミットのずれや、抽出中の更新の取りこぼしを避ける。重複は受け側のアップサートで吸収する
Reset to Full Extract 列の追加・Query Filter の変更・取込の欠落が見つかったときに、対象のデータストアだけで使う 全件を取り直す。受け側は同じ主キーのアップサートと、主キー差分で整える
通知 失敗時のみ、IT 運用チームのメールアドレスへ（Define Notification） R4 を満たす。成功の確認は取込側の監視で行い、通知を増やさない
設計シート 3（削除検知）の解答例を表示
項目 決定 理由
方式 ジョブ種別 Active Primary Key Extract（データ抽出と同じジョブで行うなら Application Data and Active Primary Key Extract） 増分では物理削除を拾えない
対象 請求書ヘッダー・配分・仕入先。GL 仕訳明細は、業務で削除が起こるかを確認してから決める 主キー抽出は全件分の主キーを読む。削除が起こらないデータストアには負荷をかけない
頻度 毎日 03:00（データ抽出の後） R3（翌朝までに反映）。仕訳明細のように行数が多いものは、週次への変更も検討する
受け側の処理 一覧にない行に削除フラグ（is_deleted = 1）を立てる。行が一覧に戻ったら 0 に戻す 履歴と監査のため、物理削除しない
取込の順序 ESS 要求 ID の昇順（実行順） 遅れて取り込んでも、削除と再出現の判定が実行順とずれない
設計シート 4（出力と自動化）の解答例を表示
項目 決定
格納先と保持 OCI Object Storage のバケット 1 つ（データレイクと同じ OCI）。30 日で削除（OCI 側で設定。BICC のジョブ種別 Delete Files in Storage でも掃除できる）
取込の起動 毎日 06:00。バケットのマニフェストを一覧し、未処理のものを ESS 要求 ID の昇順で処理する
完全性の検証 マニフェストの全ファイルが存在し、MD5 が一致してから取り込む。ステータス JSON の status と rowCount も確認する（JSON の構造は実ファイルで確認）
冪等性 主キーでアップサート。処理済みのマニフェストを記録し、再実行しても同じ結果になるようにする。コミットはマニフェスト単位
失敗時 検証が NG なら取り込まずに止め、運用チームへ通知する。ファイルが揃ってから、同じマニフェストを再実行する。BICC の抽出が失敗した場合は、Last Extract Date が進まないため、次回の増分で取り戻される
監視 抽出の所要時間（totalDurationInSec など）、行数の推移、未処理のマニフェストの滞留
権限 BICC を動かすユーザー、OCI のバケットへ書き込むユーザー、取込処理が読み取るユーザーを分け、それぞれ最小権限にする（R7）
取込処理の設計判断とコードの対応を表示
設計の判断 コードの該当箇所
完全性の検証 全ファイルの存在と MD5 を確認してから取り込む（NG: で停止）
冪等性 processed_manifest への記録と、ON CONFLICT ... DO UPDATE（アップサート）
実行順 request_id() で ESS 要求 ID を数値として並べる（文字列順だと 9 が 10 より後になる）
削除検知 .pecsv の主キー一覧にない行を mark_deleted() で論理削除
再出現 アップサートで is_deleted を 0 に戻す
失敗時の整合性 マニフェスト単位でコミット。途中で失敗したら、同じマニフェストを最初から再処理できる

よくあるエラーと対処

Section titled “よくあるエラーと対処”
症状 原因と対処
Test Connection が失敗する OCI のユーザーに公開鍵が登録されていない、または OCID・Namespace・Bucket・Host の入力誤りです。公開鍵のフィンガープリント、各 ID、バケットへの権限を順に確認します
Initial Extract Date を設定できない 作成日を表す列（Is Creation Date）の定義が必要です。Advanced Extract Configuration で確認します
取り込んだ後に行が二重になる INSERT だけで取り込んでいます。主キーでアップサートにします（増分は重複して届く前提です）
削除したはずの請求書が残る 主キー抽出を設計していない、または取込の順序が実行順になっていません
数値の桁が落ちる、抽出が失敗する 15 桁を超える数値や CLOB の列は、Cast to String が必要です。文字列の上限を超えるデータがあると、抽出が失敗します
書き込み途中のファイルを読んでしまう ファイルの一覧ではなく、マニフェストを起点にします。全ファイルの存在と MD5 を確認してから取り込みます
実行がタイムアウトする タイムアウトは既定で 10 時間で、超えると失敗します。ジョブを分ける、Query Filter で範囲を絞る、Support Chunking を使うことを検討します
UCM のファイルが見つからない UCM のファイルは既定で 90 日で期限切れになります。取込の遅れと保持期間を確認します

発展課題

Section titled “発展課題”
  1. ステータス JSON の rowCount と、取り込んだ件数を突合する処理を、取込処理に追加する（JSON の構造は実ファイルで確認する）
  2. 取込処理を、OIC のスケジュール統合で作る場合の構成を設計する（スケジュール統合とファイル処理）
  3. 後継の Data Extraction ツールで、同じ要件（対象・増分・削除の反映）を満たせるかを、対象オブジェクトの有無から調べ、BICC との差を表にまとめる（連携方式の全体像）
  4. 抽出を 1 日 2 回に増やす場合の、Prune Time・取込の順序・Fusion への負荷の影響を設計し直す
  5. 出力先を UCM に切り替える場合の、マニフェスト形式（MANIFEST.MF、vo_name;ucm_document_id;md5_check_sum_value）の違いと、取得方法を整理する（UCM からの取得方法は公式ドキュメントで確認する）
  6. 失敗時、環境のリフレッシュ後、四半期アップデート後の運用手順書を作る（環境管理・四半期アップデート・テスト）

公式ドキュメント

Section titled “公式ドキュメント”

関連ページ

Section titled “関連ページ”

本サイトは学習目的で個人が作成した非公式の教材であり、Oracle Corporation およびその関連会社とは関係ありません。内容の正確性は保証しません。最新の仕様・試験範囲は必ず公式ドキュメントと Oracle University の情報で確認してください。

模擬テストの問題は公開されている出題範囲（Exam Topics）と公式ドキュメントをもとに独自に作成したもので、実際の試験問題ではありません。

Oracle、Java、MySQL は Oracle Corporation およびその子会社、関連会社の米国およびその他の国における登録商標です。その他の製品名・会社名は各社の商標です。

© 2026 Fusion ERP 実装ラボ