เริ่มต้นใช้งานด่วน: คัดลอกข้อมูลจํานวนมากระหว่างฐานข้อมูล SQL ใน Fabric ด้วยไดรเวอร์ mssql-python

ในการเริ่มต้นใช้งานด่วนนี้ คุณใช้โปรแกรม mssql-python ควบคุมเพื่อคัดลอกข้อมูลจํานวนมากระหว่างฐานข้อมูล SQL ใน Fabric แอปพลิเคชันจะดาวน์โหลดตารางจากสคีมาฐานข้อมูลต้นทางไปยังไฟล์ Parquet ในเครื่องโดยใช้ Apache Arrow จากนั้นอัปโหลดไปยังฐานข้อมูลปลายทางโดยใช้วิธีการที่มีประสิทธิภาพ bulkcopy สูง คุณสามารถใช้รูปแบบนี้เพื่อโยกย้าย จําลองแบบ หรือแปลงข้อมูลระหว่างฐานข้อมูล SQL ใน Fabric

mssql-pythonไดรเวอร์ไม่ต้องการการพึ่งพาภายนอกใดๆ บนเครื่อง Windows ไดรเวอร์จะติดตั้งทุกอย่างที่จําเป็นด้วยการติดตั้งเพียงครั้งเดียว pip ช่วยให้คุณใช้ไดรเวอร์เวอร์ชันล่าสุดสําหรับสคริปต์ใหม่ได้โดยไม่ทําลายสคริปต์อื่นๆ ที่คุณไม่มีเวลาอัปเกรดและทดสอบ

เอกสารประกอบ | ซอร์สโค้ด | mssql-pythonแพ็คเกจ (PyPI) | ยูวี

ข้อกำหนดเบื้องต้น

โหลดข้อมูลตัวอย่าง AdventureWorks ในฐานข้อมูล SQL ของคุณ เป็นฐานข้อมูลต้นทาง
(ไม่บังคับ) ฐานข้อมูล SQL ที่สองใน Fabric เพื่อใช้เป็นปลายทาง ผู้ใช้ต้องมีสิทธิ์ในการสร้างและเขียนลงในตาราง ถ้าคุณไม่มีฐานข้อมูลที่สอง คุณสามารถเปลี่ยนสตริงการเชื่อมต่อปลายทางให้ชี้ไปยังฐานข้อมูลเดียวกัน และใช้ Schema อื่นสําหรับตารางปลายทางได้
หลาม 3
- หากคุณยังไม่มี Python ให้ติดตั้งรันไทม์ Python และตัวจัดการแพ็คเกจ pip จาก python.org
- ไม่ต้องการใช้สภาพแวดล้อมของคุณเองใช่ไหม เปิดเป็น devcontainer โดยใช้ GitHub Codespaces
Visual Studio Code ที่มีส่วนขยายต่อไปนี้:
- ส่วนขยาย Python สําหรับ Visual Studio Code
Azure Command-Line Interface (CLI) - จําเป็นสําหรับการรับรองความถูกต้องแบบไม่ใช้รหัสผ่านบน macOS และ Linux
หากคุณยังไม่มี uvให้ติดตั้ง uv โดยทําตามคําแนะนําจาก https://docs.astral.sh/uv/getting-started/installation/.
ติดตั้งข้อกําหนดเบื้องต้นเฉพาะระบบปฏิบัติการแบบใช้ครั้งเดียว
```
apk add libtool krb5-libs krb5-dev
```
```
apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2
```
```
dnf install -y libtool-ltdl krb5-libs
```
```
zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2
```
```
zypper install -y libltdl7
```
```
brew install openssl
```

สร้างโครงการใหม่

เปิดพรอมต์คําสั่งในไดเรกทอรีการพัฒนาของคุณ หากคุณไม่มี ให้สร้างไดเร็กทอรีใหม่ที่เรียกว่า python, scriptsฯลฯหลีกเลี่ยงโฟลเดอร์บน OneDrive ของคุณการซิงโครไนซ์อาจรบกวนการจัดการสภาพแวดล้อมเสมือนของคุณ
สร้างโปรเจ็กต์ใหม่ด้วยuv
```
uv init mssql-python-bcp-qs
cd mssql-python-bcp-qs
```

เพิ่มการขึ้นต่อกัน

ในไดเร็กทอรีเดียวกัน ให้ติดตั้งmssql-pythonแพคเกจ , python-dotenvและpyarrow

uv add mssql-python python-dotenv pyarrow

เปิดใช้ Visual Studio Code

ในไดเร็กทอรีเดียวกัน ให้เรียกใช้คําสั่งต่อไปนี้

code .

อัปเดต pyproject.toml

pyproject.toml มีข้อมูลเมตาสําหรับโครงการของคุณ เปิดไฟล์ในโปรแกรมแก้ไขที่คุณชื่นชอบ
ตรวจสอบเนื้อหาของไฟล์ ควรคล้ายกับตัวอย่างนี้ สังเกตเวอร์ชัน Python และการขึ้นต่อกันสําหรับการใช้งานmssql-python>=เพื่อกําหนดเวอร์ชันขั้นต่ํา หากคุณต้องการเวอร์ชันที่แน่นอน ให้เปลี่ยนหมายเลข>=ก่อนเวอร์ชันเป็น== เวอร์ชันที่แก้ไขแล้วของแต่ละแพ็คเกจจะถูกเก็บไว้ใน uv.lock ไฟล์ล็อคช่วยให้แน่ใจว่านักพัฒนาที่ทํางานในโครงการใช้เวอร์ชันแพ็คเกจที่สอดคล้องกัน นอกจากนี้ยังช่วยให้แน่ใจว่ามีการใช้เวอร์ชันแพ็คเกจชุดเดียวกันทุกประการเมื่อแจกจ่ายแพ็คเกจของคุณให้กับผู้ใช้ปลายทาง คุณไม่ควรแก้ไข uv.lock ไฟล์
```
[project]
name = "mssql-python-bcp-qs"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "mssql-python>=1.4.0",
    "python-dotenv>=1.1.1",
    "pyarrow>=19.0.0",
]
```
อัปเดตคําอธิบายให้สื่อความหมายมากขึ้น
```
description = "Bulk copies data between SQL databases using mssql-python and Apache Arrow"
```
บันทึกและปิดไฟล์

อัปเดต main.py

เปิดไฟล์ชื่อ main.py. ควรคล้ายกับตัวอย่างนี้
```
def main():
    print("Hello from mssql-python-bcp-qs!")

if __name__ == "__main__":
    main()
```
แทนที่เนื้อหาของแฟ้มด้วยบล็อกโค้ดต่อไปนี้ แต่ละบล็อกสร้างขึ้นจากบล็อกก่อนหน้าและควรเพิ่มตามลําดับmain.py

เคล็ดลับ

ถ้า Visual Studio Code มีปัญหาในการแก้ไขแพคเกจ คุณจําเป็นต้องปรับปรุงล่ามเพื่อใช้สภาพแวดล้อมเสมือน
ที่ด้านบนของ main.pyเพิ่มการนําเข้าและค่าคงที่ สคริปต์ใช้สําหรับmssql_pythonการเชื่อมต่อฐานข้อมูล และpyarrowสําหรับการจัดการข้อมูลแบบคอลัมน์และ I/O pyarrow.parquet ของไฟล์ Parquet สําหรับการโหลดสตริงการเชื่อมต่อจากpython-dotenvไฟล์ และรูปแบบนิพจน์ทั่วไปที่คอมไพล์แล้วซึ่งตรวจสอบความถูกต้องของตัวระบุ .env SQL เพื่อป้องกันการแทรก
```
"""Round-trip: download tables from a source DB/schema to parquet, upload to a destination DB/schema."""

import os
import re
import time
from uuid import UUID

import pyarrow as pa
import pyarrow.parquet as pq
from dotenv import load_dotenv
import mssql_python

BATCH_SIZE = 64_000
_SAFE_IDENT = re.compile(r"^[A-Za-z0-9_]+$")


def _validate_ident(name: str) -> str:
    if not _SAFE_IDENT.match(name):
        raise ValueError(f"Unsafe SQL identifier: {name!r}")
    return name
```

ด้านล่างการนําเข้า ให้เพิ่มการแมปชนิด SQL กับลูกศร พจนานุกรมนี้แปลชนิดคอลัมน์ SQL Server เป็นคอลัมน์ที่เทียบเท่ากับ Apache Arrow เพื่อให้ความเที่ยงตรงของข้อมูลถูกเก็บรักษาไว้เมื่อเขียนไปยัง Parquet ฟังก์ชันตัวช่วยทั้งสองสร้างสตริงชนิด SQL ที่แน่นอน (ตัวอย่างเช่น NVARCHAR(100) หรือ DECIMAL(18,2)) จาก INFORMATION_SCHEMA เมตาดาต้า และแก้ไขชนิดลูกศรที่ตรงกันสําหรับแต่ละคอลัมน์

_SQL_TO_ARROW = {
    "bit": pa.bool_(),
    "tinyint": pa.uint8(),
    "smallint": pa.int16(),
    "int": pa.int32(),
    "bigint": pa.int64(),
    "float": pa.float64(),
    "real": pa.float32(),
    "smallmoney": pa.decimal128(10, 4),
    "money": pa.decimal128(19, 4),
    "date": pa.date32(),
    "datetime": pa.timestamp("ms"),
    "datetime2": pa.timestamp("us"),
    "smalldatetime": pa.timestamp("s"),
    "uniqueidentifier": pa.string(),
    "xml": pa.string(),
    "image": pa.binary(),
    "binary": pa.binary(),
    "varbinary": pa.binary(),
    "timestamp": pa.binary(),
}


def _sql_type_str(data_type: str, max_length: int, precision: int, scale: int) -> str:
    """Build the exact SQL type string from INFORMATION_SCHEMA metadata."""
    dt = data_type.lower()
    if dt in ("char", "varchar", "nchar", "nvarchar", "binary", "varbinary"):
        length = "MAX" if max_length == -1 else str(max_length)
        return f"{dt.upper()}({length})"
    if dt in ("decimal", "numeric"):
        return f"{dt.upper()}({precision},{scale})"
    return dt.upper()


def _arrow_type(sql_type: str, precision: int, scale: int) -> pa.DataType:
    sql_type = sql_type.lower()
    if sql_type in _SQL_TO_ARROW:
        return _SQL_TO_ARROW[sql_type]
    if sql_type in ("decimal", "numeric"):
        return pa.decimal128(precision, scale)
    if sql_type in ("char", "varchar", "nchar", "nvarchar", "text", "ntext", "sysname"):
        return pa.string()
    return pa.string()


def _convert_value(v):
    """Convert a SQL value to an Arrow-compatible Python type."""
    if isinstance(v, UUID):
        return str(v)
    return v

เพิ่มฟังก์ชันการวิปัสสนาสคีมาและการสร้าง DDL _get_arrow_schema สร้าง INFORMATION_SCHEMA.COLUMNS Schema ลูกศร และจัดเก็บชนิด SQL ดั้งเดิมเป็นข้อมูลเมตาของฟิลด์ เพื่อให้สามารถสร้างตารางปลายทางใหม่ด้วยข้อกําหนดคอลัมน์ที่แน่นอน _create_table_ddl อ่านข้อมูลเมตานั้นกลับเพื่อสร้าง DROP/CREATE TABLE DDL timestamp ชนิด (rowversion) ถูกแมปใหม่เนื่องจากVARBINARY(8)สร้างขึ้นโดยอัตโนมัติและไม่สามารถแทรกได้โดยตรง

def _get_arrow_schema(cursor, schema_name: str, table_name: str) -> pa.Schema:
    """Build an Arrow schema from INFORMATION_SCHEMA.COLUMNS.

    Stores the original SQL type as field metadata so the round-trip
    CREATE TABLE can reproduce exact column definitions.
    """
    cursor.execute(
        "SELECT COLUMN_NAME, DATA_TYPE, "
        "COALESCE(CHARACTER_MAXIMUM_LENGTH, 0), "
        "COALESCE(NUMERIC_PRECISION, 0), "
        "COALESCE(NUMERIC_SCALE, 0), "
        "IS_NULLABLE "
        "FROM INFORMATION_SCHEMA.COLUMNS "
        "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? "
        "ORDER BY ORDINAL_POSITION",
        (schema_name, table_name),
    )
    rows = cursor.fetchall()
    if not rows:
        raise ValueError(f"No columns found for {schema_name}.{table_name}")
    fields = []
    for col_name, data_type, max_len, precision, scale, nullable in rows:
        arrow_t = _arrow_type(data_type, precision, scale)
        sql_t = _sql_type_str(data_type, max_len, precision, scale)
        fields.append(
            pa.field(
                col_name, arrow_t,
                nullable=(nullable == "YES"),
                metadata={"sql_type": sql_t},
            )
        )
    return pa.schema(fields)


def _create_table_ddl(target: str, schema: pa.Schema) -> str:
    """Build DROP/CREATE TABLE DDL from Arrow schema with SQL type metadata."""
    col_defs = []
    for f in schema:
        sql_t = f.metadata[b"sql_type"].decode()
        # timestamp/rowversion is auto-generated and not insertable
        if sql_t == "TIMESTAMP":
            sql_t = "VARBINARY(8)"
        null = "" if f.nullable else " NOT NULL"
        col_defs.append(f"[{f.name}] {sql_t}{null}")
    col_defs_str = ",\n    ".join(col_defs)
    return (
        f"IF OBJECT_ID('{target}', 'U') IS NOT NULL DROP TABLE {target};\n"
        f"CREATE TABLE {target} (\n    {col_defs_str}\n);"
    )

เพิ่มฟังก์ชันดาวน์โหลด download_tableอ่านแถวจากตารางต้นฉบับเป็นชุดfetchmanyโดยใช้ แปลงแต่ละค่าเป็นประเภท Python ที่เข้ากันได้กับ Arrow และเขียนแบทช์บันทึกทีละน้อยไปยังไฟล์ Parquet ด้วยpq.ParquetWriter วิธีนี้หลีกเลี่ยงการโหลดทั้งตารางลงในหน่วยความจํา ฟังก์ชันนี้ใช้เคอร์เซอร์สองตัวแยกกัน: เคอร์เซอร์หนึ่งเพื่ออ่านข้อมูลเมตาของคอลัมน์ และอีกเคอร์เซอร์หนึ่งเพื่อสตรีมข้อมูล หากตารางว่างเปล่า ตารางจะกลับมาก่อนกําหนด

def download_table(conn, schema_name: str, table_name: str, parquet_file: str) -> int:
    """Download a SQL table to a parquet file. Returns row count (0 if empty)."""
    _validate_ident(table_name)
    source = f"{schema_name}.[{table_name}]"

    with conn.cursor() as cursor:
        schema = _get_arrow_schema(cursor, schema_name, table_name)

    n_cols = len(schema)
    row_count = 0
    t0 = time.perf_counter()

    with conn.cursor() as cursor:
        cursor.execute(f"SELECT * FROM {source}")
        with pq.ParquetWriter(parquet_file, schema) as writer:
            while True:
                rows = cursor.fetchmany(BATCH_SIZE)
                if not rows:
                    break
                columns = [[] for _ in range(n_cols)]
                for row in rows:
                    for i in range(n_cols):
                        columns[i].append(_convert_value(row[i]))
                arrays = [
                    pa.array(columns[i], type=schema.field(i).type)
                    for i in range(n_cols)
                ]
                batch = pa.record_batch(arrays, schema=schema)
                writer.write_batch(batch)
                row_count += len(rows)

    if row_count == 0:
        os.remove(parquet_file)
        return 0

    elapsed = time.perf_counter() - t0
    rate = f"{int(row_count / elapsed):,} rows/sec" if elapsed > 0 else "n/a"
    print(
        f"{schema_name}.{table_name} → {parquet_file}: {row_count:,} rows downloaded "
        f"in {elapsed:.2f}s ({rate})"
    )
    return row_count

เพิ่มตะขอเพิ่มความสมบูรณ์ enrich_parquet เป็นตัวยึดตําแหน่งที่คุณสามารถเพิ่มการแปลง คอลัมน์ที่ได้รับ หรือการรวมเข้ากับข้อมูลก่อนที่จะอัปโหลด ในการเริ่มต้นใช้งานด่วนนี้ จะเป็น no-op ที่ส่งคืนเส้นทางไฟล์ที่ไม่เปลี่ยนแปลง
```
def enrich_parquet(parquet_file: str) -> str:
    """Enrich a parquet file before upload. Returns the (possibly new) file path."""
    # TODO: add transformations, derived columns, joins, etc.
    print(f"Enriching {parquet_file} (no-op)")
    return parquet_file
```

เพิ่มฟังก์ชันการอัปโหลด upload_parquet อ่านสคีมาลูกศรจากไฟล์ Parquet สร้างและดําเนินการ DROP/CREATE TABLE DDL เพื่อเตรียมปลายทาง จากนั้นอ่านไฟล์เป็นชุดและเรียกใช้ cursor.bulkcopy() การแทรกจํานวนมากที่มีประสิทธิภาพสูง ตัวเลือกนี้ table_lock=True ปรับปรุงปริมาณงานโดยลดการแย่งชิงการล็อคให้เหลือน้อยที่สุด หลังจากการอัปโหลดเสร็จสมบูรณ์ ฟังก์ชันจะเรียกใช้ a SELECT COUNT(*) เพื่อตรวจสอบว่าจํานวนแถวตรงกัน

def upload_parquet(conn, parquet_file: str, target: str) -> int:
    """Upload a parquet file into a SQL table via BCP. Returns row count."""
    # ── Create target table from parquet schema ──
    pf_schema = pq.read_schema(parquet_file)
    with conn.cursor() as cursor:
        cursor.execute(_create_table_ddl(target, pf_schema))
    conn.commit()

    # ── Bulk insert ──
    uploaded = 0
    t0 = time.perf_counter()
    with pq.ParquetFile(parquet_file) as pf:
        with conn.cursor() as cursor:
            for batch in pf.iter_batches(batch_size=BATCH_SIZE):
                rows = zip(*(col.to_pylist() for col in batch.columns))
                cursor.bulkcopy(
                    target, rows, batch_size=BATCH_SIZE,
                    table_lock=True, timeout=3600,
                )
                uploaded += batch.num_rows
    conn.commit()
    elapsed = time.perf_counter() - t0

    # ── Verify ──
    with conn.cursor() as cursor:
        cursor.execute(f"SELECT COUNT(*) FROM {target}")
        count = cursor.fetchone()[0]

    rate = f"{int(uploaded / elapsed):,} rows/sec" if elapsed > 0 else "n/a"
    print(
        f"{parquet_file} → {target}: {uploaded:,} rows uploaded "
        f"in {elapsed:.2f}s ({rate}) "
        f"| verified: {count:,}"
    )
    return uploaded

เพิ่มฟังก์ชันการประสานรวม transfer_tables เชื่อมโยงทั้งสามขั้นตอนเข้าด้วยกัน มันเชื่อมต่อกับฐานข้อมูลต้นทางค้นหาตารางฐานทั้งหมดในสคีมาที่กําหนดผ่าน INFORMATION_SCHEMA.TABLESดาวน์โหลดแต่ละตารางไปยังไฟล์ Parquet ในเครื่องเรียกใช้ตะขอเพิ่มคุณค่าจากนั้นเชื่อมต่อกับฐานข้อมูลปลายทางและอัปโหลดแต่ละไฟล์

def transfer_tables(
    source_conn_str: str,
    dest_conn_str: str,
    source_schema: str,
    dest_schema: str,
) -> None:
    """Download all tables from source DB/schema to parquet, upload to dest DB/schema."""
    _validate_ident(source_schema)
    _validate_ident(dest_schema)

    parquet_dir = source_schema
    os.makedirs(parquet_dir, exist_ok=True)

    # ── Download from source ──
    with mssql_python.connect(source_conn_str) as src_conn:
        with src_conn.cursor() as cursor:
            cursor.execute(
                "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES "
                "WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE' "
                "ORDER BY TABLE_NAME",
                (source_schema,),
            )
            tables = [row[0] for row in cursor.fetchall()]

        print(f"Found {len(tables)} {source_schema} tables: {', '.join(tables)}\n")

        parquet_files = []
        for table_name in tables:
            parquet_file = os.path.join(parquet_dir, f"{table_name}.parquet")
            row_count = download_table(src_conn, source_schema, table_name, parquet_file)
            if row_count == 0:
                print(f"{source_schema}.{table_name}: empty, skipping")
            else:
                parquet_files.append((table_name, parquet_file))

    # ── Enrich parquet files ──
    enriched_files = []
    for table_name, parquet_file in parquet_files:
        enriched_file = enrich_parquet(parquet_file)
        enriched_files.append((table_name, enriched_file))

    # ── Upload to destination ──
    with mssql_python.connect(dest_conn_str) as dest_conn:
        for table_name, parquet_file in enriched_files:
            target = f"{dest_schema}.[{table_name}]"
            upload_parquet(dest_conn, parquet_file, target)

สุดท้าย เพิ่ม main จุดเริ่มต้น โหลด .env ไฟล์ เรียก transfer_tables ด้วยสตริงการเชื่อมต่อต้นทางและปลายทาง และพิมพ์เวลาที่ผ่านไปทั้งหมด

def main():
    load_dotenv()
    t_start = time.perf_counter()

    transfer_tables(
        source_conn_str=os.environ["SOURCE_CONNECTION_STRING"],
        dest_conn_str=os.environ["DEST_CONNECTION_STRING"],
        source_schema="SalesLT",
        dest_schema="dbo",
    )

    print(f"Total: {time.perf_counter() - t_start:.2f}s")


if __name__ == "__main__":
    main()

บันทึกและปิดmain.py

บันทึกสตริงการเชื่อมต่อ

เปิด .gitignore ไฟล์และเพิ่มการยกเว้นสําหรับ .env ไฟล์ ไฟล์ของคุณควรคล้ายกับตัวอย่างนี้ อย่าลืมบันทึกและปิดเมื่อเสร็จแล้ว
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env
```
ในไดเร็กทอรีปัจจุบัน ให้สร้างไฟล์ใหม่ชื่อ .env.
ภายใน .env ไฟล์ ให้เพิ่มรายการสําหรับสตริงการเชื่อมต่อต้นทางและปลายทางของคุณ แทนที่ค่าตัวยึดด้วยชื่อเซิร์ฟเวอร์และฐานข้อมูลจริงของคุณ
```
SOURCE_CONNECTION_STRING="Server=<source_server_name>;Database=<source_database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
DEST_CONNECTION_STRING="Server=<dest_server_name>;Database=<dest_database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
```
เคล็ดลับ

สําหรับฐานข้อมูล SQL ใน Fabric ให้ใช้สตริงการเชื่อมต่อ ODBC จากแท็บ สตริงการเชื่อมต่อโดยไม่มีข้อมูล DRIVER สําหรับข้อมูลเพิ่มเติม โปรดดู เชื่อมต่อกับฐานข้อมูล SQL ของคุณใน Fabric

เคล็ดลับ

บน macOS ทั้งสองอย่าง ActiveDirectoryInteractive และ ActiveDirectoryDefault ใช้ได้กับการรับรองความถูกต้องของ Microsoft Entra ActiveDirectoryInteractive พร้อมท์ให้คุณลงชื่อเข้าใช้ทุกครั้งที่คุณเรียกใช้สคริปต์ เพื่อหลีกเลี่ยงการแจ้งการลงชื่อเข้าใช้ซ้ําๆ ให้เข้าสู่ระบบหนึ่งครั้งผ่าน Azure CLI โดยเรียกใช้ az loginจากนั้นใช้ ActiveDirectoryDefaultซึ่งใช้ข้อมูลประจําตัวที่แคชไว้ซ้ํา

ใช้ uv run เพื่อเรียกใช้สคริปต์

ในหน้าต่างเทอร์มินัลจากก่อนหน้านี้ หรือหน้าต่างเทอร์มินัลใหม่ที่เปิดไปยังไดเร็กทอรีเดียวกัน ให้เรียกใช้คําสั่งต่อไปนี้
```
 uv run main.py
```
นี่คือผลลัพธ์ที่คาดไว้เมื่อสคริปต์เสร็จสมบูรณ์
```
Found 12 SalesLT tables: Address, Customer, CustomerAddress, ...

SalesLT.Address → SalesLT/Address.parquet: 450 rows downloaded in 0.15s (3,000 rows/sec)
...
SalesLT/Address.parquet → dbo.[Address]: 450 rows uploaded in 0.10s (4,500 rows/sec) | verified: 450
...
Total: 2.35s
```
เชื่อมต่อกับฐานข้อมูลปลายทางและตรวจสอบว่าตารางและข้อมูลถูกสร้างขึ้นเรียบร้อยแล้ว สําหรับตัวเลือกเพิ่มเติมเกี่ยวกับวิธีการเชื่อมต่อ โปรดดู เชื่อมต่อกับฐานข้อมูล SQL ของคุณใน Fabric
หากต้องการปรับใช้สคริปต์ของคุณไปยังเครื่องอื่น ให้คัดลอกไฟล์ทั้งหมดยกเว้นโฟลเดอร์ไปยัง .venv เครื่องอื่น สภาพแวดล้อมเสมือนถูกสร้างขึ้นใหม่ด้วยการเรียกใช้ครั้งแรก

วิธีการทํางานของโค้ด

แอปพลิเคชันจะทําการถ่ายโอนข้อมูลไป-กลับเต็มรูปแบบในสามขั้นตอน:

ดาวน์โหลด: เชื่อมต่อกับฐานข้อมูลต้นทาง อ่านข้อมูลเมตาของคอลัมน์จาก INFORMATION_SCHEMA.COLUMNSสร้างสคีมา Apache Arrow จากนั้นดาวน์โหลดแต่ละตารางเป็นชุดไปยังไฟล์ Parquet ในเครื่องโดยใช้pq.ParquetWriter
เพิ่มความสมบูรณ์ (ไม่บังคับ): มี hook (enrich_parquet) ที่คุณสามารถเพิ่มการแปลง คอลัมน์ที่ได้รับ หรือการรวมก่อนอัปโหลด
อัปโหลด: อ่านไฟล์ Parquet แต่ละไฟล์เป็นชุด สร้างตารางใหม่ในฐานข้อมูลปลายทางโดยใช้ DDL ที่สร้างจากข้อมูลเมตาของสคีมา Arrow จากนั้นใช้ cursor.bulkcopy() สําหรับการแทรกจํานวนมากที่มีประสิทธิภาพสูง

ขั้นตอนต่อไป

เยี่ยมชมที่เก็บ GitHub ของ mssql-python ไดรเวอร์เพื่อดูตัวอย่างเพิ่มเติม เพื่อแสดงความคิดเห็นหรือรายงานปัญหา

ไดรเวอร์ mssql-python บน GitHub

คำติชม

หน้านี้มีประโยชน์หรือไม่

Last updated on 2026-03-18