В предыдущих частях мы рассмотрели вопросы мониторинга потоков данных и состояния системы средствами GUI NiFi и задач отчетности. В этом материале поближе познакомимся с задачами отчетности Site-to-Site. При отправке данных из одного экземпляра NiFi в другой можно использовать множество различных протоколов, однако, предпочтительным является NiFi Site-to-Site. Данный протокол предлагает безопасную и эффективную передачу данных из узлов в одном экземпляре NiFi, производящем данные, на узлы в другом экземпляре, являющимся приемником этих данных.

1. SiteToSiteBulletinReportingTask – публикует события об ошибках и предупреждениях в бюллетенях с использованием протокола Site-to-Site;

2. SiteToSiteProvenanceReportingTask – публикует события о происхождении данных, используя протокол Site-to-Site;

3. SiteToSiteStatusReportingTask – публикует события состояния, используя протокол Site-to-Site;

4. SiteToSiteMetricsReportingTask – публикует те же метрики, что и задача Ambari Reporting, используя протокол Site-to-Site.

В NiFi есть возможность использовать Site-to-Site (S2S), а это означает, что задачи отчетности постоянно выполняются для сбора данных и их отправки в удаленный экземпляр NiFi. Аналогично можно использовать S2S для отправки данных в экземпляр, на котором выполняется отчетность.

Таким образом, используя входной порт на рабочей области, вы можете фактически получать данные, сгенерированные задачей создания отчетов, и применять их в рабочем процессе NiFi. Эта функциональность позволяет использовать все возможности NiFi для обработки метрик задач отчетности.

Настройка Site-to-Site

Прежде чем можно будет использовать S2S, необходимо установить следующие свойства в файле nifi.properties на всех узлах в вашем кластере NiFi:

1)     Свойства с именем nifi.remote.input.socket.* зависят от транспортного протокола RAW;

2)     Свойства nifi.remote.input.http.* являются специфическими свойствами транспортного протокола HTTP.

Чтобы эти изменения вступили в силу, потребуется перезапуск ваших экземпляров NiFi.

SiteToSiteBulletinReportingTask

Публикует события бюллетеня, используя протокол Site-to-Site. Для каждого компонента сохраняется пять-десять сводок на уровне контроллера продолжительностью до пяти минут. Если отчетная задача запланирована недостаточно часто, некоторые бюллетени могут не отправляться.

Пользователь может определить Record Writer и напрямую указать выходной формат и формат данных, используя следующую входную схему:

{
  "type" : "record",
  "name" : "bulletins",
  "namespace" : "bulletins",
  "fields" : [ 
	{ "name" : "objectId", "type" : "string" },
	{ "name" : "platform", "type" : "string" },
	{ "name" : "bulletinId", "type" : "long" },
	{ "name" : "bulletinCategory", "type" : ["string", "null"] },
	{ "name" : "bulletinGroupId", "type" : ["string", "null"] },
	{ "name" : "bulletinGroupName", "type" : ["string", "null"] },
	{ "name" : "bulletinGroupPath", "type" : ["string", "null"] },
	{ "name" : "bulletinLevel", "type" : ["string", "null"] },
	{ "name" : "bulletinMessage", "type" : ["string", "null"] },
	{ "name" : "bulletinNodeAddress", "type" : ["string", "null"] },
	{ "name" : "bulletinNodeId", "type" : ["string", "null"] },
	{ "name" : "bulletinSourceId", "type" : ["string", "null"] },
	{ "name" : "bulletinSourceName", "type" : ["string", "null"] },
	{ "name" : "bulletinSourceType", "type" : ["string", "null"] },
	{ "name" : "bulletinTimestamp", "type" : ["string", "null"], "doc" : "Format: yyyy-MM-dd'T'HH:mm:ss.SSS'Z'" }
  ]
}

Для хранения статистики по ошибкам создадим таблицу в БД PostgreSQL:

CREATE TABLE nifi_bulletin_err_log (
  id                 BIGSERIAL
, flowId             VARCHAR(36)  
, bulletinGroupId    VARCHAR(36)
, bulletinGroupName  VARCHAR(255)
, bulletinNodeId     VARCHAR(36)
, bulletinSourceId   VARCHAR(36)
, bulletinSourceName VARCHAR(255)  
, bulletinSourceType VARCHAR(50)
, bulletinMessage    TEXT
, bulletinTimestamp  VARCHAR(30)
, logTimestamp       TIMESTAMP DEFAULT LOCALTIMESTAMP(0)
, CONSTRAINT nifi_bulletin_err_log_pkey PRIMARY KEY (id));

И настроим блок обработки и записи метрик, получаемых на удаленный порт NiFi:

В Grafana создадим источник для работы с базой PostgreSQL:

И на основе созданной таблицы построим дашборд для мониторинга ошибок NiFi в Grafana:

SiteToSiteProvenanceReportingTask

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

Рекомендуется отправлять данные о происхождении в экземпляр NiFi, отличный от того, на котором выполняется задача создания отчетов SiteToSiteProvenanceReportingTask, потому что, когда данные получены через Site-to-Site и обработаны, это само по себе будет генерировать Provenance события. В результате создается цикл. Необходимо учесть, что данные отправляются пакетами (по умолчанию – 1000). Это означает, что для каждой партии событий Provenance, которые отправляются обратно в NiFi, получатель NiFi должен будет генерировать только одно событие для каждого компонента.

По умолчанию при публикации в экземпляре NiFi данные Provenance отправляются в виде массива JSON. Однако, пользователь может определить Record Writer и напрямую указать выходной формат и данные, предполагая, что входная схема определена следующим образом:

{
  "type" : "record",
  "name" : "provenance",
  "namespace" : "provenance",
  "fields": [
    { "name": "eventId", "type": "string" },
    { "name": "eventOrdinal", "type": "long" },
    { "name": "eventType", "type": "string" },
    { "name": "timestampMillis", "type": "long" },
    { "name": "durationMillis", "type": "long" },
    { "name": "lineageStart", "type": { "type": "long", "logicalType": "timestamp-millis" } },
    { "name": "details", "type": ["null", "string"] },
    { "name": "componentId", "type": ["null", "string"] },
    { "name": "componentType", "type": ["null", "string"] },
    { "name": "componentName", "type": ["null", "string"] },
    { "name": "processGroupId", "type": ["null", "string"] },
    { "name": "processGroupName", "type": ["null", "string"] },
    { "name": "entityId", "type": ["null", "string"] },
    { "name": "entityType", "type": ["null", "string"] },
    { "name": "entitySize", "type": ["null", "long"] },
    { "name": "previousEntitySize", "type": ["null", "long"] },
    { "name": "updatedAttributes", "type": { "type": "map", "values": "string" } },
    { "name": "previousAttributes", "type": { "type": "map", "values": "string" } },
    { "name": "actorHostname", "type": ["null", "string"] },
    { "name": "contentURI", "type": ["null", "string"] },
    { "name": "previousContentURI", "type": ["null", "string"] },
    { "name": "parentIds", "type": { "type": "array", "items": "string" } },
    { "name": "childIds", "type": { "type": "array", "items": "string" } },
    { "name": "platform", "type": "string" },
    { "name": "application", "type": "string" },
    { "name": "remoteIdentifier", "type": ["null", "string"] },
    { "name": "alternateIdentifier", "type": ["null", "string"] },
    { "name": "transitUri", "type": ["null", "string"] }
  ]
}

В задаче отчетности можно настроить следующие свойства:

SiteToSiteStatusReportingTask

Публикует события состояния, используя протокол Site-to-Site.

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

В задаче отчетности можно настроить следующие свойства:

Пользователь может определить Record Writer, напрямую указать формат вывода и данные, предполагая, что схема ввода следующая:

{
  "type" : "record",
  "name" : "status",
  "namespace" : "status",
  "fields" : [
    // common fields for all components
	{ "name" : "statusId", "type" : "string"},
	{ "name" : "timestampMillis", "type": { "type": "long", "logicalType": "timestamp-millis" } },
	{ "name" : "timestamp", "type" : "string"},
	{ "name" : "actorHostname", "type" : "string"},
	{ "name" : "componentType", "type" : "string"},
	{ "name" : "componentName", "type" : "string"},
	{ "name" : "parentId", "type" : ["string", "null"]},
	{ "name" : "parentName", "type" : ["string", "null"]},
	{ "name" : "parentPath", "type" : ["string", "null"]},
	{ "name" : "platform", "type" : "string"},
	{ "name" : "application", "type" : "string"},
	{ "name" : "componentId", "type" : "string"},
	
	// PG + RPG + Ports + Processors
	{ "name" : "activeThreadCount", "type" : ["long", "null"]},
	
	// PG + Ports + Processors
	{ "name" : "flowFilesReceived", "type" : ["long", "null"]},
	{ "name" : "flowFilesSent", "type" : ["long", "null"]},
	
	// PG + Ports + Processors
	{ "name" : "bytesReceived", "type" : ["long", "null"]},
	{ "name" : "bytesSent", "type" : ["long", "null"]},
	
	// PG + Connections
	{ "name" : "queuedCount", "type" : ["long", "null"]},
	
	// PG + Processors
	{ "name" : "bytesRead", "type" : ["long", "null"]},
	{ "name" : "bytesWritten", "type" : ["long", "null"]},
	{ "name" : "terminatedThreadCount", "type" : ["long", "null"]},

	// Processors + Ports
	{ "name" : "runStatus", "type" : ["string", "null"]},

	// fields for process group status
	{ "name" : "bytesTransferred", "type" : ["long", "null"]},
	{ "name" : "flowFilesTransferred", "type" : ["long", "null"]},
	{ "name" : "inputContentSize", "type" : ["long", "null"]},
	{ "name" : "outputContentSize", "type" : ["long", "null"]},
	{ "name" : "queuedContentSize", "type" : ["long", "null"]},
	{ "name" : "versionedFlowState", "type" : ["string", "null"]},
	
	// fields for remote process groups
	{ "name" : "activeRemotePortCount", "type" : ["long", "null"]},
	{ "name" : "inactiveRemotePortCount", "type" : ["long", "null"]},
	{ "name" : "receivedContentSize", "type" : ["long", "null"]},
	{ "name" : "receivedCount", "type" : ["long", "null"]},
	{ "name" : "sentContentSize", "type" : ["long", "null"]},
	{ "name" : "sentCount", "type" : ["long", "null"]},
	{ "name" : "averageLineageDuration", "type" : ["long", "null"]},
	{ "name" : "transmissionStatus", "type" : ["string", "null"]},
	{ "name" : "targetURI", "type" : ["string", "null"]},
	
	// fields for input/output ports + connections + PG
	{ "name" : "inputBytes", "type" : ["long", "null"]},
	{ "name" : "inputCount", "type" : ["long", "null"]},
	{ "name" : "outputBytes", "type" : ["long", "null"]},
	{ "name" : "outputCount", "type" : ["long", "null"]},
	{ "name" : "transmitting", "type" : ["boolean", "null"]},
	
	// fields for connections
	{ "name" : "sourceId", "type" : ["string", "null"]},
	{ "name" : "sourceName", "type" : ["string", "null"]},
	{ "name" : "destinationId", "type" : ["string", "null"]},
	{ "name" : "destinationName", "type" : ["string", "null"]},
	{ "name" : "maxQueuedBytes", "type" : ["long", "null"]},
	{ "name" : "maxQueuedCount", "type" : ["long", "null"]},
	{ "name" : "queuedBytes", "type" : ["long", "null"]},
	{ "name" : "backPressureBytesThreshold", "type" : ["long", "null"]},
	{ "name" : "backPressureObjectThreshold", "type" : ["long", "null"]},
	{ "name" : "backPressureDataSizeThreshold", "type" : ["string", "null"]},
	{ "name" : "isBackPressureEnabled", "type" : ["string", "null"]},
	
    // fields for processors
	{ "name" : "processorType", "type" : ["string", "null"]},
	{ "name" : "averageLineageDurationMS", "type" : ["long", "null"]},
	{ "name" : "flowFilesRemoved", "type" : ["long", "null"]},
	{ "name" : "invocations", "type" : ["long", "null"]},
	{ "name" : "processingNanos", "type" : ["long", "null"]},
	{ "name" : "executionNode", "type" : ["string", "null"]},
	{ "name" : "counters", "type": ["null", { "type": "map", "values": "string" }]}
  ]
}

SiteToSiteMetricsReportingTask

Публикует те же метрики, что и задача Ambari Reporting, используя протокол Site-To-Site:

Есть два доступных формата вывода. Первый — это формат Ambari, определенный в API-интерфейсе сборщика метрик Ambari, который представляет собой JSON с динамическими ключами. При использовании этого формата вас может заинтересовать приведенная ниже спецификация Jolt для преобразования данных.

Формат Ambari

			[
			  {
			    "operation": "shift",
			    "spec": {
			      "metrics": {
			        "*": {
			          "metrics": {
			            "*": {
			              "$": "metrics.[#4].metrics.time",
			              "@": "metrics.[#4].metrics.value"
			            }
			          },
			          "*": "metrics.[&1].&"
			        }
			      }
			    }
			  }
			]

Схема преобразует приведенный ниже образец:

{
				"metrics": [{
					"metricname": "jvm.gc.time.G1OldGeneration",
					"appid": "nifi",
					"instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
					"hostname": "localhost",
					"timestamp": "1520456854361",
					"starttime": "1520456854361",
					"metrics": {
						"1520456854361": "0"
					}
				}, {
					"metricname": "jvm.thread_states.terminated",
					"appid": "nifi",
					"instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
					"hostname": "localhost",
					"timestamp": "1520456854361",
					"starttime": "1520456854361",
					"metrics": {
						"1520456854361": "0"
					}
				}]
			}

к следующему виду:

{
				"metrics": [{
					"metricname": "jvm.gc.time.G1OldGeneration",
					"appid": "nifi",
					"instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
					"hostname": "localhost",
					"timestamp": "1520456854361",
					"starttime": "1520456854361",
					"metrics": {
						"time": "1520456854361",
						"value": "0"
					}
				}, {
					"metricname": "jvm.thread_states.terminated",
					"appid": "nifi",
					"instanceid": "8927f4c0-0160-1000-597a-ea764ccd81a7",
					"hostname": "localhost",
					"timestamp": "1520456854361",
					"starttime": "1520456854361",
					"metrics": {
						"time": "1520456854361",
						"value": "0"
					}
				}]
			}

Record format

Второй формат использует структуру записи NiFi, чтобы пользователь мог определить Record Writer и напрямую указать формат вывода и данные, предполагая, что схема ввода следующая:

{
			  "type" : "record",
			  "name" : "metrics",
			  "namespace" : "metrics",
			  "fields" : [ 
				{ "name" : "appid", "type" : "string" },
				{ "name" : "instanceid", "type" : "string" },
				{ "name" : "hostname", "type" : "string" },
				{ "name" : "timestamp", "type" : "long" },
				{ "name" : "loadAverage1min", "type" : "double" },
				{ "name" : "availableCores", "type" : "int" },
				{ "name" : "FlowFilesReceivedLast5Minutes", "type" : "int" },
				{ "name" : "BytesReceivedLast5Minutes", "type" : "long" },
				{ "name" : "FlowFilesSentLast5Minutes", "type" : "int" },
				{ "name" : "BytesSentLast5Minutes", "type" : "long" },
				{ "name" : "FlowFilesQueued", "type" : "int" },
				{ "name" : "BytesQueued", "type" : "long" },
				{ "name" : "BytesReadLast5Minutes", "type" : "long" },
				{ "name" : "BytesWrittenLast5Minutes", "type" : "long" },
				{ "name" : "ActiveThreads", "type" : "int" },
				{ "name" : "TotalTaskDurationSeconds", "type" : "long" },
				{ "name" : "TotalTaskDurationNanoSeconds", "type" : "long" },
				{ "name" : "jvmuptime", "type" : "long" },
				{ "name" : "jvmheap_used", "type" : "double" },
				{ "name" : "jvmheap_usage", "type" : "double" },
				{ "name" : "jvmnon_heap_usage", "type" : "double" },
				{ "name" : "jvmthread_statesrunnable", "type" : ["int", "null"] },
				{ "name" : "jvmthread_statesblocked", "type" : ["int", "null"] },
				{ "name" : "jvmthread_statestimed_waiting", "type" : ["int", "null"] },
				{ "name" : "jvmthread_statesterminated", "type" : ["int", "null"] },
				{ "name" : "jvmthread_count", "type" : "int" },
				{ "name" : "jvmdaemon_thread_count", "type" : "int" },
				{ "name" : "jvmfile_descriptor_usage", "type" : "double" },
				{ "name" : "jvmgcruns", "type" : ["long", "null"] },
				{ "name" : "jvmgctime", "type" : ["long", "null"] }
			  ]
			}

Заключение

В этой статье мы познакомили вас с задачами отчетности Site-to-Site. Это удобный и гибкий механизм, позволяющий организовывать постоянный сбор данных и их отправку как в удаленный экземпляр NiFi, так и в экземпляр, выполняющий данную задачу отчетности. Таким образом, используя входной порт в рабочей области NiFi, вы можете фактически получать данные, сгенерированные задачей создания отчетов и использовать их в рабочем процессе NiFi. Это позволяет вам применять все возможности NiFi для обработки собранных метрик и дальнейшего использования результатов для визуализации и анализа.