Ingesting Data From REST API endpoints: Data En gineering with Fabric (2024-11-13)

import pyodbc
from google.cloud import bigquery
from datetime import datetime, timedelta

# SQL Server Connection Configuration
sql_conn_str = "DRIVER={ODBC Driver 17 for SQL Server};SERVER=your_server_name;DATABASE=your_db_name;UID=your_username;PWD=your_password"

# BigQuery Configuration
bq_client = bigquery.Client()
bq_table_id = "your_project_id.your_dataset.your_table"

# Initialize BigQuery Schema for Streaming Inserts
schema = [
bigquery.SchemaField("id", "INTEGER"),
bigquery.SchemaField("column1", "STRING"),
bigquery.SchemaField("column2", "STRING"),
bigquery.SchemaField("operation_type", "STRING"), # To track inserts, updates, deletes
bigquery.SchemaField("timestamp", "TIMESTAMP")
]

# Step 1: Retrieve changes from CDC in SQL Server
def fetch_cdc_changes(last_sync_time):
conn = pyodbc.connect(sql_conn_str)
cursor = conn.cursor()

# Query the CDC table to get changes since the last sync (inserts/updates/deletes)
query = f"""
SELECT __$operation, id, column1, column2, __$start_lsn, __$seqval
FROM cdc.your_table_CT
WHERE __$start_lsn > ? ORDER BY __$start_lsn ASC
"""
cursor.execute(query, last_sync_time)
changes = cursor.fetchall()

cursor.close()
conn.close()

return changes

# Step 2: Stream data to BigQuery
def stream_to_bigquery(rows):
errors = bq_client.insert_rows_json(bq_table_id, rows) # Stream rows to BigQuery
if errors:
print(f"Errors occurred: {errors}")
else:
print(f"Successfully streamed {len(rows)} rows to BigQuery.")

# Helper: Map CDC operation codes to human-readable operations
def map_operation(op_code):
operation_map = {1: 'DELETE', 2: 'INSERT', 3: 'UPDATE'}
return operation_map.get(op_code, 'UNKNOWN')

# Main ETL Process
def real_time_etl(last_sync_time):
# Fetch changes from SQL Server
changes = fetch_cdc_changes(last_sync_time)
rows_to_insert = []

for change in changes:
operation_type = map_operation(change[0]) # Map CDC operation codes
rows_to_insert.append({
"id": change[1],
"column1": change[2],
"column2": change[3],
"operation_type": operation_type,
"timestamp": datetime.utcnow() # Use current UTC time for tracking
})

if rows_to_insert:
# Stream the changes to BigQuery in real-time
stream_to_bigquery(rows_to_insert)

if __name__ == "__main__":
# Step 3: Track the last synchronization time
# For the first run, initialize last_sync_time with a past date or store it externally (e.g., in a database)
last_sync_time = datetime.utcnow() - timedelta(minutes=5) # Example: sync changes in the last 5 minutes

# Real-time ETL process
while True:
real_time_etl(last_sync_time)
# Update last_sync_time after each successful sync (could be saved externally)
last_sync_time = datetime.utcnow()

# Sleep for a short interval to mimic near-real-time processing (e.g., every 1 minute)
time.sleep(60)