В предыдущем блоге этой серии «Превращение потоков в продукты данных» мы говорили о растущей необходимости сократить задержку между созданием/приемом данных и генерировать аналитику и аналитические сведения на основе этих данных. Мы обсудили, как использовать Cloudera Streaming (CSA) с Apache Kafka и Apache Flink для обработки этих данных в режиме реального времени и в большом масштабе. В этом блоге мы покажем реальный пример того, как это сделать, и посмотрим, как мы можем использовать CSP для обнаружения мошенничества в режиме реального времени.

Для построения конвейеров данных потоковой аналитики в реальном времени требуется возможность обработки данных в потоках. Ключевым условием для потоковой обработки является возможность собирать и перемещать данные, генерируемые в источнике. Это то, что мы называем проблемой первой мили. Этот блог будет опубликован в двух частях. В первой части мы изучим Apache NiFi При поддержке Cloudera DataFlowКак получить это легко и эффективно、КонвертироватьидвигатьсяДанные для решения проблем первой мили,чтобы мы могли легко реализовать потоковую аналитикувариант использование. Мы также кратко обсудим Cloudera DataFlow облачный родной Kubernetes Преимущества запуска этого процесса при развертывании.

Во второй части мы рассмотрим, как запускать потоковую аналитику в реальном времени с помощью Apache Flink, а также воспользуемся графическим интерфейсом Cloudera SQL Stream Builder, чтобы легко создавать задания потоковой передачи, используя только язык SQL (кодирование Java/Scala не требуется). Мы также будем использовать информацию, полученную в ходе наших заданий потоковой аналитики, для предоставления различных последующих систем и информационных панелей.

вариант использования

Обнаружение мошенничества — вариант «Мы исследуем, время имеет решающее значение» использованияизхорошийизпример。Мы все сталкивались с такой ситуацией,А именно, информация о нашей кредитной карте или, как мы знаем, от людей, данные о карте были утекли.,А незаконные транзакции списываются с карты. Чтобы минимизировать потери в этом случае,Компании-эмитенты кредитных карт должны иметь возможность немедленно выявить потенциальное мошенничество.,чтобы он мог заблокировать кредитную карту и связаться с пользователем для проверки транзакции,И новая карта может быть выпущена для замены поврежденной кредитной карты.

Данные о транзакциях по картам обычно являются источником данных, управляемым событиями.,Новые данные будут появляться по мере совершения реальных покупок по картам. Но да,Помимо потоковой передачи данных,У нас еще есть традицииизхранение данных（база данные, хранилища «ключ-значение», хранилища объектов и т. д.), которые содержат данные, которые, возможно, придется использовать для обогащения потоковых данных. У нас извариант использованиясередина,Данные потоковой передачи не содержат данные учетной записи и пользователя.,Итак, нам нужно объединить потоки и справочные данные.,Чтобы получить всю информацию, нам необходимо проверить каждую потенциально мошенническую транзакцию.

В зависимости от дальнейшего использования сгенерированной информации нам может потребоваться хранить данные в разных форматах: создать список потенциально мошеннических транзакций для темы Kafka, чтобы система уведомлений могла немедленно принять меры и сохранить статистику в реляционном или оперативном формате; панель мониторинга для дальнейшего анализа или предоставления панелей мониторинга или сохранения необработанных потоков транзакций в долгосрочное хранилище для дальнейшего использования и дополнительного анализа.

В наших примерах в этом блоге будут использоваться функции Cloudera DataFlow и CDP для достижения следующих функций:

1. Apache NiFi в Cloudera DataFlow будет считывать поток транзакций, отправляемый по сети.
2. Для каждой транзакции NiFi вызывает производственную модель в Cloudera Machine Learning (CML), чтобы оценить потенциал мошенничества в транзакции.
3. Если показатель мошенничества превышает определенный порог, NiFi Немедленно уведомит систему подписки о маршрутизации транзакции. Kafka Тема, которая вызовет соответствующее действие.
4. Оцененные транзакции записываются в тему Kafka, которая будет предоставлять данные для процесса аналитики в реальном времени, работающего на Apache Flink.
5. Данные транзакций с оценками также сохраняются в базе данных Apache Kudu для последующего запроса и предоставления панели мониторинга мошенничества.
6. использовать SQL Stream Builder (SSB),насиспользоватьнепрерывныйпотоковая передача SQL для анализа потока транзакций и выявления потенциального мошенничества на основе географического местоположения покупок.
7. Выявленные мошеннические транзакции записываются в другой топик Kafka, который обеспечивает необходимые действия системе.
8. потоковая передача SQL Работа также экономит выявление мошенничества приезжать Kudu база данных。
9. от Kudu база В канале информационной панели dataiz отображается сводная статистика мошенничества.

Используйте Cloudera DataFlow

Apache NiFi да Cloudera DataFlow из Компонент, который упрощает работу над вариантами использованияполучать данные и внедрить необходимые для очистки трубопроводы, Конвертировать и обеспечить Потоковую рабочий процесс обработки. в силу 300 Благодаря нескольким готовым процессорам его можно использовать для распределения, извлечения и обработки данных общего назначения практически из любого типа источника или приемника.

В этом варианте использования, мы создали относительно простой из NiFi процесс, который реализует вышеуказанные шаги 1 приезжать 5 Все операции, которые мы более подробно опишем ниже.

В нашем извариантном использовании мы обрабатываем данные финансовых транзакций от внешних агентов. Агент будет отправлять сетевой адрес при каждой транзакции. Каждая транзакция содержит следующую информацию:

• временная метка транзакции
• Идентификатор связанной учетной записи
• Уникальный идентификатор транзакции
• Сумма транзакции
• Географические координаты (широта и долгота) места совершения транзакции

Сообщения о транзакциях имеют формат JSON, как показано в следующем примере:

{

  "ts": "2022-06-21 11:17:26",

  "account_id": "716",

  "transaction_id": "e933787c-f0ff-11ec-8cad-acde48001122",

  "amount": 1926,

  "lat": -35.40439536601375,

  "lon": 174.68080620053922

}

NiFi Возможность создания сетевых прослушивателей для получения данных из сети. В этом примере мы можем просто ListenUDP Перетаскивание процессора NiFi Canvas и настройте его с необходимыми портами. Конфигурацию процессора можно параметризовать, чтобы обеспечить возможность повторного использования потоков. В этом случае мы определяем функцию с именем #{input.udp.port} параметр, который позже мы сможем установить на тот порт, который нам нужен.

Описывать данные с помощью шаблонов

модельда Описать структуру данныхиздокумент。在环境серединаиз多个应用程序甚至 NiFi При отправке и получении данных между процессорами в потоке полезно иметь репозиторий, в котором централизованно управляется и хранится схема для всех различных типов данных. Это упрощает взаимодействие приложений друг с другом.

Cloudera Платформа данных (CDP) Поставляется с Schema Registry Служить. Для нашего примера вариант использования, мы сохранили данные транзакции из схемы в службе реестра схем и поместили наши NiFi Поток настроен на использование правильного имени схемы. Ни Фи и Schema Registry Интегрированный, он автоматически подключает место проживания для получения определений схемы, когда это необходимо, на протяжении всего процесса.

данные в NiFi Путь в потоке определяется визуальными связями между различными процессорами. Например, здесь ListenUDP 处理器先前接收приезжатьиз Данные“отметка”为нас要использоватьизмодельизимя：“дела”。

Скоринг и маршрутизация транзакций

Мы обучили и создали модель машинного обучения (ML) с использованием Cloudera Machine Learning (CML) для оценки каждой транзакции на основе ее потенциального мошенничества. CML предоставляет сервис с конечной точкой REST, которую мы можем использовать для оценки. Когда данные проходят через поток данных NiFi, мы хотим вызвать службу модели ML точки данных, чтобы получить оценку мошенничества для каждой точки данных.

Для этого мы используем LookupRecord от NiFi, который позволяет осуществлять поиск по сервисам REST. Ответ модели CML содержит оценку мошенничества, представленную действительным числом от 0 до 1.

LookupRecord Выход процессора, который содержит и ML Модель ответа объединяет необработанные данные о транзакциях, а затем объединяет их. NiFi Очень полезный процессор: QueryRecord. процессор.

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

В этом процессе мы определяем три SQL-запроса, которые будут одновременно выполняться в этом процессоре:

Обратите внимание, что некоторые процессоры также определяют дополнительные выходные данные, такие как «не удалось», «повторить попытку» и т. д., так что вы можете определить свою собственную логику обработки ошибок для процесса.

Отправлять потоки в другие системы

На этом этапе процесса мы использовали ML Модель оценки мошенничества обогащает наш поток и конвертирует его в соответствии с нашими последующими потребностями. После завершения приема данных оставшиеся данные будут отправлены в службу проживания. Kafka, мы будем использовать его, чтобы обеспечить процесс анализа в реальном времени и сохранить транзакции. Kudu таблицу, которую мы будем использовать позже для предоставления наших информационных панелей, а также других процессов анализа не в реальном времени.

Apache Kafka и Apache Kudu такжеда CDP часть конфигурации Kafka и Kudu Очень просто поручить работу за нас конкретному процессору.

Запускайте потоки данных локально в облаке

После того, как вы создадите процесс NiFi, его можно будет выполнить в любом развертывании NiFi, которое у вас может быть. Cloudera DataFlow для публичного облака (CDF-PC) предоставляет облачную среду выполнения эластичных потоков для эффективного управления потоками.

NiFi с фиксированным размером. По сравнению с кластеризацией, облачные родные потоки CDF имеют множество преимуществ при запуске:

• Вам не нужно управлять NiFi кластер. Вы можете просто подключиться приезжать CDF консоль, загрузите определение потока и выполните его. необходимый NiFi Служба автоматически создается как Kubernetes Служба для выполнения процесса, прозрачная для пользователей.
• Это обеспечивает лучшую изоляцию ресурсов между потоками.
• Потоковое выполнение может автоматически масштабироваться вверх и вниз.,Чтобы гарантировать наличие соответствующих ресурсов для обработки объема данных, обрабатываемых в данный момент. Это позволяет избежать нехватки ресурсов,И экономьте затраты за счет перераспределения ненужных ресурсов, когда они больше не нужны.
• Встроенный мониторинг с определяемыми пользователем KPI можно настроить для каждого конкретного потока с различной степенью детализации (система, поток, процессор, соединение и т. д.).

Безопасное входящее соединение

В дополнение к вышесказанному,Настройка конечной точки защищенной сети в качестве входного шлюза. Общеизвестно сложная проблема, которую можно решить в облаке.,И шаги разные для каждого поставщика облачных услуг.

Требуется настройка балансировщика нагрузки, DNS Записи, сертификаты и управление хранилищем ключей.

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

Параметризированное и настраиваемое развертывание

в процессе развертывания,Вы можете определить параметры выполнения процесса,Вы также можете выбрать размер процесса и функции автоматического масштабирования:

Встроенный мониторинг и оповещение

Можно определить пользовательские ключевые показатели эффективности для мониторинга важных для вас аспектов процесса. Также можно определить оповещения для генерации уведомлений при превышении настроенных пороговых значений:

После развертывания метрики, собранные для определенных ключевых показателей эффективности, можно отслеживать на информационной панели CDF:

Cloudera DataFlow Конвекция тоже есть. NiFi Прямой доступ к холсту, позволяющий при необходимости проверять детали выполнения или устранять неполадки. графический интерфейс Все функции также доступны через CDP CLI или CDF API Программно использовать. Процесс создания и управления может быть полностью автоматизирован и CD/CI Трубопроводная интеграция.

в заключение

Собирайте данные по мере их создания и быстро предоставляйте их на аналитической платформе.,Это имеет решающее значение для успеха любого проекта, требующего обработки потоков данных в реальном времени. в этом блоге,нас展示了 Cloudera DataFlow Как упростить создание, тестирование и развертывание конвейеров данных в облаке.

Графический пользовательский интерфейс Apache NiFi и мощный процессор позволяют пользователям создавать простые и сложные потоки данных без написания кода. Интерактивный интерфейс упрощает тестирование и устранение неполадок процессов во время разработки.

Cloudera DataFlow Stream Runtime повышает надежность и эффективность выполнения рабочих потоков в облачных и эластичных средах, позволяя масштабировать их вверх и вниз в соответствии с требованиями рабочей нагрузки.

Во второй части этого блога мы увидим, как использовать Cloudera Потоковая обработка (CSP) Завершите наш вариант обнаружения мошенничества реализация использования, выполняющая потоковый анализ в реальном времени только что полученных данных.

Узнайте о Cloudera DataFlow из Больше информации и попробуйте Быстрый способ Что? Сначала посетите нашу новую Клаудеру DataFlow Домашняя страница. Затем пройдите наш интерактивный тур по продукту и зарегистрируйтесь на бесплатную пробную версию.

Автор оригинала: Андре Араужо

Исходная ссылка: https://blog.cloudera.com/fraud-detection-with-cloudera-stream-processing-part-1/