HBase to Cosmos DB Migration¶

# Connect
from azure.cosmos import CosmosClient, PartitionKey

client = CosmosClient(endpoint, credential)
database = client.get_database_client("orders_db")
container = database.get_container_client("orders")

# Upsert (insert or update)
container.upsert_item({
    "id": "order-12345",
    "partitionKey": "user-789",
    "info": {"status": "shipped"}
})

# Point read (fastest operation)
item = container.read_item(item="order-12345", partition_key="user-789")

# Query (range equivalent of HBase scan)
results = container.query_items(
    query="SELECT * FROM c WHERE c.partitionKey = @pk",
    parameters=[{"name": "@pk", "value": "user-789"}],
    partition_key="user-789"
)
for item in results:
    # Process each document
    pass

# Delete
container.delete_item(item="order-12345", partition_key="user-789")

# Connect using cassandra-driver (familiar for HBase teams)
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider

auth = PlainTextAuthProvider(username, password)
cluster = Cluster([contact_point], port=10350, auth_provider=auth, ssl_options=ssl_opts)
session = cluster.connect("orders_keyspace")

# Insert
session.execute("""
    INSERT INTO orders (user_id, order_id, status, total)
    VALUES (%s, %s, %s, %s)
""", ("user-789", "order-12345", "shipped", 209.97))

# Read by partition key
rows = session.execute("""
    SELECT * FROM orders WHERE user_id = %s
""", ("user-789",))

# Read by partition key + clustering key
row = session.execute("""
    SELECT * FROM orders WHERE user_id = %s AND order_id = %s
""", ("user-789", "order-12345"))

# Delete
session.execute("""
    DELETE FROM orders WHERE user_id = %s AND order_id = %s
""", ("user-789", "order-12345"))

# Azure Function triggered by Cosmos DB Change Feed
import azure.functions as func
import json

def main(documents: func.DocumentList):
    """Process changes from Cosmos DB orders container."""
    for doc in documents:
        order = json.loads(doc.to_json())

        # Pattern 1: Audit logging (replaces Observer coprocessor)
        log_audit_event(order)

        # Pattern 2: Secondary index update
        update_customer_index(order)

        # Pattern 3: Notification on status change
        if order.get("status") == "shipped":
            send_shipping_notification(order)

def log_audit_event(order):
    """Write audit log to Log Analytics or a separate container."""
    # Implementation here
    pass

def update_customer_index(order):
    """Maintain a denormalized view by customer."""
    # Implementation here
    pass

def send_shipping_notification(order):
    """Send notification via Event Grid or Service Bus."""
    # Implementation here
    pass

# Read from HBase using Spark HBase connector
hbase_df = spark.read.format("org.apache.spark.sql.execution.datasources.hbase") \
    .options(catalog=hbase_catalog) \
    .load()

# Transform to Cosmos DB document structure
cosmos_df = hbase_df.select(
    col("row_key").alias("id"),
    col("info:customer_name").alias("customer_name"),
    col("info:email").alias("email"),
    col("order:date").alias("order_date"),
    col("order:status").alias("status"),
    col("order:total").cast("double").alias("total")
).withColumn(
    "partitionKey",
    # Extract user ID from composite row key
    split(col("id"), "\\|")[0]
)

# Write to Cosmos DB using Spark connector
cosmos_df.write.format("cosmos.oltp") \
    .options(**{
        "spark.cosmos.accountEndpoint": cosmos_endpoint,
        "spark.cosmos.accountKey": cosmos_key,
        "spark.cosmos.database": "orders_db",
        "spark.cosmos.container": "orders",
        "spark.cosmos.write.strategy": "ItemOverwrite",
        "spark.cosmos.write.bulk.enabled": "true"
    }) \
    .mode("append") \
    .save()

# Compare row counts
hbase_count = spark.read.format("hbase").options(catalog=cat).load().count()
cosmos_count = spark.read.format("cosmos.oltp").options(**cosmos_opts).load().count()
assert hbase_count == cosmos_count

# Compare sample data
hbase_sample = spark.read.format("hbase").options(catalog=cat).load() \
    .filter("row_key = 'user-789|order-12345'").collect()
cosmos_sample = container.read_item("order-12345", "user-789")
# Compare field-by-field