Python High-Level SDK

Method 1: Environment variables (recommended)

The Alibaba Cloud SDK reads credentials from environment variables, so you don't need to hard-code them.

export ALIBABA_CLOUD_ACCESS_KEY_ID=<your_access_key_id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<your_access_key_secret>

import os
from alibabacloud_credentials.client import Client as CredClient
from alibabacloud_credentials.models import Config as CredConfig
from datahub.client import ProducerConfig, ConsumerConfig

config = CredConfig(
    type='access_key',
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
)
credential = CredClient(config)

# Producer
producer_config = ProducerConfig.from_credential(credential, endpoint)

# Consumer
consumer_config = ConsumerConfig.from_credential(credential, endpoint)

Method 2: Construct with an AccessKey pair

from datahub.client import ProducerConfig

# Method A: Pass as arguments directly
producer_config = ProducerConfig(access_id, access_key, endpoint)

# Method B: Use the from_access factory method (recommended)
producer_config = ProducerConfig.from_access(access_id, access_key, endpoint)

Important: If the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are set, they take precedence over the configuration file.

Limitations

• Thread safety: A producer is thread-safe. Use one instance per topic per process.

ProducerConfig parameters

Note: The Python SDK supports NONE, LZ4, DEFLATE, and ZLIB compression formats. ZSTD is not supported.

Synchronous write

import os
from alibabacloud_credentials.client import Client as CredClient
from alibabacloud_credentials.models import Config as CredConfig
from datahub.core import DatahubProtocolType
from datahub.models import TupleRecord, BlobRecord, FieldType, RecordSchema, CompressFormat
from datahub.client import DatahubProducer, ProducerConfig
from datahub.exceptions import DatahubException

# ====== Configuration ======
endpoint = "https://dh-cn-hangzhou.aliyuncs.com"
project_name = "test_project"
topic_name = "test_topic"

# ====== Authentication ======
config = CredConfig(
    type='access_key',
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
)
credential = CredClient(config)

producer_config = ProducerConfig.from_credential(credential, endpoint)
producer_config.protocol_type = DatahubProtocolType.PB
producer_config.compress_format = CompressFormat.LZ4
producer_config.retry_times = 3

# ====== Create a producer ======
producer = DatahubProducer(project_name, topic_name, producer_config)

try:
    # Get topic metadata (contains the field list)
    topic_meta = producer.topic_meta

    # Construct TUPLE records
    records = []
    for i in range(100):
        record = TupleRecord(schema=topic_meta.record_schema)
        record.set_value(0, f"user_{i}")          # Set field value by index
        record.set_value(1, "login")
        record.set_value(2, 1700000000 + i)
        record.put_attribute("source", "python-sdk")
        records.append(record)

    # Synchronous write (blocks until the write operation completes)
    shard_id = producer.write(records)
    print(f"Synchronous write succeeded. Written to Shard: {shard_id}")

finally:
    producer.close()

Note: The TupleRecord object does not have a .schema attribute. To access field metadata, use topic_meta.record_schema.field_list on the Producer side, or record.field_list on the Consumer side.

Asynchronous write

producer = DatahubProducer(project_name, topic_name, producer_config)

try:
    topic_meta = producer.topic_meta

    records = []
    for i in range(100):
        record = TupleRecord(schema=topic_meta.record_schema)
        record.set_value(0, f"user_{i}")
        record.set_value(1, "click")
        record.set_value(2, 1700000000 + i)
        records.append(record)

    # Asynchronous write (non-blocking, returns immediately)
    shard_id = producer.write_async(records)
    print(f"Asynchronous write submitted. Written to Shard: {shard_id}")

    # Call flush() before the program exits to ensure all data is sent.
    producer.flush()

finally:
    producer.close()

Note: Async writes buffer data locally and send it via a background thread pool. Call producer.flush() before the program exits to ensure all data is sent.

Limitations

• Subscription ID: Consumption requires a subscription, which maintains independent offsets.

• Thread safety: A consumer is not thread-safe. Each thread must use its own instance.

• Consumption modes: Group consumption (automatic shard assignment) and specified shard consumption (manual assignment).

ConsumerConfig parameters

Auto ACK (recommended)

The offset is auto-committed after each successful read. This is the simplest approach, suitable for scenarios that tolerate minor data loss.

import os
from alibabacloud_credentials.client import Client as CredClient
from alibabacloud_credentials.models import Config as CredConfig
from datahub.core import DatahubProtocolType
from datahub.models import CompressFormat, TupleRecord, BlobRecord, FieldType
from datahub.client import DatahubConsumer, ConsumerConfig
from datahub.exceptions import DatahubException

# ====== Configuration ======
endpoint = "https://dh-cn-hangzhou.aliyuncs.com"
project_name = "test_project"
topic_name = "test_topic"
sub_id = "YOUR_SUBSCRIPTION_ID"  # Create this in the DataHub console

# ====== Authentication ======
config = CredConfig(
    type='access_key',
    access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
)
credential = CredClient(config)

consumer_config = ConsumerConfig.from_credential(credential, endpoint)
consumer_config.auto_ack_offset = True    # This is the default value and can be omitted.

# ====== Create a consumer (group consumption mode) ======
datahub_consumer = DatahubConsumer(
    project_name, topic_name, sub_id, consumer_config
)

record_cnt = 0
try:
    while True:
        record = datahub_consumer.read(timeout=60)
        if record is None:
            # Returns None if no data is available within the timeout period.
            break

        # Differentiate between TupleRecord and BlobRecord
        if isinstance(record, TupleRecord):
            fields = record.values            # Tuple of field values
            field_list = record.field_list    # List of field metadata
            print(f"Tuple record: {fields}")
            # Access by field name
            # for idx, field in enumerate(field_list):
            #     print(f"  {field.name} = {fields[idx]}")
        elif isinstance(record, BlobRecord):
            print(f"Blob record: {record.blob_data}")

        record_cnt += 1
        # When auto_ack_offset=True, the offset is automatically committed after read() returns.

except DatahubException as e:
    print(f"Consumption error: {e}")
finally:
    datahub_consumer.close()

print(f"Consumed {record_cnt} records in total.")

Manual ACK

Commit the offset only after each record is successfully processed.

consumer_config = ConsumerConfig.from_credential(credential, endpoint)
consumer_config.auto_ack_offset = False   # Disable auto ACK

datahub_consumer = DatahubConsumer(
    project_name, topic_name, sub_id, consumer_config
)

try:
    while True:
        record = None
        try:
            record = datahub_consumer.read(timeout=60)
            if record is not None:
                # TODO: Process the data.
                process_record(record)
        except DatahubException as e:
            print(f"Read error: {e}")
        finally:
            # After handling a successfully read record, you must manually ACK it.
            if record is not None:
                record.record_key.ack()
                print("Manual ACK succeeded.")
finally:
    datahub_consumer.close()

Note: With auto ACK disabled, you must call record.record_key.ack() for each record. Otherwise, the offset will not advance.

Group consumption

Without shard_ids, the consumer joins a consumer group and DataHub auto-assigns shards.

# Do not pass shard_ids to enable automatic group consumption.
datahub_consumer = DatahubConsumer(
    project_name, topic_name, sub_id, consumer_config
)

# Start multiple consumer instances (in different processes or threads),
# and the service will automatically assign shards.
# Instance 1 -> Assigned Shards 0, 1
# Instance 2 -> Assigned Shards 2, 3

Behavior:

• Consumers with the same subscription ID form a consumer group.

• Shards are auto-assigned so each shard is consumed by only one consumer at a time.

• When a consumer goes offline, its shards are reassigned to other active consumers.

Q: Synchronous vs. asynchronous writes?

• Synchronous write (write): Blocks until the write succeeds or retries are exhausted. Use when data reliability is critical.

• Asynchronous write (write_async): Non-blocking; buffers data for background sending. Use for high-throughput scenarios. Call flush() before the program exits.

Q: Auto ACK vs. manual ACK?

• Auto ACK: Simple and efficient, but risks minor data loss if the process crashes during processing.

• Manual ACK: Commits the offset only after the record is processed. Use for mission-critical scenarios such as financial transactions.

Q: How do I view logs?

Logs are written to ./DatahubClient.log by default. Override the path in the config:

import logging

producer_config.logging_level = logging.DEBUG
producer_config.logging_filename = "/var/log/datahub/producer.log"

Q: How do I use record.record_key.ack()?

record.record_key.ack() works only in manual ACK mode. The record must be returned by consumer.read(), which injects record_key automatically.

consumer_config.auto_ack_offset = False

datahub_consumer = DatahubConsumer(
    project_name, topic_name, sub_id, consumer_config
)

try:
    while True:
        record = datahub_consumer.read(timeout=60)
        if record is not None:
            # TODO: Process the data.
            process_record(record)
            # Manually ACK after processing is complete.
            record.record_key.ack()
finally:
    datahub_consumer.close()

Note: Manually created records (e.g., a TupleRecord from a producer) have no record_key. Calling record.record_key.ack() raises AttributeError: 'NoneType' object has no attribute 'ack'.