Python SDK Emitter Connection Failures

Python SDK emitter fails to connect to DataHub or emitted metadata is not appearing. This typically occurs due to incorrect GMS endpoint URL, network connectivity issues, authentication problems, or using the wrong emitter type.

Common Error Messages:

• Failed to connect to DataHub at http://localhost:8080
• Connection refused
• 401 Unauthorized
• SSL certificate verification failed

You Might Be Asking:

• How do I emit metadata using the Python SDK?
• What's the difference between REST and Kafka emitter?
• Why is my metadata not showing up after emission?

1. Configure REST emitter correctly:

from datahub.emitter.rest_emitter import DatahubRestEmitter

# Basic configuration
emitter = DatahubRestEmitter(
    gms_server="http://localhost:8080",
    token="your-access-token"  # If authentication enabled
)

# For HTTPS with custom cert
emitter = DatahubRestEmitter(
    gms_server="https://datahub.company.com",
    token="your-token",
    extra_headers={"Custom-Header": "value"},
    verify_ssl=True,  # Or path to CA cert
    timeout_sec=30
)

2. Test connection before emitting:

from datahub.emitter.rest_emitter import DatahubRestEmitter

emitter = DatahubRestEmitter(gms_server="http://localhost:8080")

# Test the connection
try:
    emitter.test_connection()
    print("Connection successful!")
except Exception as e:
    print(f"Connection failed: {e}")

3. Use Kafka emitter for high-volume scenarios:

from datahub.emitter.kafka_emitter import DatahubKafkaEmitter, KafkaEmitterConfig

emitter = DatahubKafkaEmitter(
    config=KafkaEmitterConfig(
        connection=KafkaProducerConnectionConfig(
            bootstrap="kafka:9092",
            schema_registry_url="http://schema-registry:8081"
        )
    )
)

4. Emit metadata with error handling:

Example for Snowflake. This can be adapted for other data sources/connectors.

from datahub.emitter.mce_builder import make_dataset_urn
from datahub.metadata.schema_classes import DatasetPropertiesClass
from datahub.emitter.mcp import MetadataChangeProposalWrapper

# Create MCP
dataset_urn = make_dataset_urn("snowflake", "db.schema.table")
mcp = MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=DatasetPropertiesClass(
        description="Updated via Python SDK"
    )
)

# Emit with error handling
try:
    emitter.emit(mcp)
    print("Metadata emitted successfully")
except Exception as e:
    print(f"Failed to emit: {e}")
finally:
    emitter.close()

5. For DataHub Cloud/Managed, use correct endpoint:

emitter = DatahubRestEmitter(
    gms_server="https://your-instance.acryl.io/gms",
    token="your-api-token"
)

Always close the emitter after use to ensure all events are flushed. The REST emitter provides immediate feedback on errors, while Kafka emitter is asynchronous and better for high throughput.

• Python SDK Guide
• Emitters