pip install git+https://github.com/TencentCloud/tencentcloud-cls-sdk-python.git
log_group {
source //日志来源,一般为机器的 IP
filename //日志文件名
logs {
time //日志时间,unix 时间戳,微秒级别
user_defined_log_kvs //用户日志字段
}
}
class SampleConsumer(ConsumerProcessorBase):last_check_time = 0log_results = []lock = RLock()def initialize(self, topic_id):self.topic_id = topic_id# 处理消费的数据def process(self, log_groups, offset_tracker):for log_group in log_groups:for log in log_group.logs:# 处理单行数据item = dict()item['filename'] = log_group.filenameitem['source'] = log_group.sourceitem['time'] = log.timefor content in log.contents:item[content.key] = content.valuewith SampleConsumer.lock:# 数据汇总到SampleConsumer.log_resultsSampleConsumer.log_results.append(item)# 每隔3s提交一次offsetcurrent_time = time.time()if current_time - self.last_check_time > 3:try:self.last_check_time = current_timeoffset_tracker.save_offset(True)except Exception:import tracebacktraceback.print_exc()else:try:offset_tracker.save_offset(False)except Exception:import tracebacktraceback.print_exc()return None# Worker退出时,会调用该函数,可以在此处执行清理工作def shutdown(self, offset_tracker):try:offset_tracker.save_offset(True)except Exception:import tracebacktraceback.print_exc()
配置参数 | 说明 | 默认值 | 取值范围 |
endpoint | - | 支持地域:北京,上海,广州,南京,中国香港,东京,美东,新加坡,法兰克福 | |
access_key_id | - | - | |
access_key | - | - | |
region | - | 支持地域:北京,上海,广州,南京,中国香港,东京,美东,新加坡,法兰克福 | |
logset_id | 日志集 ID,仅支持一个日志集 | - | - |
topic_ids | 日志主题 ID,多个主题请使用','隔开 | - | - |
consumer_group_name | 消费者组名称 | - | - |
internal | 内网:TRUE 公网:FALSE 说明: | FALSE | TRUE/FALSE |
consumer_name | 消费者名称。同一个消费者组内,消费者名称不可重复 | - | 0-9、aA-zZ、'-'、'_'、'.'组成的字符串 |
heartbeat_interval | 消费者心跳上报间隔,2个间隔没有上报心跳,会被认为是消费者下线 | 20 | 0-30分钟 |
data_fetch_interval | 消费者拉取数据间隔,不小于1秒 | 2 | - |
offset_start_time | 拉取数据的开始时间,字符串类型的 unix 时间戳,精度为秒,例如"1711607794",也可以直接可配置为"begin"和"end"。 begin:日志主题生命周期内的最早数据 end:日志主题生命周期内的最新数据 | "end" | "begin"/"end"/unix时间戳 |
max_fetch_log_group_size | 消费者单次拉取数据大小,默认2M,最大10M | 2097152 | 2M - 10M |
offset_end_time | 拉取数据的结束时间,支持字符串类型的 unix 时间戳,精度为秒,例如"1711607794"。不填写代表持续拉取 | - | - |
def sample_consumer_group():# 日志服务接入点,请您根据实际情况填写endpoint = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_ENDPOINT', '')# 访问的地域region = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_REGION', '')# 用户的Secret_idaccess_key_id = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_ACCESSID', '')# 用户的Secret_keyaccess_key = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_ACCESSKEY', '')# 消费的日志集IDlogset_id = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_LOGSET_ID', '')# 消费的日志主题ID列表,多个主题用英文逗号分隔topic_ids = os.environ.get('TENCENTCLOUD_LOG_SAMPLE_TOPICS', '').split(',')# 消费组名称,同一个日志集下的消费组名称唯一consumer_group = 'consumer-group-1'# 消费者名称consumer_name1 = "consumer-group-1-A"consumer_name2 = "consumer-group-1-B"assert endpoint and access_key_id and access_key and logset_id, ValueError("endpoint/access_id/access_key and ""logset_id cannot be empty")# 创建访问云API接口的Clientclient = YunApiLogClient(access_key_id, access_key, region=region)SampleConsumer.log_results = []try:# 创建两个消费者配置option1 = LogHubConfig(endpoint, access_key_id, access_key, region, logset_id, topic_ids, consumer_group,consumer_name1, heartbeat_interval=3, data_fetch_interval=1,offset_start_time='end', max_fetch_log_group_size=1048576)option2 = LogHubConfig(endpoint, access_key_id, access_key, region, logset_id, topic_ids, consumer_group,consumer_name2, heartbeat_interval=3, data_fetch_interval=1,offset_start_time='end', max_fetch_log_group_size=1048576)# 创建消费者print("*** start to consume data...")client_worker1 = ConsumerWorker(SampleConsumer, consumer_option=option1)client_worker2 = ConsumerWorker(SampleConsumer, consumer_option=option2)# 启动消费者client_worker1.start()client_worker2.start()# 等待2分钟,或者获取到数据后继续往后执行sleep_until(120, lambda: len(SampleConsumer.log_results) > 0)# 关闭消费者print("*** stopping workers")client_worker1.shutdown()client_worker2.shutdown()# 打印汇总的日志数据print("*** get content:")for log in SampleConsumer.log_results:print(json.dumps(log))# 打印消费组信息:消费组的名称、消费的日志主题、消费者心跳超时时间print("*** consumer group status ***")ret = client.list_consumer_group(logset_id, topic_ids)ret.log_print()# 删除消费组print("*** delete consumer group")time.sleep(30)client.delete_consumer_group(logset_id, consumer_group)except Exception as e:import tracebacktraceback.print_exc()raise eif __name__ == '__main__':sample_consumer_group()
本页内容是否解决了您的问题?