Nifi - усечение и загрузка в таблицу mysql db

Я читаю файлы CSV с сайта SFTP и загружаю их в mysql db с помощью Nifi.

У меня есть следующий рабочий процесс, который, похоже, работает нормально. Мне просто нужна помощь в выяснении того, как обрезать таблицу, прежде чем я начну с загрузкой данных.

Nifi Flow:

ListSFTP -> FetchSFTP -> InferAvroSchema -> ConvertCSVtoAvro -> ConvertAvrotoJSON -> SplitJSON -> ConvertJSONtoSQL -> PutSQL

Кажется, что этот поток работает нормально, но каждый раз, когда я его запускаю, мне нужно сначала усечь таблицу, а затем начать загрузку.

Может ли кто-нибудь помочь мне с некоторой информацией о том, как я могу этого добиться. Или есть лучший поток, чем то, что я написал, посоветуйте, пожалуйста.

Спасибо, Аадил


mysql data-ingestion load apache-nifi truncate
person Aadil M    schedule 29.12.2017    source источник

Ответы (1)


arrow_upward
3
arrow_downward

Прежде чем я перейду к части «обрезать перед вставкой», я рекомендую заменить все, от ConvertCSVtoAvro до PutSQL, на PutDatabaseRecord. По сути, он выполняет «ConvertXtoSQL-> PutSQL» вместе и устраняет необходимость во всех этих преобразованиях только для выполнения операторов SQL.

Усечь перед вставкой немного сложнее. Начиная с версии NiFi 1.5.0 (еще не выпущенной на момент написания этой статьи) через NIFI- 4522, вы сможете сохранить содержимое файла потока при выполнении команды SQL. Это означает, что до PutDatabaseRecord вы могли бы иметь PutSQL с параметром «SQL Statement», установленным на «TRUNCATE myTable». PutSQL выдаст TRUNCATE, но передаст файл потока.

На данный момент единственный обходной путь, который я могу придумать, - это процессор ExecuteScript. Если вы используете PutDatabaseRecord, вам, вероятно, придется получить DBCPConnectionPool по имени (см. мое сообщение в блоге в качестве примера) и выполните оператор TRUNCATE самостоятельно, а затем передайте файл входящего потока. Если вы используете приведенный выше поток с PutSQL, вы можете получить файл входящего потока, а затем записать файл потока «TRUNCATE myTable», ТОЛЬКО если fragment.index равен нулю и вы используете Prioritizer в соединении, затем передайте файл входящего потока.

person mattyb    schedule 29.12.2017
comment
Большое спасибо! Это просто отлично сработало. Сообщение в блоге о Execute script также полезно. - person Aadil M; 04.01.2018