[Autonomous AI Database]Data TransformsのジョブをREST APIで実行してみた

はじめに

Data TransformsはREST APIをサポートしています。作成済みのデータフローをREST APIで実行してみました。
REST APIの詳細は以下のドキュメントをご覧ください。

前提

• 環境はPriavete EndpointのADB26ai(最低ECPU数で指定）付属のData Transformsを利用。
• ADBと同じVCNのPublic EndpointにComputeを作成済み(Oracle Linux8)。
• ADBのPrivate Endpointで指定したPrivate Subnetのセキュリティ・ルールでComputeを作成したPublic SubnetからのTCPアクセスはすべて許可済み

1.Data Transformsでデータ・フローを作成

adminユーザでData Transformsにログインし、実行するサンプルのデータフローを作成し、正しく実行されることを確認しておきます。今回はオブジェクトストレージにあるCSVファイルのデータをADBの表にデータを投入するフローを作成しました。

オプションでTruncate target tableをTrueにセットし、データは洗い替えになるように設定しています。

実行すると正常完了しました。

2.Computeでスクリプトを作成、実行

Computeからシェルスクリプトを使ってREST APIでデータフローを実行させます。こちらのスクリプトを作成しました。
<HOST>はOCIコンソールのツール構成で確認できるアクセスURLにあわせます。今回はプライベート・アクセスURLを利用しました。パブリック・アクセスをOnにしている場合はパブリック・アクセスURLでも大丈夫でした。

callrestapi.sh

#!/usr/bin/env bash

# ------------
# 1. トークン取得
# ------------

# トークン取得用エンドポイント
TOKEN_URL="https://<HOST>.adb.ap-osaka-1.oraclecloudapps.com/odi/broker/pdbcs/public/v1/token"

# トークン取得に必要な認証情報
AUTH_PAYLOAD='{
  "username": "admin",
  "password": "<adminユーザーのPASSWORD>",
  "tenant_name": "<TenancyのOCID>",
  "database_name": "<ADB名>",
  "cloud_database_name": "<ADBのOCID>",
  "grant_type": "password"
}'

# トークン取得リクエスト
echo "トークン取得中..."
TOKEN=$(curl -s -X POST "$TOKEN_URL" \
        -H "Content-Type: application/json" \
        -d "$AUTH_PAYLOAD" | jq -r '.access_token')

# 取得に失敗した場合はエラーメッセージを出して終了
if [[ -z "$TOKEN" || "$TOKEN" == "null" ]]; then
    echo "トークンの取得に失敗しました。レスポンスを確認してください。" >&2
    exit 1
fi

echo "トークン取得完了: $TOKEN"

# ---------------------
# 2.対象データフロー情報取得
# ---------------------

# Data Flowの情報取得用エンドポイント（dataflows）
# プロジェクト名とフロー名を指定
API_URL="https://<HOST>.adb.ap-osaka-1.oraclecloudapps.com/odi/dt-rest/v2/dataflows?name=test_flow3&projectName=test_proj"


# データフロー情報取得リクエスト
echo "dataflows API へリクエスト送信中..."

RESPONSE=$(curl -s -X GET "$API_URL" \
            -H "Authorization: Bearer $TOKEN" \
            -H "Accept: application/json")

echo "dataflows API のレスポンス:"
echo $RESPONSE | jq .
echo "dataflows API のレスポンスからIDを取得:"

FlowID=$(echo "$RESPONSE" | jq -r '.[0].dataFlowId')
echo "FlowID: $FlowID"

# ------------
# 3.ジョブの実行
# ------------

# ジョブ実行エンドポイント（submit）
API_URL="https://<HOST>.adb.ap-osaka-1.oraclecloudapps.com/odi/dt-rest/v2/jobs/submit"

# ジョブ実行に必要な情報
REQUEST_PAYLOAD='{
  "action":"RUN",
  "objectType":"DATAFLOW",
  "objectId":"'$FlowID'",
  "objectName":"test_flow3",
  "ignorePreviousRunningJob":true,
  "jobName":"callbyrest_job1"
  }'

# ジョブ実行リクエスト
echo "実行 API 呼び出し中..."
RESPONSE=$(curl -s -X POST "$API_URL" \
            -H "Authorization: Bearer $TOKEN" \
            -H "Content-Type: application/json" \
            -d "$REQUEST_PAYLOAD" )

echo "実行 API のレスポンス:"
echo "$RESPONSE" | jq .

# ----------------------
# 4.ジョブの実行ステータスを取得
# ----------------------

# ジョブ実行リクエストのレスポンスからjob ID を取得
JOB_ID=$(echo "$RESPONSE" | jq -r '.jobId')
if [[ -z "$JOB_ID" || "$JOB_ID" == "null" ]]; then
    echo "job_id が取得できませんでした" >&2
    echo "$RESPONSE" | jq .
    exit 1
fi

echo "job_id: $JOB_ID"

# 進捗確認用エンドポイント（ポーリング）
PROGRESS_URL="https://<HOST>.adb.ap-osaka-1.oraclecloudapps.com/odi/dt-rest/v2/jobs/$JOB_ID"

# 進捗確認のリクエスト
while true; do
    STATUS_RESP=$(curl -s -X GET "$PROGRESS_URL" -H "Authorization: Bearer $TOKEN")
    STATUS=$(echo "$STATUS_RESP" | jq -r '.status')
    echo "現在のステータス: $STATUS"
    if [[ "$STATUS" == "DONE" ]]; then
        echo "処理完了"
        echo "$STATUS_RESP" | jq .
        break
    elif [[ "$STATUS" == "ERROR" ]]; then
        echo "処理失敗" >&2
        echo "$STATUS_RESP" | jq .
        exit 1
    fi
    sleep 10  # 10 秒待機して再チェック
done

実行します。結果はこちらのとおりです。

[opc@compute1 work]$ ./callrestapi.sh
トークン取得中...
トークン取得完了: eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJBRE1JTiIsInRlbmFudF9uYW1lIjoiT0NJRDEuVEVOQU5DWS5PQzEuLkFBQUFBQUFBM01CN1dSQ1kyTFMzVTNKU1kyU09RNUNLM0xDM1E0TUNaSVRQREFZTUJVQVpDNVRLR1VDQSIsImRhdGFiYXNlX25hbWUiOiJBUkFBVFAyMyIsInBoeXNpY2FsX2RhdGFiYXNlX25hbWUiOiJMWDQ2S044VUtTOTJVTkxfQVJBQVRQMjMiLCJ:
X9ZUPzgJQoNPbsdEy7tOHPFU2ZpxqppZv2qTwwYNo6jwOp4mesAeJSFfLIRz5eE-dhYEtgSiIniHIwcMKMAQnmXuveGK86jQS4dl14SEpviJ4K7LhiXB6LmuwMS2PihsunxqMWJXXtOY5o2HJ16vgnxA8E4EP6K6-ggd0M0fbk2NpgWHwe2REM_W7R-A
dataflows API へリクエスト送信中...
dataflows API のレスポンス:
[
  {
    "name": "test_flow3",
    "dataFlowId": "57cfb042-eb22-4034-a6af-0914f3ae5325",
    "dateCreated": "2026-01-27T12:35:36.000+0000",
    "dateUpdated": "2026-02-13T13:08:31.000+0000",
    "projectName": "test_proj",
    "parentFolder": "DefaultFolder",
    "description": "大きなサイズのファイルを利用"
  }
]
dataflows API のレスポンスからIDを取得:
FlowID: 57cfb042-eb22-4034-a6af-0914f3ae5325
実行 API 呼び出し中...
実行 API のレスポンス:
{
  "status": "WAITING",
  "jobName": "callbyrest_job1",
  "jobNumber": 228,
  "jobId": "2a367be7-bc4f-4ca5-8894-58b95203f1a7",
  "executionStatistics": {
    "timeZone": "UTC",
    "serverTimeZone": "UTC"
  },
  "recordStatistics": {
    "insertCount": 0,
    "updateCount": 0,
    "deleteCount": 0,
    "recordCount": 0,
    "errorCount": 0
  },
  "synchronous": false
}
job_id: 2a367be7-bc4f-4ca5-8894-58b95203f1a7
現在のステータス: RUNNING
現在のステータス: RUNNING
:
現在のステータス: RUNNING
現在のステータス: RUNNING
現在のステータス: DONE
処理完了
{
  "status": "DONE",
  "jobName": "callbyrest_job1",
  "jobNumber": 228,
  "jobId": "2a367be7-bc4f-4ca5-8894-58b95203f1a7",
  "startDate": "2026-02-16T06:24:35.000+0000",
  "endDate": "2026-02-16T06:26:52.000+0000",
  "executionStatistics": {
    "startDate": "2026-02-16T06:24:35.000+0000",
    "endDate": "2026-02-16T06:26:52.000+0000",
    "duration": 137,
    "returnCode": "0",
    "timeZone": "UTC",
    "serverTimeZone": "UTC"
  },
  "recordStatistics": {
    "insertCount": 7100000,
    "updateCount": 0,
    "deleteCount": 0,
    "recordCount": 7100000,
    "errorCount": 0
  },
  "synchronous": false
}

3.Data TransformsのUIで確認

実行結果にあったJobNumber: 228の情報がありました。指定したジョブ名で作成されています。

4.気づいたこと

Data Transformsが起動していないと、トークン取得以外のAPI実行はエラーになります。

{
  "status": 503,
  "message": "Service Status",
  "result": "Service provisioning in progress"
}

ADBのData Transformsは指定時間アイドルの場合停止されるので、上記エラーの場合に再実行するようにコードを書いたほうがよさそうです。

おわりに

Data Transformsのジョブ実行をREST APIで行っていました。
ジョブ管理ツールなど別のサービスから実行させたい場合に利用できるのではないかと思います。

参考資料

• REST API for Oracle Data Transforms