Οδηγός γρήγορης εκκίνησης: Μαζική αντιγραφή δεδομένων μεταξύ βάσεων δεδομένων SQL στο Fabric με το πρόγραμμα οδήγησης mssql-python

Σε αυτήν τη γρήγορη εκκίνηση, μπορείτε να χρησιμοποιήσετε το mssql-python πρόγραμμα οδήγησης για να αντιγράψετε μαζικά δεδομένα μεταξύ βάσεων δεδομένων SQL στο Fabric. Η εφαρμογή κατεβάζει πίνακες από ένα σχήμα βάσης δεδομένων προέλευσης σε τοπικά αρχεία Parquet χρησιμοποιώντας το Apache Arrow και, στη συνέχεια, τους ανεβάζει σε μια βάση δεδομένων προορισμού χρησιμοποιώντας τη μέθοδο υψηλής απόδοσης bulkcopy . Μπορείτε να χρησιμοποιήσετε αυτό το μοτίβο για μετεγκατάσταση, αναπαραγωγή ή μετασχηματισμό δεδομένων μεταξύ βάσεων δεδομένων SQL στο Fabric.

Το mssql-python πρόγραμμα οδήγησης δεν απαιτεί εξωτερικές εξαρτήσεις σε μηχανήματα Windows. Το πρόγραμμα οδήγησης εγκαθιστά όλα όσα χρειάζεται με μία μόνο pip εγκατάσταση, επιτρέποντάς σας να χρησιμοποιήσετε την πιο πρόσφατη έκδοση του προγράμματος οδήγησης για νέα σενάρια χωρίς να σπάσετε άλλα σενάρια που δεν έχετε χρόνο να αναβαθμίσετε και να δοκιμάσετε.

Τεκμηρίωση | Πηγαίος κώδικας | Πακέτο (PyPI) | UV

Προϋποθέσεις

Φόρτωση δείγματος δεδομένων AdventureWorks στη βάση δεδομένων SQL ως βάση δεδομένων προέλευσης.
(Προαιρετικός) Μια δεύτερη βάση δεδομένων SQL στο Fabric που θα χρησιμοποιηθεί ως προορισμός. Ο χρήστης πρέπει να έχει δικαίωμα δημιουργίας και εγγραφής σε πίνακες. Εάν δεν έχετε δεύτερη βάση δεδομένων, μπορείτε να αλλάξετε τη συμβολοσειρά σύνδεσης προορισμού ώστε να οδηγεί στην ίδια βάση δεδομένων και να χρησιμοποιήσετε διαφορετικό σχήμα για τους πίνακες προορισμού.
Πύθωνας 3
- Εάν δεν έχετε ήδη Python, εγκαταστήστε τον χρόνο εκτέλεσης Python και τη διαχείριση πακέτων pip από python.org.
- Δεν θέλετε να χρησιμοποιήσετε το δικό σας περιβάλλον; Ανοίξτε ως devcontainer χρησιμοποιώντας χώρους κώδικα GitHub.
Κώδικας του Visual Studio με τις ακόλουθες επεκτάσεις:
- Επέκταση Python για κώδικα Visual Studio
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

Δημιουργήστε το έργο και εκτελέστε τον κώδικα

Δημιουργήστε ένα νέο έργο
Προσθήκη εξαρτήσεων
Εκκινήστε τον κώδικα του Visual Studio
Ενημέρωση pyproject.toml
Ενημέρωση main.py
Αποθήκευση των συμβολοσειρών σύνδεσης
Χρησιμοποιήστε το uv run για να εκτελέσετε το σενάριο

Δημιουργία νέου έργου

Ανοίξτε μια γραμμή εντολών στον κατάλογο ανάπτυξης. Εάν δεν έχετε, δημιουργήστε έναν νέο κατάλογο που ονομάζεται 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 .

Ενημέρωση 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 αντιμετωπίζει προβλήματα με την επίλυση πακέτων, πρέπει να ενημερώσετε τον διερμηνέα για να χρησιμοποιήσετε το εικονικό περιβάλλον.
Στο επάνω μέρος του main.py, προσθέστε τις εισαγωγές και τις σταθερές. Το σενάριο χρησιμοποιείται mssql_python για συνδεσιμότητα pyarrow βάσης δεδομένων και pyarrow.parquet για χειρισμό δεδομένων στηλών και I/O αρχείου Parquet, python-dotenv για φόρτωση συμβολοσειρών σύνδεσης από ένα .env αρχείο και ένα μεταγλωττισμένο μοτίβο regex που επικυρώνει αναγνωριστικά 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, έτσι ώστε να διατηρείται η πιστότητα των δεδομένων κατά την εγγραφή σε 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 ερωτήματα χρησιμοποιώντας παραμετροποιημένα ερωτήματα, δημιουργεί ένα σχήμα βέλους και αποθηκεύει τον αρχικό τύπο 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 συμβατό με βέλος και εγγράφει τις παρτίδες εγγραφών σταδιακά σε ένα αρχείο 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 ταχύτητα μετάδοσης ελαχιστοποιώντας τη διένεξη κλειδώματος. Μετά την ολοκλήρωση της μεταφόρτωσης, η συνάρτηση εκτελεί ένα 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το , το οποίο επαναχρησιμοποιεί τα διαπιστευτήρια στο cache.

Χρησιμοποιήστε το 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 και, στη συνέχεια, κατεβάζει κάθε πίνακα σε παρτίδες σε ένα τοπικό αρχείο Parquet χρησιμοποιώντας pq.ParquetWriterτο .
Εμπλουτισμός (προαιρετικό): Παρέχει ένα άγκιστρο (enrich_parquet) όπου μπορείτε να προσθέσετε μετασχηματισμούς, παράγωγες στήλες ή συνδέσμους πριν από τη μεταφόρτωση.
Μεταφόρτωση: Διαβάζει κάθε αρχείο Parquet σε παρτίδες, αναδημιουργεί τον πίνακα στη βάση δεδομένων προορισμού χρησιμοποιώντας DDL που δημιουργείται από μεταδεδομένα σχήματος βέλους και, στη συνέχεια, χρησιμοποιεί cursor.bulkcopy() για μαζική εισαγωγή υψηλής απόδοσης.

Επόμενο βήμα

Επισκεφτείτε το αποθετήριο GitHub του mssql-python προγράμματος οδήγησης για περισσότερα παραδείγματα, για να συνεισφέρετε ιδέες ή να αναφέρετε προβλήματα.

πρόγραμμα οδήγησης mssql-python στο GitHub

Σχόλια

Ήταν χρήσιμη αυτή η σελίδα;

Last updated on 2026-03-18