#python #apache-kafka #apache-nifi #confluent-kafka-python
#python #apache-kafka #apache-nifi #confluent-kafka-python
Вопрос:
Я пишу сообщения в kafka из CSV-файлов. Мой производитель говорит, что все данные, созданные для темы Kafka.
Наряду с этим, я использую apache nifi в качестве потребителя для темы kafka (процессор ConsumeKafka_2_0).
Если бы я создавал данные для kafka в одном потоке — все в порядке, но если бы я попытался использовать multiple producer с несколькими файлами параллельно, я потерял бы много строк.
Суть моего производителя следующая:
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:
Kafka содержит 3 узла, коэффициент репликации темы — 3
Может быть, проблема в том, что производитель записывает быстрее, чем читает потребитель, и данные удаляются при переполнении определенного блока?
Пожалуйста, дай мне совет.
Комментарии:
1. Пожалуйста, поделитесь конфигурациями темы. Также вы используете acks в конфигурации производителя. Я не вижу этого в вашем фрагменте кода.
2. Также поделитесь спецификацией nifi и расскажите о соображениях, которые вы сделали в flow для параллелизма. Сколько потоковых файлов перемещается по системе, когда вы испытываете потерю данных и т.д..
3. Тема: количество разделов транзакций: 1 Фактор репликации: 3 Конфигурации: Тема: раздел транзакций: 0 Лидер: 2 Реплики: 2,3,1 Isr: 3,2,1
4. Я только что добавил acks в producer, протестирую его
5. В моем случае параллелизм заключается в запуске производителя для набора файлов. В nifi я читаю из раздела kafka только с одним разделом, после этого я преобразую сообщения в один ORC-файл и сохраняю его в hive.