Schnellstart: Massenkopie mit dem mssql-python-Treiber für Python

In dieser Schnellstartanleitung verwenden Sie den mssql-python Treiber zum Massenkopien von Daten zwischen Datenbanken. Die Anwendung lädt Tabellen aus einem Quelldatenbankschema in lokale Parkettdateien mithilfe von Apache Arrow herunter und lädt sie dann mithilfe der hochleistungsorientierten bulkcopy Methode in eine Zieldatenbank hoch. Sie können dieses Muster verwenden, um Daten zwischen SQL Server, Azure SQL-Datenbank und SQL-Datenbank in Fabric zu migrieren, zu replizieren oder zu transformieren.

Der mssql-python Treiber erfordert keine externen Abhängigkeiten von Windows-Computern. Der Treiber installiert alles, was er mit einer einzigen pip Installation benötigt, sodass Sie die neueste Version des Treibers für neue Skripts verwenden können, ohne andere Skripts zu unterbrechen, die Sie nicht zum Upgrade und Test benötigen.

mssql-python-Dokumentation | mssql-python-Quellcode | Paket (PyPI) | Uv

Voraussetzungen

  • Python 3.10 oder höher

  • Wenn Sie noch nicht über Python verfügen, installieren Sie den Python-Runtime - und Pip-Paket-Manager aus python.org.

  • Möchten Sie Ihre eigene Umgebung nicht verwenden? Unter Container- und lokale Entwicklung erfahren Sie, wie Sie einen reproduzierbaren Devcontainer oder eine GitHub-Codespaces-Umgebung erstellen.

  • Visual Studio Code mit den folgenden Erweiterungen:

  • Azure Command-Line Interface (CLI) für die kennwortlose Authentifizierung unter macOS und Linux.

  • Wenn Sie uv noch nicht haben, befolgen Sie die Installationsanleitung.

  • Eine Quelldatenbank mit dem AdventureWorksLT Beispielschema und einer gültigen Verbindungszeichenfolge.

  • Eine Zieldatenbank mit einer gültigen Verbindungszeichenfolge. Der Benutzer muss über die Berechtigung zum Erstellen und Schreiben in Tabellen verfügen. Wenn du keine zweite Datenbank hast, kannst du dieselbe Datenbank und ein anderes Schema für die Zieltabellen verwenden.

Installieren Sie die einmaligen betriebsystem-spezifischen Voraussetzungen. Windows-Nutzer können diesen Schritt überspringen. Für vollständige Plattformdetails siehe Install mssql-python.

apk add libtool krb5-libs krb5-dev

Erstellen einer SQL-Datenbank

Erstellen oder verbinden Sie sich mit einer SQL-Datenbank auf einer der folgenden Plattformen:

Erstellen Sie das Projekt, und führen Sie den Code aus.

  1. Ein neues Projekt erstellen
  2. Hinzufügen von Abhängigkeiten
  3. Starten von Visual Studio Code
  4. Pyproject.toml aktualisieren
  5. Aktualisieren von main.py
  6. Speichern der Verbindungszeichenfolgen
  7. Verwenden Sie 'uv run', um das Skript auszuführen

Neues Projekt erstellen

  1. Öffnen Sie eine Eingabeaufforderung in Ihrem Entwicklungsverzeichnis. Wenn du keines hast, erstelle ein neues Verzeichnis, wie zum Beispiel python oder scripts. Vermeide Ordner auf deinem OneDrive, da Synchronisation die Verwaltung deiner virtuellen Umgebung beeinträchtigen kann.

  2. Erstellen Sie ein neues Projekt mit uv.

    uv init mssql-python-bcp-qs
    cd mssql-python-bcp-qs
    

Hinzufügen von Abhängigkeiten

Im selben Verzeichnis installieren Sie die Pakete mssql-python, python-dotenv und pyarrow.

uv add mssql-python python-dotenv pyarrow

Starten Sie Visual Studio Code.

Führen Sie im selben Verzeichnis den folgenden Befehl aus.

code .

Pyproject.toml aktualisieren

  1. Das pyproject.toml enthält die Metadaten für Ihr Projekt. Öffnen Sie die Datei in Ihrem bevorzugten Editor.

  2. Überprüfen Sie den Inhalt der Datei. Es sollte mit diesem Beispiel vergleichbar sein. Beachten Sie die Python-Version und Abhängigkeit, denn mssql-python verwendet >=, um eine Mindestversion zu definieren. Wenn Sie eine genaue Version bevorzugen, ändern Sie die >= Vorversionsnummer in ==. Die aufgelösten Versionen der einzelnen Pakete werden dann im uv.lock gespeichert. Die Sperrdatei stellt sicher, dass Entwickler, die am Projekt arbeiten, konsistente Paketversionen verwenden. Außerdem wird sichergestellt, dass derselbe Satz von Paketversionen beim Verteilen des Pakets an Endbenutzer verwendet wird. Sie sollten die uv.lock Datei nicht bearbeiten.

    [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.5.0",
        "python-dotenv>=1.1.1",
        "pyarrow>=19.0.0",
    ]
    
  3. Aktualisieren Sie die Beschreibung so, dass sie aussagekräftiger ist.

    description = "Bulk copies data between SQL databases using mssql-python and Apache Arrow"
    
  4. Speichern und schließen Sie die Datei.

Aktualisieren von main.py

  1. Öffnen Sie die Datei mit dem Namen main.py. Es sollte mit diesem Beispiel vergleichbar sein.

    def main():
        print("Hello from mssql-python-bcp-qs!")
    
    if __name__ == "__main__":
        main()
    
  2. Ersetzen Sie den Inhalt von main.py mit den folgenden Codeblöcken. Jeder Block baut auf dem vorherigen auf und sollte in main.py der Reihenfolge platziert werden.

    Tipp

    Wenn Visual Studio Code Probleme beim Beheben von Paketen hat, müssen Sie den Interpreter aktualisieren, um die virtuelle Umgebung zu verwenden.

  3. Fügen Sie am Anfang von main.py die Importe und Konstanten hinzu. Das Skript verwendet mssql_python für Datenbankkonnektivität und Arrow Fetch sowie pyarrowpyarrow.parquet für die Verarbeitung von Spaltendaten und Parquet-Datei-I/O, python-dotenv um Verbindungsstrings aus einer .env Datei zu laden, sowie ein kompiliertes Regex-Muster, das SQL-Identifikatoren validiert, um Einspeisung zu verhindern.

    """Round-trip: download tables from a source DB/schema to parquet, upload to a destination DB/schema."""
    
    import os
    import re
    import time
    
    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
    
  4. Fügen Sie unterhalb der Importe die SQL-zu-Arrow-Typzuordnung hinzu. Dieses Wörterbuch übersetzt SQL Server-Spaltentypen in die entsprechenden Apache Arrow-Typen, sodass die Datentreue beim Schreiben in Parquet beibehalten wird. Die Hilfsfunktionen erstellen exakte SQL-Typ-Zeichenketten (zum Beispiel NVARCHAR(100) oder DECIMAL(18,2)) aus INFORMATION_SCHEMA Metadaten und lösen den entsprechenden Pfeiltyp für jede Spalte auf. Diese Typen werden als Feldmetadaten in den Parquet-Dateien gespeichert, sodass die Zieltabelle mit exakten Spaltendefinitionen nachgebildet werden kann.

    _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("us"),
        "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()
    
  5. Fügen Sie die Schemaintrospection- und DDL-Generierungsfunktionen hinzu. _get_arrow_schema fragt INFORMATION_SCHEMA.COLUMNS mithilfe von parameterisierten Abfragen ab, erstellt ein Pfeilschema und speichert den ursprünglichen SQL-Typ als Feldmetadaten, damit die Zieltabelle mit exakten Spaltendefinitionen neu erstellt werden kann. _create_table_ddl liest diese Metadaten zurück, um DDL zu generieren DROP/CREATE TABLE . Der Typ timestamp (Rowversion) wird zu VARBINARY(8) neu zugeordnet, weil er automatisch generiert wird und nicht direkt eingefügt werden kann.

    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);"
        )
    
  6. Fügen Sie die Downloadfunktion hinzu. download_table verwendet cursor.arrow_batch(), um Daten direkt als Arrow-Record-Batches in der C++-Schicht des Treibers abzurufen, wodurch das Erstellen von Python-Zwischenobjekten vermieden wird. Jeder Batch wird in das metadatenangereicherte Schema aus _get_arrow_schema umgewandelt, sodass die ursprünglichen SQL-Typen (zum Beispiel NVARCHAR(100)) in der Parquet-Datei erhalten bleiben. Die Funktion verwendet zwei separate Cursor: einen zum Lesen von Spaltenmetadaten und einen zum Streamen der Daten.

    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(schema_name)
        _validate_ident(table_name)
        source = f"{schema_name}.[{table_name}]"
    
        with conn.cursor() as cursor:
            schema = _get_arrow_schema(cursor, schema_name, table_name)
    
        row_count = 0
        t0 = time.perf_counter()
    
        with conn.cursor() as cursor:
            cursor.execute(f"SELECT * FROM {source}")
            writer = None
            try:
                while True:
                    batch = cursor.arrow_batch(BATCH_SIZE)
                    if batch.num_rows == 0:
                        break
                    # Cast to the schema to preserve SQL type metadata in Parquet
                    arrays = [
                        batch.column(i).cast(schema.field(i).type)
                        for i in range(batch.num_columns)
                    ]
                    batch = pa.record_batch(arrays, schema=schema)
                    if writer is None:
                        writer = pq.ParquetWriter(parquet_file, schema)
                    writer.write_batch(batch)
                    row_count += batch.num_rows
            finally:
                if writer is not None:
                    writer.close()
    
        if row_count == 0:
            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
    
  7. Fügen Sie den Anreicherungshaken hinzu. enrich_parquet ist ein Platzhalter, in dem Sie Transformationen, abgeleitete Spalten oder Verknüpfungen zu Daten hinzufügen können, bevor sie hochgeladen werden. In dieser Schnellstartanleitung handelt es sich um eine no-op, die den Dateipfad unverändert zurückgibt.

    def enrich_parquet(parquet_file: str) -> str:
        """Enrich a parquet file before upload. Returns the (possibly new) file path."""
        # TODO: add transformations, derived columns, or joins
        print(f"Enriching {parquet_file} (no-op)")
        return parquet_file
    
  8. Fügen Sie die Upload-Funktion hinzu. upload_parquet liest das Arrow-Schema aus der Parquet-Datei, generiert und führt DROP/CREATE TABLE DDL aus, um das Ziel vorzubereiten, liest dann die Datei in Batches und ruft cursor.bulkcopy() für hochleistungsfähige Masseneinfügungen auf. Die table_lock=True-Option verbessert den Durchsatz, indem die Sperrkonkurrenz minimiert wird. Nachdem der Upload abgeschlossen ist, führt die Funktion ein SELECT COUNT(*) aus und gibt einen Fehler aus, wenn die Zielzeilenanzahl nicht mit der Anzahl der hochgeladenen Zeilen übereinstimmt.

    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
        elapsed = time.perf_counter() - t0
    
        # ── Verify ──
        with conn.cursor() as cursor:
            cursor.execute(f"SELECT COUNT(*) FROM {target}")
            count = cursor.fetchone()[0]
        if count != uploaded:
            raise ValueError(
                f"Row count mismatch for {target}: uploaded {uploaded:,}, destination has {count:,}"
            )
    
        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"| destination rows: {count:,}"
        )
        return uploaded
    
  9. Fügen Sie die Orchestrierungsfunktion hinzu. transfer_tables verbindet die drei Phasen zusammen. Es verbindet sich mit der Quelldatenbank, entdeckt alle Basistabellen im gegebenen Schema durch INFORMATION_SCHEMA.TABLES, lädt jede einzelne in eine lokale Parquet-Datei herunter, führt den Enrichment-Hook aus, verbindet sich dann mit der Zieldatenbank und lädt jede Datei hoch.

    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 = []
        for table_name, parquet_file in parquet_files:
            enriched.append((table_name, enrich_parquet(parquet_file)))
    
        # ── Upload to destination ──
        with mssql_python.connect(dest_conn_str) as dest_conn:
            for table_name, parquet_file in enriched:
                target = f"{dest_schema}.[{table_name}]"
                upload_parquet(dest_conn, parquet_file, target)
    
  10. Fügen Sie schließlich den main Einstiegspunkt hinzu. Sie lädt die .env Datei, ruft transfer_tables die Quell- und Zielverbindungszeichenfolgen auf und druckt die gesamte verstrichene Zeit.

    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()
    
  11. Speichern Sie und schließen Sie main.py.

Speichern der Verbindungszeichenfolgen

  1. Öffnen Sie die .gitignore Datei, und fügen Sie einen Ausschluss für Dateien hinzu .env . Ihre Datei sollte mit diesem Beispiel vergleichbar sein. Achten Sie darauf, sie zu speichern und zu schließen, wenn Sie fertig sind.

    # Python-generated files
    __pycache__/
    *.py[oc]
    build/
    dist/
    wheels/
    *.egg-info
    
    # Virtual environments
    .venv
    
    # Connection strings and secrets
    .env
    
  2. Erstellen Sie im aktuellen Verzeichnis eine neue Datei mit dem Namen .env.

  3. Fügen Sie in der .env Datei Einträge für die Quell- und Zielverbindungsstrings hinzu. Ersetzen Sie die Platzhalterwerte durch die tatsächlichen Server- und Datenbanknamen.

    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"
    

    Tipp

    Die hier verwendete Verbindungszeichenfolge hängt weitgehend vom Typ der SQL-Datenbank ab, mit der Sie eine Verbindung herstellen. Wenn Sie eine Verbindung mit einer Azure SQL-Datenbank oder einer SQL-Datenbank in Fabric herstellen, verwenden Sie die ODBC-Verbindungszeichenfolge auf der Registerkarte "Verbindungszeichenfolgen". Möglicherweise müssen Sie den Authentifizierungstyp je nach Szenario anpassen. Weitere Informationen zu Verbindungszeichenfolgen und deren Syntax finden Sie in der Referenz zur Verbindungszeichenfolgensyntax.

Tipp

Unter macOS funktionieren sowohl ActiveDirectoryInteractive als auch ActiveDirectoryDefault für die Microsoft Entra-Authentifizierung. ActiveDirectoryInteractive fordert Sie auf, sich jedes Mal anzumelden, wenn Sie das Skript ausführen. Um wiederholte Anmeldeaufforderungen zu vermeiden, melden Sie sich einmal über die Azure CLI an, indem Sie az login ausführen, und verwenden Sie dann ActiveDirectoryDefault, das die zwischengespeicherten Anmeldeinformationen wiederverwendet.

Verwenden Sie "uv run", um das Skript auszuführen.

  1. Führen Sie im Terminalfenster vor oder in einem neuen Terminalfenster, das im selben Verzeichnis geöffnet ist, den folgenden Befehl aus.

     uv run main.py
    

    Dies ist die erwartete Ausgabe, wenn das Skript abgeschlossen ist.

    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
    
  2. Verbinden Sie sich mit der Zieldatenbank über die MSSQL-Erweiterung für VS Code und überprüfen Sie, ob die Tabellen und Daten erfolgreich erstellt wurden.

  3. Um Ihr Skript auf einem anderen Computer bereitzustellen, kopieren Sie alle Dateien mit Ausnahme des .venv Ordners auf den anderen Computer. Die virtuelle Umgebung wird mit der ersten Ausführung neu erstellt.

Funktionsweise des Codes

Die Anwendung führt eine vollständige Roundtrip-Datenübertragung in drei Phasen durch:

  1. Download: Stellt eine Verbindung mit der Quelldatenbank her, liest Spaltenmetadaten aus INFORMATION_SCHEMA.COLUMNS, erstellt ein Apache Arrow-Schema und lädt dann jede Tabelle in eine lokale Parkettdatei herunter.
  2. Anreicher (optional): Stellt einen Hook (enrich_parquet) bereit, an dem Sie Transformationen, abgeleitete Spalten oder Verknüpfungen vor dem Hochladen hinzufügen können.
  3. Upload: Liest jede Parquet-Datei in Batches, erstellt die Tabelle in der Zieldatenbank mithilfe von DDL, die aus Arrow-Schemametadaten generiert wurde, und verwendet dann cursor.bulkcopy() für leistungsstarke Masseneinfügungen.

Nächste Schritte

Nutzen Sie diese Artikel, um weiter aufzubauen:

  • Massendatenkopie für erweiterte Muster für die Massendatenkopie, einschließlich Spaltenzuordnungen, Batchgrößen und Fehlerbehandlung.
  • Datenlade- und Bewegungsmuster für Strategien, um Daten aus Dateien, APIs und anderen Datenbanken zu laden.
  • Leistungsoptimierung zur Optimierung des Durchsatzes für große Datenoperationen.