如何将超量JSON事件导入Azure Event Hub（解决1MB大小限制）

嘿，我来帮你搞定这个Event Hub的消息大小限制问题！针对你这种包含大量格式一致JSON事件的场景，有几个实用的方案可以试试：

1. 用Event Hub原生批量发送API（最省心的方案） #

Azure的Event Hub SDK本身就支持自动批量处理，它会帮你把事件打包成不超过大小限制的批次，不用你自己手动算数量。你只需要设置好批次的最大大小（建议留些余量，比如设成900KB，避免因为序列化后的微小差异超过1MB），然后把所有事件丢进去就行。

举个Python的例子（用最新的azure-eventhub SDK）：

from azure.eventhub import EventHubProducerClient, EventData
import json

# 你的大型JSON数据
DATA = [{"Id": "393092", "UID": "7f0034ee", "date": "2023-01-06", "f_id": "430", "origin": "CN"}, ...]

# 初始化生产者客户端
producer = EventHubProducerClient.from_connection_string(
    conn_str="你的Event Hub连接字符串",
    eventhub_name="你的Event Hub名称"
)

with producer:
    # 创建一个批次，设置最大大小为900KB（900*1024字节）
    batch = producer.create_batch(max_size_in_bytes=900*1024)
    for event in DATA:
        # 把JSON事件序列化成字节
        event_data = EventData(json.dumps(event).encode('utf-8'))
        try:
            batch.add(event_data)
        except ValueError:
            # 如果当前批次满了，发送这个批次，再创建新的批次
            producer.send_batch(batch)
            batch = producer.create_batch(max_size_in_bytes=900*1024)
            batch.add(event_data)
    # 发送最后一个批次
    if batch:
        producer.send_batch(batch)

这个方法的好处是SDK自动帮你处理批次拆分，不用你自己计算每条事件的大小，容错性很强。

2. 动态按大小分片（精准控制） #

如果你需要更精准的控制，可以自己计算每条序列化后JSON的字节大小，累加直到接近1MB（比如950KB），然后打包成一个消息发送。这种方式适合对批次大小有严格要求的场景。

示例代码片段：

import json

def split_events_by_size(events, max_size_bytes=950*1024):
    batches = []
    current_batch = []
    current_size = 0

    for event in events:
        # 序列化当前事件并计算大小
        event_str = json.dumps(event)
        event_size = len(event_str.encode('utf-8'))
        
        # 如果加入当前批次会超过上限，就把当前批次存入列表，新建批次
        if current_size + event_size > max_size_bytes:
            batches.append(current_batch)
            current_batch = [event]
            current_size = event_size
        else:
            current_batch.append(event)
            current_size += event_size
    # 加入最后一个批次
    if current_batch:
        batches.append(current_batch)
    
    return batches

# 拆分你的数据
event_batches = split_events_by_size(DATA)

# 然后逐个批次发送到Event Hub（发送逻辑和上面的批量API类似）

3. 压缩消息批次（提升容量） #

如果单个事件本身比较大，你可以把整个批次的JSON数组压缩后再发送，这样1MB的限制就能装下更多事件。常用的压缩方式是gzip，消费端收到消息后需要先解压再处理。

示例代码（发送端）：

import gzip
import json
from azure.eventhub import EventHubProducerClient, EventData

producer = EventHubProducerClient.from_connection_string(...)

with producer:
    batch = producer.create_batch(max_size_in_bytes=900*1024)
    temp_batch = []
    for event in DATA:
        temp_batch.append(event)
        # 试试压缩当前临时批次，看大小是否符合要求
        compressed_data = gzip.compress(json.dumps(temp_batch).encode('utf-8'))
        if len(compressed_data) > 900*1024:
            # 压缩后超过大小，移除最后一个事件，发送当前批次
            temp_batch.pop()
            compressed_batch = gzip.compress(json.dumps(temp_batch).encode('utf-8'))
            batch.add(EventData(compressed_batch))
            producer.send_batch(batch)
            batch = producer.create_batch(max_size_in_bytes=900*1024)
            temp_batch = [event]
    # 发送最后一个压缩批次
    if temp_batch:
        compressed_batch = gzip.compress(json.dumps(temp_batch).encode('utf-8'))
        batch.add(EventData(compressed_batch))
        producer.send_batch(batch)

消费端需要对应解压：

import gzip
import json

def process_event(event_data):
    compressed_data = event_data.body_as_bytes()
    uncompressed_data = gzip.decompress(compressed_data)
    events = json.loads(uncompressed_data)
    # 处理事件...

4. 优化JSON结构（减少冗余） #

因为你的事件格式一致，很多字段（比如Id、UID、date）可能重复出现，你可以优化JSON结构，把公共字段提出来，减少整体大小。比如把原来的数组改成：

{
  "common_fields": {"Id": "393092", "UID": "7f0034ee", "date": "2023-01-06"},
  "events": [{"f_id": "430", "origin": "CN"}, {"f_id": "430", "origin": "CN"}, ...]
}

这样每个事件只需要存储差异字段，能大幅减少总大小，同样的1MB能装下更多事件。消费端收到消息后，再把公共字段和每个事件合并还原即可。

优先推荐用方案1的批量API，因为它最省心，SDK已经帮你处理了大部分细节。如果还是觉得容量不够，可以搭配方案3的压缩一起用，效果会更明显。

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