AIDPノートブック(データ分析用の統合開発環境)内から、AIDP の `aidataplatform` Spark形式ハンドラー(データ読み書き機能)を経由して PostgreSQL(リレーショナルデータベース)の読み書きを行います。 次のような場合に使用: ユーザーが PostgreSQL、Postgres、「psql」に言及している、または接続対象の PostgreSQL ホスト/ポート情報を持っている場合。 公式の `POSTGRESQL` コネクター(接続機能)サンプルを優先的に使用し、ビルトインコネクターで SSL必須モード(sslmode=require)を表現できない場合にのみ、Spark のネイティブ JDBC(直接接続技術)をフォールバック(代替手段)として使用してください。
Read or write PostgreSQL from an AIDP notebook via the AIDP `aidataplatform` Spark format handler. Use when the user mentions PostgreSQL, Postgres, "psql", or has a Postgres host/port to connect to. Prefer the official `POSTGRESQL` connector sample; use native Spark JDBC only as an SSL-required fallback when the built-in connector cannot express sslmode=require.
aidp-postgresql — AIDP aidataplatform 経由での PostgreSQL 接続aidp-mysqlaidp-sqlserveraidp-jdbc-customaidataplatform フォーマット(デフォルト)まず公式の AIDP PostgreSQL コネクタサンプルを使用してください。
import os
from oracle_ai_data_platform_connectors.aidataplatform import (
AIDP_FORMAT, aidataplatform_options,
)
opts = aidataplatform_options(
type="POSTGRESQL",
host=os.environ["PG_HOST"],
port=int(os.environ.get("PG_PORT", "5432")),
user=os.environ["PG_USER"],
password=os.environ["PG_PASSWORD"],
schema=os.environ.get("PG_SCHEMA", "public"),
table=os.environ["PG_TABLE"],
)
df = spark.read.format(AIDP_FORMAT).options(**opts).load()
df.show()
opts = aidataplatform_options(
type="POSTGRESQL",
host=os.environ["PG_HOST"],
port=int(os.environ.get("PG_PORT", "5432")),
user=os.environ["PG_USER"],
password=os.environ["PG_PASSWORD"],
schema=os.environ.get("PG_SCHEMA", "public"),
table=os.environ["PG_TARGET_TABLE"],
extra={"write.mode": "CREATE"}, # CREATE | APPEND | OVERWRITE
)
df.write.format(AIDP_FORMAT).options(**opts).save()
opts = aidataplatform_options(
type="POSTGRESQL",
host=os.environ["PG_HOST"],
port=int(os.environ.get("PG_PORT", "5432")),
user=os.environ["PG_USER"],
password=os.environ["PG_PASSWORD"],
extra={
"pushdown.sql": (
f"SELECT * FROM {os.environ.get('PG_SCHEMA', 'public')}.{os.environ['PG_TABLE']} "
"WHERE updated_at >= DATE '2024-01-01'"
),
},
)
df = spark.read.format(AIDP_FORMAT).options(**opts).load()
df.show()
接続先の PostgreSQL サービスが非 TLS 接続を拒否し、かつ AIDP コネクタが必要な SSL オプションを渡せない場合にのみ使用してください。
確認されている障害モードは [PostgreSQL]connection is insecure (try using sslmode=require) です。
その場合は、URL に sslmode=require を埋め込んだ Spark ネイティブ JDBC を使用します。
クラスタに org.postgresql.Driver がプリインストールされていない可能性があるため、aidp-jdbc-custom と同様の方法でランタイム時にロードしてください。
import os, urllib.request
from py4j.java_gateway import java_import
JAR_PATH = "/tmp/postgresql-42.7.4.jar"
if not os.path.exists(JAR_PATH):
urllib.request.urlretrieve(
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.7.4/postgresql-42.7.4.jar",
JAR_PATH,
)
# ドライバ JVM にドライバを登録
gw = spark._sc._gateway
url = spark._jvm.java.io.File(JAR_PATH).toURI().toURL()
arr = gw.new_array(spark._jvm.java.net.URL, 1); arr[0] = url
ucl = spark._jvm.java.net.URLClassLoader(arr, spark._jvm.java.lang.ClassLoader.getSystemClassLoader())
spark._jvm.java.lang.Thread.currentThread().setContextClassLoader(ucl)
DriverCls = spark._jvm.java.lang.Class.forName("org.postgresql.Driver", True, ucl)
spark._jvm.java.sql.DriverManager.registerDriver(DriverCls.newInstance())
spark._jsc.addJar(JAR_PATH) # エグゼキュータに配布
# 読み込み — sslmode=require を URL に埋め込んでいることに注意
JDBC_URL = (
f"jdbc:postgresql://{os.environ['PG_HOST']}:{os.environ.get('PG_PORT','5432')}"
f"/{os.environ['PG_DB']}?sslmode=require"
)
df = (spark.read.format("jdbc")
.option("url", JDBC_URL)
.option("driver", "org.postgresql.Driver")
.option("user", os.environ["PG_USER"])
.option("password", os.environ["PG_PASSWORD"])
.option("dbtable", f"{os.environ.get('PG_SCHEMA','public')}.{os.environ['PG_TABLE']}")
.load())
df.show(5)
まず aidataplatform を使用すること。
これが公式のサンプルパスであり、専用のバンドルドライバが付属しています。
SSL フォールバック —
AIDP の POSTGRESQL ハンドラが SSL オプションを渡せず、サーバが sslmode=require を要求する場合は、オプション B(ネイティブ Spark JDBC)を使用してください。
2026-04-27 に Neon サーバレス 17.8 で動作確認済み。
ネイティブ Spark JDBC 用のバンドルドライバは存在しない —
オプション B を使用する場合は、PostgreSQL の JAR をランタイム時にロードし、URLClassLoader + DriverManager.registerDriver + spark._jsc.addJar で登録してください。
ネットワーク到達性 —
PostgreSQL が AIDP クラスタの NAT 出口 IP から到達可能である必要があります。
パブリックインターネット上のエンドポイント(Neon、Supabase、RDS パブリック)はクラスタの NAT 経路経由で接続可能です。
ユーザー管理の VCN 内にセルフホストされた PostgreSQL は、明示的な VCN ピアリングなしではクラスタの Pod CIDR からルートが存在しないため、通常は接続できません。
疎通確認には Python の socket を使用してください: socket.create_connection((host, 5432), timeout=8)
schema について —
schema は Postgres の論理スキーマ(例: public)であり、データベース名ではありません。
データベース名は別途 PG_DB 環境変数として JDBC URL に渡します。
書き込みモード —
CREATE(既存の場合は失敗)、APPEND、OVERWRITE の3種類。デフォルトは CREATE。
aidp-postgresql — PostgreSQL via AIDP aidataplatformaidp-mysql.aidp-sqlserver.aidp-jdbc-custom.aidataplatform format (default)Use the official AIDP PostgreSQL connector sample first.
import os
from oracle_ai_data_platform_connectors.aidataplatform import (
AIDP_FORMAT, aidataplatform_options,
)
opts = aidataplatform_options(
type="POSTGRESQL",
host=os.environ["PG_HOST"],
port=int(os.environ.get("PG_PORT", "5432")),
user=os.environ["PG_USER"],
password=os.environ["PG_PASSWORD"],
schema=os.environ.get("PG_SCHEMA", "public"),
table=os.environ["PG_TABLE"],
)
df = spark.read.format(AIDP_FORMAT).options(**opts).load()
df.show()
opts = aidataplatform_options(
type="POSTGRESQL",
host=os.environ["PG_HOST"],
port=int(os.environ.get("PG_PORT", "5432")),
user=os.environ["PG_USER"],
password=os.environ["PG_PASSWORD"],
schema=os.environ.get("PG_SCHEMA", "public"),
table=os.environ["PG_TARGET_TABLE"],
extra={"write.mode": "CREATE"}, # CREATE | APPEND | OVERWRITE
)
df.write.format(AIDP_FORMAT).options(**opts).save()
opts = aidataplatform_options(
type="POSTGRESQL",
host=os.environ["PG_HOST"],
port=int(os.environ.get("PG_PORT", "5432")),
user=os.environ["PG_USER"],
password=os.environ["PG_PASSWORD"],
extra={
"pushdown.sql": (
f"SELECT * FROM {os.environ.get('PG_SCHEMA', 'public')}.{os.environ['PG_TABLE']} "
"WHERE updated_at >= DATE '2024-01-01'"
),
},
)
df = spark.read.format(AIDP_FORMAT).options(**opts).load()
df.show()
Use this only when the target PostgreSQL service rejects non-TLS connections and the AIDP connector cannot pass the required SSL option. The observed failure mode is [PostgreSQL]connection is insecure (try using sslmode=require). In that case, use Spark native JDBC with URL-embedded sslmode=require.
The cluster may not have org.postgresql.Driver pre-installed, so runtime-load it the same way aidp-jdbc-custom does.
import os, urllib.request
from py4j.java_gateway import java_import
JAR_PATH = "/tmp/postgresql-42.7.4.jar"
if not os.path.exists(JAR_PATH):
urllib.request.urlretrieve(
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.7.4/postgresql-42.7.4.jar",
JAR_PATH,
)
# Register driver on driver JVM
gw = spark._sc._gateway
url = spark._jvm.java.io.File(JAR_PATH).toURI().toURL()
arr = gw.new_array(spark._jvm.java.net.URL, 1); arr[0] = url
ucl = spark._jvm.java.net.URLClassLoader(arr, spark._jvm.java.lang.ClassLoader.getSystemClassLoader())
spark._jvm.java.lang.Thread.currentThread().setContextClassLoader(ucl)
DriverCls = spark._jvm.java.lang.Class.forName("org.postgresql.Driver", True, ucl)
spark._jvm.java.sql.DriverManager.registerDriver(DriverCls.newInstance())
spark._jsc.addJar(JAR_PATH) # distribute to executors
# Now read — note sslmode=require URL-embedded
JDBC_URL = (
f"jdbc:postgresql://{os.environ['PG_HOST']}:{os.environ.get('PG_PORT','5432')}"
f"/{os.environ['PG_DB']}?sslmode=require"
)
df = (spark.read.format("jdbc")
.option("url", JDBC_URL)
.option("driver", "org.postgresql.Driver")
.option("user", os.environ["PG_USER"])
.option("password", os.environ["PG_PASSWORD"])
.option("dbtable", f"{os.environ.get('PG_SCHEMA','public')}.{os.environ['PG_TABLE']}")
.load())
df.show(5)
aidataplatform first. It is the official sample path and has its own bundled driver.POSTGRESQL handler cannot pass SSL options and the server requires sslmode=require, use Option B with native Spark JDBC. Verified live 2026-04-27 against Neon serverless 17.8.URLClassLoader + DriverManager.registerDriver + spark._jsc.addJar.socket.create_connection((host, 5432), timeout=8).schema is the Postgres logical schema (e.g. public), not the database name. The database name is a separate PG_DB env var that goes into the JDBC URL.CREATE (fail if exists), APPEND, OVERWRITE. Default is CREATE.原文・著作権は Anthropic および各プラグイン作者に帰属します。日本語訳は Claude API による自動翻訳です。