python每5分钟从kafka中提取数据的例子
我就废话不多说了，直接上代码吧！
importsys
sys.path.append("..")
fromdatetimeimportdatetime
fromutils.kafka2fileimportKafkaDownloader
importos
"""
实现取kafka数据，文件按照取数据的间隔命名
如每5分钟从kafka取数据写入文件中，文件名为当前时间加5
"""
TOPIC="rtz_queue"
HOSTS="ip:9092,ip:9092"
GROUP="2001"
defget_end_time(hour,minute,time_step):
if(minute+time_step)%60<60:
if(minute+time_step)%60<10:
returnstr(hour+int((minute+time_step)/60))+":"+"0"+str((minute+time_step)%60)
else:
returnstr(hour+int((minute+time_step)/60))+":"+str((minute+time_step)%60)
else:
pass
defkafkawritefile(time_step,time_num):
start=datetime.now()
downloader=KafkaDownloader(HOSTS,TOPIC,GROUP)
i=1
while(i<=time_num):
end_time=get_end_time(start.hour,start.minute,i*time_step)
end_time_file=end_time.replace(':','_')
outfile_path="/data/tmp/"+end_time_file+".csv"
ifos.path.exists(outfile_path):
os.remove(outfile_path)
writefile=open(outfile_path,'a+',encoding='utf-8')
formsgindownloader.message():
curr_time=datetime.now()
curr_time=str(curr_time)
split_curr_time=curr_time.split('')
curr_time_str=split_curr_time[1][0:5]
ifcurr_time_str>=str(end_time):
break
i+=1
if__name__=='__main__':
time_step=15
time_num=1
kafkawritefile(time_step,time_num)
以上这篇python每5分钟从kafka中提取数据的例子就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持毛票票。
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。