数据丢失生产者-> 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')

Nifi Processor属性的屏幕截图：

Kafka包含3个节点，主题复制因子-3

也许问题在于生产者的写入速度比消费者读取的速度快，并且当某个块溢出时数据被删除了吗？

请给我建议。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）