在Azure Data Factory V2中添加提取时间戳的通用实现问询

Great question! You can absolutely build this generic solution without manual column mapping using ADF's built-in features—no complex custom activity code required (though I’ll mention a code-based option too if you need it). Here are the most concise, scalable approaches:

Approach 1: Mapping Data Flow (Recommended for Low-Code Genericity) #

This is the cleanest way to dynamically add an extraction timestamp while preserving all original columns automatically, no manual mapping needed.

• Step 1: Enable Schema Drift on the Source
  Connect your Salesforce (or other) source dataset. In the source settings, turn on Schema Drift (under Schema Options) to capture all columns from your source without explicitly defining them. This is the core of making the solution generic—ADF will automatically pick up any new columns added to the source later.

• Step 2: Add a Derived Column for the Timestamp
  Insert a Derived Column transformation right after the source. Create a new column (e.g., ExtractionTimestamp) and set its value to one of these ADF system expressions:

  • currentUTC(): Gets the exact UTC time when the transformation runs (per-row timestamp)
  • pipeline().TriggerTime: Uses the UTC time when the pipeline was triggered (consistent timestamp for all rows in a single batch run)
    Since schema drift is enabled, all original columns are retained automatically—no need to map each column individually.

• Step 3: Configure the Blob Sink with Schema Drift
  Connect your Blob Storage sink dataset (CSV, Parquet, JSON, etc.). In the sink settings, enable Schema Drift and Allow schema addition so ADF automatically adds the new timestamp column to your output file. For CSV sinks, make sure "First row as header" is checked so the timestamp column name appears in the output header.

Approach 2: Custom Activity (Code-Based Control) #

If you prefer full control via code, you can use a custom activity (e.g., Python script hosted in Azure Functions or Azure Batch) to append the timestamp. Here’s a simplified snippet example:

import pandas as pd
from azure.storage.blob import BlobServiceClient
from datetime import datetime

# Read source data (adjust based on your source type)
df = pd.read_csv("source_data_path")

# Append extraction timestamp (use UTC for consistency)
df['ExtractionTimestamp'] = datetime.utcnow()

# Write to Blob Storage
blob_service_client = BlobServiceClient.from_connection_string("your_blob_connection_string")
container_client = blob_service_client.get_container_client("your_target_container")
container_client.upload_blob(
    name="output_with_timestamp.csv",
    data=df.to_csv(index=False),
    overwrite=True
)

This gives you flexibility, but requires maintaining code—so the Data Flow approach is better for a low-maintenance, generic solution.

Key Tips #

• Consistency: Use pipeline().TriggerTime if you want all rows from a single pipeline run to share the same timestamp (ideal for batch processing).
• Schema Evolution: Keeping schema drift enabled ensures your solution won’t break if the source adds or removes columns later.
• Format Compatibility: For Parquet or JSON sinks, schema drift works seamlessly without extra configuration beyond enabling the setting.

内容的提问来源于stack exchange，提问作者MereLy Perfect