​Импортируйте данные журнала Json в реальном времени в Hive.

Случай: используйте NiFi для импорта файлов журналов типа JSON, созданных в определенном каталоге, в Hive. Здесь данные сначала анализируются в атрибуты данных Json через NiFi, а затем вручную задается формат данных и данные импортируются в HDFS. Hive устанавливает путь сопоставления таблиц для импорта внешних данных в Hive.

Используемые процессоры: «TailFile», «EvaluateJsonPath», «ReplaceText» и «PutHDFS».

1. Настройте процессор TailFile.

Процессор TailFile «отслеживает» файл или список файлов и принимает данные из файла по мере их записи в файл. Отслеживаемые файлы имеют текстовый формат, и данные принимаются при записи новых строк. Если файл, который вы хотите отслеживать, регулярно «переворачивается» (что часто случается с файлами журналов), вы можете использовать дополнительный «Шаблон смены имен файлов» для извлечения данных из свернутого файла. Прокрутка происходит, когда NiFi не работает. Файлы по-прежнему будут отслеживаться после перезапуска NiFi. Рекомендуется установить план работы на несколько секунд и не использовать для запуска 0 секунд по умолчанию, иначе этот процессор будет потреблять много ресурсов. Этот процессор не поддерживает мониторинг сжатых файлов.

Описание конфигурации «Свойства» процессора «TailFile» следующее:

Шаги настройки следующие:

1. Создайте процессор TailFile.

​

2. Настройте «СВОЙСТВА»

​

Уведомление:Вышеупомянутое должно быть вNiFiСоздается на каждом узле кластера“/root/test/jsonfile”документ,«jsonfile» — это файл,а не каталог.

2. Настройте процессор EvaluateJsonPath.

Процессор EvaluateJsonPath оценивает одно или несколько выражений JsonPath на основе содержимого FlowFile. В зависимости от конфигурации процессора результаты этих выражений присваиваются свойствам FlowFile или записываются в содержимое самого FlowFile.

Введите Jsonpath, добавив определенные пользователем атрибуты. Имя добавленного атрибута сопоставляется с именем атрибута в выходном потоке. Значение атрибута должно быть допустимым выражением JsonPath (например: $.name). Тип возвращаемого значения «автоопределение» будет определяться на основе настроенной цели. Если для параметра «Destination» установлено значение «flowfile-attribute», будет использоваться «скалярный» тип возвращаемого значения. Если для параметра «Destination» установлено значение «flowfile-content», будет использоваться тип возвращаемого значения «JSON».

Если JsonPath оценивается как массив JSON или объект JSON и для типа возвращаемого значения установлено значение «скалярный», файл потока не будет изменен и будет перенаправлен на сбой. Тип возвращаемого значения JSON может возвращать «скаляр», если предоставленный JsonPath оценивается как указанное значение. Если целью является «flowfile-content» и JsonPath не вычисляет соответствующее значение, то файл потока будет перенаправлен в «unmatched» без изменения его содержимого. Если целью является «flowfile-attribute» и выражение ничему не соответствует, то в качестве значения атрибута будет использоваться пустая строка, а FlowFile всегда будет перенаправляться на «соответствующий».

Описание конфигурации «Свойства» процессора «EvaluateJsonPath» следующее:

Пример описания:

• Извлечь содержимое файла потока в формате JSON,Как свойство выходного потока. (Уведомление: Когда выходные данные выбирают атрибут flowfile,Даже если jsonpath не соответствует значению,Потоковые файлы также будут перенаправлены на соответствующие)

Введите json следующим образом:

​

Вывод следующий:

• Извлечь содержимое файла потока в формате JSON,как содержимое выходного потока. (Уведомление: если выбрано содержимое потокового файла,Пользователи могут добавить только один пользовательский атрибут, если jsonPath не соответствует;,будет перенаправлено на непревзойденный)

​

Содержимое выходного потока:

​

После ознакомления с тем, как использовать «EvaluateJsonPath»,Далее для настройки,Шаги настройки следующие:

1. Создайте процессор EvaluateJsonPath.

​

2. Настройте «СВОЙСТВА»

​

3. Соедините процессор TailFile и процессор EvaluateJsonPath.

​

3. Настройте процессор «ReplaceText».

Процессор «ReplaceText» заменит содержимое FlowFile, соответствующее регулярному выражению, и создаст новое содержимое FlowFile. Здесь мы используем процессор «ReplaceText» для замены каждого содержимого FlowFile, обработанного предыдущим процессором «EvaluateJsonPath», настраиваемым содержимым. Здесь пользовательское содержимое — это значение, полученное из свойств FlowFile в соответствии с «\t». Разделяйте их символами табуляции. для облегчения последующего хранения в HDFS и сопоставления таблиц Hive.

Шаги настройки следующие:

1. Создайте новый процессор «ReplaceText» и настройте процессор «PROPERTIES».

​

2. Соедините процессор «EvaluateJsonPath» и процессор «ReplaceText» и установите

​

Установите отношение доставки между «сбой» и «несоответствие» в «EvaluateJsonPath» для автоматического прекращения:

​

4. Настройте процессор «PutHDFS».

Здесь создается процессор «PutHDFS» для записи данных, обработанных восходящим потоком, в каталог HDFS. Шаги настройки следующие:

1. Создайте и настройте процессор «PutHDFS».

​

2. Соедините процессор «ReplaceText» и процессор «PutHDFS» и настройте

​

Настройте отношение передачи FlowFile при «сбое» процессора «ReplaceText» для автоматического завершения:

​

Настройте отношения доставки FlowFile для процессоров «PutHDFS» «сбой» и «успех» для автоматического завершения:

​

5. Запустите тест

1. Создайте внешний вид personinfo в Hive. Создайте внешний вид personinfo в Hive.

CREATE TABLE personinfo(
id int,
name string,
age int
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
LOCATION '/mycluster/personinfo'

2. Запустите процесс обработки данных NiFi и обработайте данные.

Запишите следующие данные в любой файл node/root/test/jsonfile NiFi:

echo "{\"id\":1,\"name\":\"zhangsan\",\"age\":18}" >> /root/test/jsonfile
echo "{\"id\":2,\"name\":\"lisi\",\"age\":19}"	>> /root/test/jsonfile
echo "{\"id\":3,\"name\":\"wangwu\",\"age\":20}"	>> /root/test/jsonfile
echo "{\"id\":4,\"name\":\"maliu\",\"age\":21}"	>> /root/test/jsonfile
echo "{\"id\":5,\"name\":\"tianqi\",\"age\":22}"	>> /root/test/jsonfile

3. Просмотрите результаты

Страница Ни Фи:

Результаты в улье:

вопрос:Когда мы идем к кому-то сразуNiFiУзловой“/root/test/jsonfile”документсередина При записи данных,В настоящее время «EvaluateJsonPath» будет содержать несколько фрагментов данных JSON в FlowFile.,При получении атрибутов json,Будут получены только атрибуты, соответствующие первому json. Когда данные передаются в нижестоящий процессор «ReplaceText»,Поскольку каждая строка настроена на замену строкой в ​​указанном формате.,В это время все строки данных в этом пакете будут заменены первой строкой данных формата JSON. Как показано ниже:

При одновременном вводе нескольких фрагментов данных в хвостовой файл мы не хотим, чтобы все строки json заменялись первой строкой json. Тогда данные, обработанные процессором TailFile, можно напрямую передать в ConvertRecord. " процессор, а данные обрабатываются с помощью json. Формат преобразуется в данные пользовательского текстового формата, а затем передается процессору "PutHDFS". Таким образом, чтобы решить вышеуказанные проблемы, мы можем повторно использовать предыдущие процессоры "TailFile" и "PutHDFS". здесь. Далее нам нужно настроить только "ConvertRecord" Только процессор.

6. Настройте процессор ConvertRecord.

«ConvertRecord» преобразует записи из одного формата данных в другой на основе настроенного устройства чтения записей и контроллера записи записей.

Описание конфигурации «Свойства» процессора «ConvertRecord» следующее:

Шаги настройки следующие:

1. Создайте процессор ConvertRecord.

2. Настройте «СВОЙСТВА»середина“Record Конфигурация «Читатель»

​

После нажатия выше введите конфигурацию:

После настройки вышеперечисленного необходимо приступить к:

3. Поместите «Записывающее устройство» в «СВОЙСТВА»:

Нажмите выше, чтобы войти в конфигурацию:

4. Соедините процессор TailFile и процессор ConvertRecord.

5. Соедините процессор «ConvertRecord» и процессор «PutHDFS».

В то же время установите отношение «сбой» обработки «ConvertRecord» для автоматического завершения:

7. Запустите тест

Удалите исходный путь «/personinfo» в HDFS, запустите процесс обработки данных NiFi и обработайте данные:

Запишите следующие данные в любой узел кластера NiFi «/root/test/jsonfile» за один раз:

echo "{\"id\":1,\"name\":\"zhangsan\",\"age\":18}" >> /root/test/jsonfile
echo "{\"id\":2,\"name\":\"lisi\",\"age\":19}"	>> /root/test/jsonfile
echo "{\"id\":3,\"name\":\"wangwu\",\"age\":20}"	>> /root/test/jsonfile
echo "{\"id\":4,\"name\":\"maliu\",\"age\":21}"	>> /root/test/jsonfile
echo "{\"id\":5,\"name\":\"tianqi\",\"age\":22}"	>> /root/test/jsonfile

Запросить данные personinfo таблицы Hive: