neo4j-import-skill

Neo4j インポートスキル

使用する場合

• CSV、JSON、Parquet ファイルを Neo4j にインポートする場合
• ノードとリレーションシップをバッチでアップサートする場合（UNWIND + CALL IN TRANSACTIONS）
• リレーショナルデータを移行する場合（SQL → グラフ）
• 大規模なデータセットをオフラインで一括読み込みする場合（neo4j-admin import）
• オンライン（Cypher）とオフライン（admin）のインポート方法を選択する場合
• インポートの完全性を検証する場合（件数、制約、インデックスの状態）

使用しない場合

• 非構造化ドキュメント、PDF、ベクターチャンク → neo4j-document-import-skill
• 稼働中のアプリケーションによる書き込み（アプリコード内の MERGE/CREATE） → neo4j-cypher-skill
• neo4j-admin によるバックアップ／復元／設定 → neo4j-cli-tools-skill
• 既存のグラフから GDS アルゴリズムのプロジェクションを作成する場合 → neo4j-gds-skill

---

方法選択テーブル

データセットのサイズ	DB の状態	ソース	方法
任意のサイズ	オンライン	CSV（Aura またはローカル）	LOAD CSV + CALL IN TRANSACTIONS
< 1M 行	オンライン	リスト／API レスポンス	UNWIND + CALL IN TRANSACTIONS
> 10M 行	オフライン（ローカル／セルフマネージド）	CSV / Parquet	neo4j-admin database import full
任意のサイズ	オンライン	APOC が利用可能	apoc.periodic.iterate + apoc.load.csv
任意のサイズ	オンライン	JSON/API	apoc.load.json またはドライバーによるバッチ処理
差分の増分	オフライン（Enterprise）	CSV	neo4j-admin database import incremental

Aura：https:// URL のみ対応します — file:/// は使用できません。neo4j-admin import はセルフマネージド環境でのみ使用してください。

---

インポート前のチェックリスト

必ずこの順序で実行してください。省略すると、原因の特定が難しい重複や、インデックスが使用されない問題が発生します。

インポート前に制約を作成します。追加のインデックスはインポート後に作成します。

• 制約によって暗黙的な RANGE インデックスが作成され、ロード中の MERGE で使用されるほか、一意性も強制されます
• 追加の一意でないインデックス（TEXT、キー以外のプロパティに対する RANGE、FULLTEXT）はロード後に作成します。Neo4j はコミット済みデータから非同期でインデックスを作成するため、populationPercent が 100% になるまでポーリングしてください
• インポート前に追加のインデックスを作成すると、メリットがないままロード中のすべての書き込みが遅くなります

1. 一意性制約を作成する（MERGE で使用されるインデックスを有効にします）：

   CREATE CONSTRAINT IF NOT EXISTS FOR (n:Person) REQUIRE n.id IS UNIQUE;
   CREATE CONSTRAINT IF NOT EXISTS FOR (n:Movie)  REQUIRE n.movieId IS UNIQUE;
   

   Neo4j 2026.06+（Enterprise/Aura、GA）： ALTER CURRENT GRAPH TYPE SET { … } を使うと、個別の制約文をすべて単一の宣言型ブロックに置き換えられます。neo4j-cypher-skill/references/graph-type.md を参照してください。Community Edition または 2026.02 より前では、個別の CREATE CONSTRAINT を使用してください。

2. apoc. プロシージャを使用する場合は APOC を確認する*：

   RETURN apoc.version();
   

   失敗した場合 → APOC がインストールされていません。代わりに通常の LOAD CSV を使用してください。

3. 対象がレプリカではなく PRIMARY であることを確認する：

   CALL dbms.cluster.role() YIELD role RETURN role;
   

   ロールが PRIMARY でない場合 → 停止してください。書き込み先を PRIMARY エンドポイントに変更してください。

4. インポート前にソースファイルの行数を数える（エンコードの問題を早期に検出します）：

   wc -l data/persons.csv    # Linux/macOS
   

5. UTF-8 エンコードを確認する — LOAD CSV には UTF-8 が必要です。必要に応じて再エンコードしてください：

   file -i persons.csv       # Check encoding
   iconv -f latin1 -t utf-8 persons.csv > persons_utf8.csv
   

---

LOAD CSV のパターン

型変換と null 処理を伴う基本的なノードのインポート

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///persons.csv' AS row
CALL (row) {
  MERGE (p:Person {id: row.id})
  ON CREATE SET
    p.name       = row.name,
    p.age        = toIntegerOrNull(row.age),
    p.score      = toFloatOrNull(row.score),
    p.active     = toBoolean(row.active),
    p.born       = CASE WHEN row.born IS NOT NULL AND row.born <> '' THEN date(row.born) ELSE null END,
    p.createdAt  = datetime()
  ON MATCH SET
    p.updatedAt  = datetime()
} IN TRANSACTIONS OF 10000 ROWS
  ON ERROR CONTINUE
  REPORT STATUS AS s
RETURN s.transactionId, s.committed, s.errorMessage

null／空文字列のルール：

• CSV に列がない場合 → null（安全です）
• CSV の空文字列 "" → null ではなく "" として保存されます — 変換するには nullIf(row.x, '') を使用してください
• toInteger(null) は例外をスローします → 常に toIntegerOrNull() を使用してください
• toFloat(null) は例外をスローします → 常に toFloatOrNull() を使用してください
• Neo4j は null プロパティを保存しません — SET 時に通知なく削除されます

リレーションシップのインポート（先にノードが存在している必要があります）

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///knows.csv' AS row
CALL (row) {
  MATCH (a:Person {id: row.fromId})
  MATCH (b:Person {id: row.toId})
  MERGE (a)-[:KNOWS {since: toIntegerOrNull(row.year)}]->(b)
} IN TRANSACTIONS OF 5000 ROWS
  ON ERROR CONTINUE
  REPORT STATUS AS s

リレーションシップをインポートする前に、必ずすべてのノードをインポートしてください — ノードが存在しないと MATCH は失敗します。

タブ区切りまたはカスタム区切り文字

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///data.tsv' AS row FIELDTERMINATOR '\t'
CALL (row) { MERGE (p:Person {id: row.id}) }
IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE

圧縮ファイル（ZIP／gzip — ローカルファイルのみ）

LOAD CSV WITH HEADERS FROM 'file:///archive.csv.gz' AS row ...

クラウドストレージ（Enterprise Edition）

スキーム	例
AWS S3	s3://my-bucket/data/persons.csv
Google Cloud Storage	gs://my-bucket/persons.csv
Azure Blob	azb://account/container/persons.csv

LOAD CSV 内で使用できる便利な組み込み関数

linenumber()   // current line number — use as fallback ID
file()         // absolute path of file being loaded

---

CALL IN TRANSACTIONS — 完全なリファレンス

構文

CALL (row) {
  // write logic
} IN [n CONCURRENT] TRANSACTIONS
  [OF batchSize ROW[S]]
  [ON ERROR {CONTINUE | BREAK | FAIL | RETRY [FOR duration SECONDS] [THEN {CONTINUE|BREAK|FAIL}]}]
  [REPORT STATUS AS statusVar]

ON ERROR モード

モード	動作	使用する場面
ON ERROR FAIL	デフォルトです。最初のエラーで外側のトランザクション全体をロールバックします	すべて成功するか、すべて失敗する厳密なインポート
ON ERROR CONTINUE	失敗したバッチをスキップし、残りのバッチを続行します	エラーを REPORT STATUS で追跡する、堅牢な一括読み込み
ON ERROR BREAK	最初に失敗したバッチの後で停止し、完了済みの処理は保持します	半厳密な処理：早期に停止し、成功したバッチは保持
ON ERROR RETRY	指数バックオフによる再試行（デフォルトは30s）とフォールバック	デッドロックのリスクがある並行書き込み

ON ERROR CONTINUE/BREAK → 内側のバッチが失敗しても、外側のトランザクションは成功します。 ON ERROR FAIL → REPORT STATUS AS と組み合わせることはできません。

CONCURRENT TRANSACTIONS（並列バッチ）

CYPHER 25
LOAD CSV WITH HEADERS FROM 'file:///large.csv' AS row
CALL (row) {
  MERGE (p:Person {id: row.id}) SET p.name = row.name
} IN 4 CONCURRENT TRANSACTIONS OF 5000 ROWS
  ON ERROR RETRY FOR 30 SECONDS THEN CONTINUE
  REPORT STATUS AS s

キー空間が重複しない、読み取り中心の MERGE には CONCURRENT を使用します。リスク：書き込みが重複するとデッドロックが発生する可能性があります → ON ERROR RETRY と組み合わせてください。

REPORT STATUS の列

列	型	意味
s.started	BOOLEAN	バッチトランザクションが開始されたかどうか
s.committed	BOOLEAN	バッチが正常にコミットされたかどうか
s.transactionId	STRING	トランザクション ID
s.errorMessage	STRING または null	バッチ失敗時のエラー詳細

バッチサイズの目安

行数	推奨バッチサイズ	備考
< 100k	10 000	デフォルトで問題ありません
100k – 1M	10 000 – 50 000	ヒープを監視し、高速であれば増やします
1M – 10M	50 000 – 100 000	CPU に余裕があれば CONCURRENT を有効にします
> 10M オンライン	50 000	代わりに neo4j-admin import の使用を検討してください
リレーションシップのインポート	5 000	低めに設定します — 各バッチで 2 回 MATCH を実行します

---

neo4j-admin import（オフライン一括読み込み）

最速の方法です。SSD で 3,100 万ノード／7,800 万リレーションシップを約 3 分で処理します。DB は停止中か、存在しない状態である必要があります。

コマンドの構造

neo4j-admin database import full \
  --nodes=Person="persons_header.csv,persons.csv" \
  --nodes=Movie="movies_header.csv,movies.csv" \
  --relationships=ACTED_IN="acted_in_header.csv,acted_in.csv" \
  --relationships=DIRECTED="directed_header.csv,directed.csv" \
  --delimiter=, \
  --id-type=STRING \
  --bad-tolerance=0 \
  --threads=$(nproc) \
  --high-parallel-io=on \
  neo4j

ドライラン（2026.02+）— 書き込みを行わずに検証します：

neo4j-admin database import full --dry-run ...

ノードヘッダーファイルの形式

# persons_header.csv
personId:ID,name,born:int,score:float,active:boolean,:LABEL

# persons.csv (data file — no header row)
p001,Alice,1985,9.2,true,Person
p002,Bob,1990,7.1,false,Person

フィールド	意味
:ID	リレーションシップの接続に使用する一意の ID（デフォルトではプロパティとして保存されません）
:ID(Group)	スコープ付き ID 空間 — ノードタイプ間で ID を共有する場合に使用します
:LABEL	1 つ以上のラベル。セミコロン区切りです：Person;Employee
prop:int	型付きプロパティ。型：int long float double boolean byte short string
prop:date	時間型：date localtime time localdatetime datetime duration。Parquet の INTERVAL 列は DURATION としてインポートされます [2026.07+]
prop:int[]	配列 — セル内の値はセミコロン区切りです：1;2;3
prop:vector	浮動小数点ベクトル（2025.10+）— CSV では座標をセミコロン区切りで指定します。ネイティブ Parquet リスト型から直接インポートできます [2026.06+]

リレーションシップヘッダーファイルの形式

# acted_in_header.csv
:START_ID(Person),:END_ID(Movie),role,:TYPE

# acted_in.csv
p001,tt0133093,Neo,ACTED_IN
p002,tt0133093,Morpheus,ACTED_IN

:START_ID／:END_ID は、ノードファイルと同じ :ID グループを参照する必要があります。

主なフラグ

フラグ	デフォルト	備考
--delimiter	,	UTF-8 の 1 バイト文字、TAB、\ID、または U+XXXX。改行文字は拒否されます [2026.07+]
--vector-delimiter	;	prop:vector の座標を区切ります。--delimiter および --quote とは異なる値にする必要があります [2026.06+ で適用]
--id-type	STRING	STRING | INTEGER | ACTUAL
--bad-tolerance	-1（無制限、2025.12 に変更）	本番環境で厳密にインポートする場合は 0 に設定します
--threads	CPU 数	共有ホストでは明示的に設定します
--max-off-heap-memory	RAM の 90%	ホスト上で他のサービスも稼働している場合は減らします
--high-parallel-io	off	SSD/NVMe では on に設定します
--format	standard	ノード／リレーションシップが 340 億を超える場合は block を使用します
--overwrite-destination	false	DB がすでに存在する場合に必要です
--dry-run	false	2026.02+ — 書き込みを行わずに検証します

スキーマファイル（--schema）［Enterprise、block形式］

CREATE CONSTRAINT／CREATE INDEX ステートメントを含む Cypher ファイルを渡すと、インポート完了後に自動実行されます。制約が先に作成されます（正しい順序が適用されます）。ファイルパスにはローカルパスまたはリモートパス（s3://、gs://、https://）を指定できます。

neo4j-admin database import full \
  --format=block \
  --schema=schema.cypher \
  --nodes=Person="persons_header.csv,persons.csv" \
  neo4j

// schema.cypher
CREATE CONSTRAINT person_id IF NOT EXISTS FOR (n:Person) REQUIRE n.id IS UNIQUE;
CREATE CONSTRAINT movie_id  IF NOT EXISTS FOR (n:Movie)  REQUIRE n.id IS UNIQUE;
CREATE RANGE INDEX person_email IF NOT EXISTS FOR (n:Person) ON (n.email);
CREATE TEXT  INDEX movie_title  IF NOT EXISTS FOR (n:Movie)  ON (n.title);

増分インポートでは、DROP CONSTRAINT／DROP INDEX もサポートされています［2025.02+］。マージフェーズ前にインデックスを削除し、書き込みを高速化するためにフェーズ後に再作成する用途で使用します。--schema はグラフタイプの DDL も受け付けます。full では ALTER CURRENT GRAPH TYPE SET {…}［2026.05+］、incremental では ALTER CURRENT GRAPH TYPE ADD/DROP/ALTER {…}［2026.06+］を使用します。詳しくは neo4j-cypher-skill/references/graph-type.md を参照してください。

---

増分インポート（Enterprise のみ）

3段階のプロセスです。インポート準備中もデータベースをオンラインに保つ必要がある場合に使用します。

# Phase 1: Prepare staging area
neo4j-admin database import incremental --stage=prepare \
  --nodes=Person=persons_header.csv,delta.csv --force neo4j

# Phase 2: Build indexes (DB can be read-only during this phase)
neo4j-admin database import incremental --stage=build neo4j

# Phase 3: Merge into live database (brief write-lock)
neo4j-admin database import incremental --stage=merge neo4j

Enterprise Edition と block ストア形式が必要です。

---

APOC パターン（APOC が利用可能な場合）

まず RETURN apoc.version() を実行して確認してください。失敗する場合は、代わりに LOAD CSV またはドライバーを使用してください。

apoc.periodic.iterate — 既存のグラフデータをバッチ処理する

CALL apoc.periodic.iterate(
  "MATCH (p:Person) WHERE NOT (p)-[:HAS_ACCOUNT]->() RETURN p",
  "CREATE (p)-[:HAS_ACCOUNT]->(a:Account {id: randomUUID()})",
  {batchSize: 10000, parallel: false, retries: 2}
) YIELD batches, total, errorMessages
RETURN batches, total, errorMessages

設定キー	デフォルト	備考
batchSize	10000	内部トランザクションごとの行数
parallel	false	重複しない書き込みで有効にします。デッドロックのリスクがあります
retries	0	失敗したバッチを100msの間隔でN回再試行します

新しいコードでは、apoc.periodic.iterate よりも CALL IN TRANSACTIONS（ネイティブ Cypher）を優先してください。APOC に依存せず、REPORT STATUS、CONCURRENT、RETRY が組み込まれています。

apoc.load.csv — 設定オプションを指定して読み込む

CALL apoc.load.csv('file:///persons.csv', {
  header: true,
  sep: ',',
  skip: 1,
  limit: 1000000
}) YIELD lineNo, map, list
CALL (map) {
  MERGE (p:Person {id: map.id}) SET p.name = map.name
} IN TRANSACTIONS OF 10000 ROWS ON ERROR CONTINUE

apoc.load.json — ファイルまたは URL から JSON を読み込む

CALL apoc.load.json('https://api.example.com/persons') YIELD value
CALL (value) {
  MERGE (p:Person {id: value.id}) SET p.name = value.name
} IN TRANSACTIONS OF 1000 ROWS ON ERROR CONTINUE

---

ドライバーのバッチ書き込みパターン

ソースがファイルではない場合（API のレスポンス、データベースのマイグレーション）に使用します。データを BATCH_SIZE（10 000）件ずつのリストにまとめ、各バッチで UNWIND $rows AS row MERGE ... を呼び出します。1行ずつ処理する場合より約10倍高速です。→ Python と JS の例

---

MCP ツールの使用方法

操作	MCP ツール	備考
SHOW CONSTRAINTS、SHOW INDEXES	read-cypher	インポート前に必ず確認します
CREATE CONSTRAINT、CREATE INDEX	write-cypher	ゲート：予定している制約を表示し、確認を得ます
LOAD CSV／CALL IN TRANSACTIONS	write-cypher	ゲート：行数と Cypher を表示し、確認を得ます
件数の確認	read-cypher	インポート後：MATCH (n:Label) RETURN count(n)
インデックスの状態をポーリング	read-cypher	すべての state = 'ONLINE' が確認されるまでポーリングします

書き込みゲート — MCP 経由で一括書き込みを行う前に、以下を表示します。

1. クエリと対象ラベル
2. ソースから推定した行数
3. EXPLAIN プラン

ユーザーの確認を待ってください。確認なしで CALL IN TRANSACTIONS や CREATE CONSTRAINT を自動実行しないでください。

デフォルト以外のデータベースを使用する場合は、必ず database パラメーターを渡してください：{"code": "...", "database": "neo4j"}。

---

よくあるエラー

エラー	原因	対処方法
Couldn't load the external resource	file:/// のパスが Neo4j のインポートディレクトリにありません	ファイルを $NEO4J_HOME/import/ に移動します。dbms.security.allow_csv_import_from_file_urls=true になっていることを確認します
Cannot merge node using null property value	MERGE のキーが null として解決されました	MERGE の前に row.id IS NOT NULL を検証します。WHERE row.id IS NOT NULL を追加します
toInteger() called on null	null 許容でない関数に null の列が渡されました	toInteger() → toIntegerOrNull()、toFloat() → toFloatOrNull() に置き換えます
Node N already exists / インポート中の制約違反	ソース ID が重複しています	ソース CSV の重複を除去します。CREATE ではなく MERGE を使用し、制約に IF NOT EXISTS を追加します
ヒープオーバーフロー / OutOfMemoryError	バッチが大きすぎるか、ファイルが大きすぎます	バッチサイズを小さくします。CALL IN TRANSACTIONS に切り替えます。オフラインの場合は neo4j-admin を使用します
Invalid input 'IN': expected...'	PERIODIC COMMIT が使用されています	USING PERIODIC COMMIT を CALL IN TRANSACTIONS に置き換えます。PERIODIC COMMIT は Cypher 25 で削除されています
neo4j-admin: Bad input data	ヘッダー形式が正しくないか、型が一致していません	:ID、:START_ID、:END_ID が存在することを確認します。型付き列が正しく解析されることを確認します
neo4j-admin: インポートがエラーを出さずに失敗する	2025.12 より前は --bad-tolerance のデフォルトが無制限でした	すべてのエラーを表示するには --bad-tolerance=0 を設定します
MERGE 中にインデックスが使用されない	インポート前に制約が作成されていません	データを削除し、制約を作成してから再インポートします
リレーションシップのインポートでノードが見つからない	ノードより先にリレーションシップをインポートしています	リレーションシップファイルをインポートする前に、必ずすべてのノードファイルをインポートします

---

インポート後の検証

インポート完了後、すべて実行します。

// Row counts per label
MATCH (n:Person) RETURN count(n) AS persons;
MATCH ()-[:KNOWS]->() RETURN count(*) AS knows_rels;

// After import: create additional non-unique indexes (populated async)
CREATE TEXT INDEX movie_title IF NOT EXISTS FOR (n:Movie) ON (n.title);
CREATE RANGE INDEX person_born IF NOT EXISTS FOR (n:Person) ON (n.born);

// Poll population — wait until populationPercent = 100 before opening to queries
SHOW INDEXES YIELD name, state, populationPercent
WHERE state <> 'ONLINE' OR populationPercent < 100
RETURN name, state, populationPercent
ORDER BY populationPercent;

// Spot check: null keys = import bug
MATCH (p:Person) WHERE p.id IS NULL RETURN count(p) AS missing_id;

すべてのインデックスが ONLINE になるまで、本番環境のクエリを実行しないでください。

---

参考資料

• LOAD CSV — Cypher マニュアル 25
• CALL IN TRANSACTIONS — Cypher マニュアル
• neo4j-admin データベースインポート
• APOC の定期実行
• APOC のロードプロシージャ
• GraphAcademy：CSV データのインポート
• インデックスと制約 — 種類、MERGE のロックセマンティクス、インポート前の確認事項
• Data Importer GUI — 使用する場面、Aura へのアクセス、複数パス、注意点
• インポート後のリファクタリング — リストの分割、ノードの抽出、ラベルの追加、外部キーの検証

---

チェックリスト

• MERGE ベースのインポートを行う前に、一意性制約を作成した
• apoc.* プロシージャを使用する場合、APOC が利用可能であることを確認した
• ターゲットが PRIMARY（レプリカではない）であることを確認した
• ソースファイルを検証した：UTF-8 エンコーディング、想定される行数、BOM なし
• LOAD CSV で toIntegerOrNull() / toFloatOrNull() を使用している — toInteger() / toFloat() をそのまま使用していない
• 空文字列 ≠ null の箇所に nullIf(row.x, '') を適用した
• USING PERIODIC COMMIT ではなく CALL IN TRANSACTIONS を使用した
• 本番環境でのロードに ON ERROR CONTINUE + REPORT STATUS を使用した
• リレーションシップのインポート前にノードのインポートを完了した
• neo4j-admin：--bad-tolerance=0 を設定した。SSD では --high-parallel-io=on を設定した
• インポート後：行数がソースと一致し、すべてのインデックスが ONLINE である
• 書き込み実行ゲートを適用した（MCP）：クエリと推定値を提示し、確認を得た
• 認証情報を .env に保存し、.env を .gitignore に追加した