Copy Activity in Azure Data Factory: A Practical Guide

Most Azure Data Factory pipelines are, at heart, a Copy activity with some decoration. It reads from one store and writes to another, and it is simple to configure until you ask what happens when the pipeline runs twice, when a row does not fit the target, or when the source changes shape. This guide covers the Copy activity with those questions in mind.

What the Copy activity does

Microsoft describes three steps (Microsoft Learn: Copy activity):

1. Read data from the source store.
2. Perform serialization or deserialization, compression or decompression, column mapping and so on.
3. Write data to the sink store.

It takes exactly one input dataset and one output dataset. It is a mover, not a transformer: it can change format and map columns, but it does not join, aggregate or apply business rules.

Formats and stores

The supported file formats are Avro, binary, delimited text, Excel, Iceberg (on Azure Data Lake Storage Gen2), JSON, ORC, Parquet and XML, across a long list of connectors for databases, file stores and applications.

Two modes are worth distinguishing:

• Copy as-is. Between two file stores, files can be copied without parsing, which the documentation notes is efficient because no serialization or deserialization takes place. Use the binary format for this.
• Parse and write. Read rows from a database and write Parquet; read CSV and write to a table. Here the dataset formats and the mapping matter.

A common and good pattern is database to Parquet in a data lake: the copy lands a typed, compressed, columnar file that everything downstream can read.

Mapping

By default the Copy activity maps source columns to sink columns by name. When names or types differ, define an explicit mapping. Three things to know:

• Import schemas and check the mapping before the first run; do not rely on defaults for a table that matters.
• Additional columns can be added during the copy, such as the source file path, which is invaluable for tracing a bad row back to its file.
• Auto-create table exists for SQL sinks. If the destination table does not exist, the copy can create it from the source. Microsoft positions this as a quick start, to be followed by reviewing the schema, and that is the right way to treat it. An auto-created table has generous types and no keys.

The decision that matters most: how the sink writes

For a SQL sink the default write behaviour is insert. The alternatives are a pre-copy script, which runs once before the copy, typically to delete or truncate, and upsert, which updates rows that match on key columns and inserts the rest (Microsoft Learn: Azure SQL Database connector).

The difference shows only when a pipeline runs more than once. The script below reproduces the three behaviours in SQLite, using the 59-row customer table from the Chinook sample database (Chinook on GitHub). It is a local reproduction of the semantics, not a run of Data Factory.

import shutil, sqlite3

shutil.copy("chinook.sqlite", "work.sqlite")           # never alter the downloaded file
db = sqlite3.connect("work.sqlite")
db.executescript("""
    CREATE TABLE sink_insert   (CustomerId INTEGER, LastName TEXT, Country TEXT);
    CREATE TABLE sink_truncate (CustomerId INTEGER, LastName TEXT, Country TEXT);
    CREATE TABLE sink_upsert   (CustomerId INTEGER PRIMARY KEY, LastName TEXT, Country TEXT);
""")
SOURCE = "SELECT CustomerId, LastName, Country FROM Customer"


def copy_run():
    """One run of the same copy, written three ways."""
    db.execute(f"INSERT INTO sink_insert {SOURCE}")                       # write behaviour: insert
    db.execute("DELETE FROM sink_truncate")                              # pre-copy script, then insert
    db.execute(f"INSERT INTO sink_truncate {SOURCE}")
    db.execute(f"""INSERT INTO sink_upsert {SOURCE} WHERE true
                   ON CONFLICT (CustomerId) DO UPDATE SET LastName = excluded.LastName,
                                                         Country = excluded.Country""")   # write behaviour: upsert


def report(label):
    counts = [db.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
              for t in ("sink_insert", "sink_truncate", "sink_upsert")]
    print(f"{label:34} insert {counts[0]:4}   truncate+insert {counts[1]:4}   upsert {counts[2]:4}")


print("source rows:", db.execute("SELECT COUNT(*) FROM Customer").fetchone()[0])
copy_run(); report("after run 1")
copy_run(); report("after run 2 (a rerun, no changes)")

# The source changes: one customer moves, one is added, one is deleted
db.execute("UPDATE Customer SET Country = 'Portugal' WHERE CustomerId = 12")
db.execute("INSERT INTO Customer (CustomerId, FirstName, LastName, Country, Email) VALUES (60, 'Asha', 'Verma', 'India', 'a@example.com')")
db.execute("DELETE FROM Customer WHERE CustomerId = 59")
copy_run(); report("after run 3 (source changed)")
print("source rows now:", db.execute("SELECT COUNT(*) FROM Customer").fetchone()[0])
print("customer 59 still in upsert sink:", db.execute("SELECT COUNT(*) FROM sink_upsert WHERE CustomerId = 59").fetchone()[0] == 1)

source rows: 59
after run 1                        insert   59   truncate+insert   59   upsert   59
after run 2 (a rerun, no changes)  insert  118   truncate+insert   59   upsert   59
after run 3 (source changed)       insert  177   truncate+insert   59   upsert   60
source rows now: 59
customer 59 still in upsert sink: True

Read down each column.

Insert has 118 rows after an innocent rerun and 177 after a third run. Every row is duplicated each time. This is the default.

Delete then insert has 59 rows every time and always matches the source exactly, including the deletion of customer 59 in run 3. It is the simplest correct behaviour for a table small enough to reload in full.

Upsert has 59 rows after the rerun, correctly, and 60 after run 3, although the source has 59. The moved customer was updated and the new one inserted, but the customer deleted at the source is still in the sink. Upsert does not delete.

For deletes with an incremental load you need something more: a soft-delete flag at the source, change data capture, or a periodic full comparison.

A Copy activity with the safety features on

Written from the documentation and not deployed:

{
  "name": "CopyCustomersToSql",
  "type": "Copy",
  "inputs": [ { "referenceName": "CustomersParquet", "type": "DatasetReference" } ],
  "outputs": [ { "referenceName": "CustomersStaging", "type": "DatasetReference" } ],
  "policy": { "timeout": "01:00:00", "retry": 2, "retryIntervalInSeconds": 60 },
  "typeProperties": {
    "source": { "type": "ParquetSource" },
    "sink": {
      "type": "AzureSqlSink",
      "preCopyScript": "TRUNCATE TABLE stg.customers",
      "writeBehavior": "insert"
    },
    "enableSkipIncompatibleRow": true,
    "logSettings": {
      "enableCopyActivityLog": true,
      "copyActivityLogSettings": { "logLevel": "Warning", "enableReliableLogging": false },
      "logLocationSettings": {
        "linkedServiceName": { "referenceName": "LakeStorage", "type": "LinkedServiceReference" },
        "path": "copy-logs/customers/"
      }
    }
  }
}

The pre-copy script makes the load repeatable, and it pairs correctly with retry: the script is invoked once per copy run, so when a failed attempt is retried the truncate runs again and partial rows are not doubled.

Bad rows: fail or skip?

By default the Copy activity stops and fails when it meets data it cannot write, a text value headed for an integer column, a key violation. With fault tolerance enabled it skips incompatible rows and can log them to a storage location, reporting the number skipped (Microsoft Learn: fault tolerance).

The documentation notes one restriction: skipping incompatible rows does not apply when the sink uses upsert or a stored procedure.

Neither setting is right everywhere:

• Fail when the data must be complete: financial records, reference data.
• Skip and log when a few bad rows should not block millions of good ones, and then alert on the skipped count. Skipping without looking at the log is the same as losing data quietly.

Staging

A copy can route data through interim blob storage before the sink, using enableStaging. It matters in two cases: loading a warehouse such as Azure Synapse Analytics through its bulk-loading mechanisms, and hybrid copies where staging in the cloud improves throughput. For an ordinary copy into a SQL database or a lake, leave it off.

Reading the output

Every run reports figures worth checking (Microsoft Learn: monitor copy activity):

• Rows read and rows copied, rowsRead and rowsCopied. They should match, unless rows were skipped.
• Rows skipped, rowsSkipped. Should be zero, or explained.
• Duration and throughput, copyDuration and throughput. A trend, not a single value, tells you when something is degrading.
• Compute used, usedDataIntegrationUnits and usedParallelCopies. The compute applied, which is what the copy is billed on.

Send rowsCopied to a log table after each run. A history of row counts is the cheapest monitoring there is.

Common mistakes

• Leaving the sink on insert for a table that is reloaded.
• Trusting auto-created tables in production.
• No retry and a 12-hour default timeout.
• Skipping bad rows with nobody reading the log.
• Using Copy to transform. Joins and business rules belong in a data flow, a stored procedure or the warehouse.
• Copying one table per pipeline by hand, when a parameterised copy in a loop would do.

For the pipeline around the activity, see ADF pipelines explained; for loading only what changed, incremental data loading; and for speed, ADF performance tuning and data partitioning in the Copy activity.

Azure Data Factory and ETL pipeline design are covered in the Data Engineering course.

Sources

• Copy activity in Azure Data Factory and Azure Synapse Analytics (Microsoft Learn) — Microsoft Learn
• Copy and transform data in Azure SQL Database (Microsoft Learn) — Microsoft Learn
• Fault tolerance of copy activity in Azure Data Factory (Microsoft Learn) — Microsoft Learn
• Monitor copy activity (Microsoft Learn) — Microsoft Learn
• Chinook sample database (GitHub) — GitHub