1

我想从名为customers 的SQL 数据库中的“niftest”表中提取所有记录。我正在附上模板以供您的帮助。我面临以下问题：一旦我运行“QueryDatabaseTable”处理器，我会在队列中获取流文件，但是当我清空队列并尝试重新运行“QueryDatabaseTable”处理器时，我无法获取队列中的记录。我该如何解决这个问题？

该流程的整个想法是从 mySql 数据库中提取记录并将文件本地存储在我的桌面上。

我已经使用以下网站上的模板来实现这一点，并根据我的要求更改了我的表名和数据库连接池服务：-

https://www.batchiq.com/database-extract-with-nifi.html

此外，在合并内容之后，我将文件作为文件类型获取。使用哪个处理器将合并文件存储为 CSV 文件类型（所有记录都存储为 Test.csv 而不是文件类型）我使用“UpdateAttribute”处理器并将属性“文件名”和值添加为“Test.csv” '但在输出文件中我只看到一条记录而不是所有记录。如何解决这个问题？

谢谢！

4

1 回答 1

2

QueryDatabaseTable将在运行时存储状态，并在再次运行时提取only the incremental records已添加到表中的状态。

要从表中获取所有记录，您需要清除处理器的状态，然后才能从表中获取所有记录。

如何清除 QueryDatabaseTable 处理器的状态？

 1. Stop Query Database processor //make sure no threads are running at top right corner

 2. Right Click on the processor

 3. Go to View state tab

 4. Click on clear state //this will clear all the stored state in the processor

在此处输入图像描述有关清除 QueryDatabase 表处理器状态的更多详细信息， 请参阅此链接

-> 另一个问题是MergeContent 处理器作为

最小条目数设置为100，因此处理器将等到您在 MergeContent 处理器之前的队列中有 100 个流文件。

使用Max Bin Age 属性值为1 min ..etc 以便处理器将强制合并流文件并将合并的流文件转换为合并关系。

有关 MergeContent 处理器使用/配置的更多详细信息，请参阅此链接。

更新：

1.

如果要将最终文件格式转换为 csv，则保存到本地路径。

然后你可以在不需要转换后使用ConvertRecord处理器QueryDatabaseTableAvro --> Json --> Csv

将 ConvertRecord处理器配置为Avro Reader和CsvRecordSetWriter控制器服务，然后处理器读取 Avro 数据然后转换为 Csv 格式。

要更改流文件的文件名，请使用UpdateAttribute处理器将所需的文件名作为新属性添加到处理器。

流动：

  - QueryDatabaseTable //get data from source in avro format 
  - ConvertRecord //convert Avro format to CSV format
  - UpdateAttribute //change filename
  - PutFile //store the csv file into local

2.

如果您担心数据在输出流文件中存储为一行，则将合并内容处理器配置为

Delimiter Strategy Text

Demarcator shift+enter

有关此合并内容分隔符策略配置的更多详细信息，请参阅此链接。

使用此配置，我们不会更改数据的格式（数据仍将在 json 中），但我们在每个 json 记录后添加新行。

于 2018-09-26T16:40:48.250 回答