如何解决数据丢失生产者-> Kafka-> Nifi
我从csv文件向kafka写消息。我的生产者说所有数据都以Kafka为主题。
此外,我使用apache nifi作为kafka主题(ConsumeKafka_2_0处理器)的使用者。
如果我将卡夫卡的数据生成到一个流中-一切正常,但是如果我尝试使用多个文件并行的多个生产者,则会丢失很多行。
我的制作人的核心如下:
def produce(for_produce_file):
log.info('Producing started')
avro_schema = avro.schema.Parse(open(config.AVRO_SCHEME_FILE).read())
names_from_schema = [field.name for field in avro_schema.fields]
producer = Producer({'bootstrap.servers': config.KAFKA_BROKERS,'security.protocol': config.SECURITY_PROTOCOL,'ssl.ca.location': config.SSL_CAFILE,'sasl.mechanism': config.SASL_MECHANISM,'sasl.username': config.SASL_PLAIN_USERNAME,'sasl.password': config.SASL_PLAIN_PASSWORD,'queue.buffering.max.messages': 1000000,'queue.buffering.max.ms': 5000})
try:
file = open(for_produce_file,'r',encoding='utf-8')
except FileNotFoundError:
log.error(f'File {for_produce_file} not found')
else:
produced_str_count = 0
csv_reader = csv.DictReader(file,delimiter="|",fieldnames=names_from_schema)
log.info(f'File {for_produce_file} opened')
for row in csv_reader:
record = dict(zip(names_from_schema,row.values()))
while True:
try:
producer.produce(config.TOPIC_NAME,json.dumps(record).encode('utf8'))
producer.poll(0)
produced_str_count += 1
break
except BufferError as e:
log.info(e)
producer.poll(5)
except KafkaException as e:
log.error('Kafka error: {},message was {} bytes'.format(e,sys.getsizeof(json.dumps(record))))
log.error(row)
break
producer.flush()
log.info(f'Producing ended. Produced {produced_str_count} rows')
Kafka包含3个节点,主题复制因子-3
也许问题在于生产者的写入速度比消费者读取的速度快,并且当某个块溢出时数据被删除了吗?
请给我建议。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。