JSONLデータベースインポート＆エクスポート：PostgreSQL、MongoDB等

-- Create a staging table with a single jsonb column
CREATE TABLE staging_import (data jsonb);

-- Import the JSONL file (each line becomes one row)
\COPY staging_import (data) FROM 'users.jsonl';

-- Verify the import
SELECT count(*) FROM staging_import;
SELECT data->>'name' AS name, data->>'email' AS email
FROM staging_import
LIMIT 5;

-- Move data into a structured table
INSERT INTO users (name, email, age)
SELECT
  data->>'name',
  data->>'email',
  (data->>'age')::int
FROM staging_import;

-- Export entire table as JSONL
\COPY (
  SELECT row_to_json(t)
  FROM (SELECT id, name, email, created_at FROM users) t
) TO 'users_export.jsonl';

-- Export with filtering and transformation
\COPY (
  SELECT row_to_json(t)
  FROM (
    SELECT id, name, email,
           created_at::text AS created_at
    FROM users
    WHERE active = true
    ORDER BY id
  ) t
) TO 'active_users.jsonl';

-- Verify: each line is valid JSON
-- {"id":1,"name":"Alice","email":"alice@example.com",...}

import json
import psycopg2
from psycopg2.extras import execute_values

def import_jsonl_to_postgres(file_path, conn_string):
    conn = psycopg2.connect(conn_string)
    cur = conn.cursor()
    batch = []
    batch_size = 1000
    total = 0

    with open(file_path, 'r') as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
                batch.append((
                    record['name'],
                    record['email'],
                    record.get('age'),
                ))
            except (json.JSONDecodeError, KeyError) as e:
                print(f"Skipping line {line_num}: {e}")
                continue

            if len(batch) >= batch_size:
                execute_values(
                    cur,
                    "INSERT INTO users (name, email, age) "
                    "VALUES %s ON CONFLICT (email) DO NOTHING",
                    batch
                )
                total += len(batch)
                batch = []
                print(f"Imported {total} records...")

    if batch:
        execute_values(
            cur,
            "INSERT INTO users (name, email, age) "
            "VALUES %s ON CONFLICT (email) DO NOTHING",
            batch
        )
        total += len(batch)

    conn.commit()
    cur.close()
    conn.close()
    print(f"Done. Imported {total} records.")

# Usage
import_jsonl_to_postgres('users.jsonl', 'postgresql://localhost/mydb')

-- Step 1: Create a staging table for raw lines
CREATE TABLE jsonl_staging (
  id INT AUTO_INCREMENT PRIMARY KEY,
  raw_line TEXT NOT NULL
);

-- Step 2: Load the JSONL file as raw text
LOAD DATA INFILE '/var/lib/mysql-files/users.jsonl'
INTO TABLE jsonl_staging
LINES TERMINATED BY '\n'
(raw_line);

-- Step 3: Extract structured data with JSON_TABLE
INSERT INTO users (name, email, age)
SELECT jt.name, jt.email, jt.age
FROM jsonl_staging,
  JSON_TABLE(
    raw_line, '$'
    COLUMNS (
      name VARCHAR(255) PATH '$.name',
      email VARCHAR(255) PATH '$.email',
      age INT PATH '$.age'
    )
  ) AS jt;

-- Step 4: Clean up staging table
DROP TABLE jsonl_staging;

-- Verify the import
SELECT * FROM users LIMIT 5;

import json
import mysql.connector

def import_jsonl_to_mysql(file_path, config):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    batch = []
    batch_size = 1000
    total = 0

    insert_sql = (
        "INSERT INTO users (name, email, age) "
        "VALUES (%s, %s, %s) "
        "ON DUPLICATE KEY UPDATE name=VALUES(name)"
    )

    with open(file_path, 'r') as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
                batch.append((
                    record['name'],
                    record['email'],
                    record.get('age'),
                ))
            except (json.JSONDecodeError, KeyError) as e:
                print(f"Skipping line {line_num}: {e}")
                continue

            if len(batch) >= batch_size:
                cursor.executemany(insert_sql, batch)
                conn.commit()
                total += len(batch)
                batch = []
                print(f"Imported {total} records...")

    if batch:
        cursor.executemany(insert_sql, batch)
        conn.commit()
        total += len(batch)

    cursor.close()
    conn.close()
    print(f"Done. Imported {total} records.")

# Usage
import_jsonl_to_mysql('users.jsonl', {
    'host': 'localhost',
    'user': 'root',
    'password': 'secret',
    'database': 'mydb'
})