#arenadata

В этой статье мы рассмотрим, что такое Apache Zeppelin, как он полезен для интерактивной аналитики и визуализации больших данных (Big Data), а также чем этот инструмент отличается от популярного среди Data Scientist’ов и Python-разработчиков Jupyter Notebook.  

Что такое Apache Zeppelin и чем он полезен Data Scientist’у

Начнем с определения: Apache Zeppelin – это интерактивный веб-блокнот (или «ноутбук» от notebook) с открытым исходным кодом, который поддерживает практически все этапы работы с данными в Data Science, от извлечения до визуализации, в т.ч. интерактивный анализ и совместное использование документов. Он интегрирован с Apache Spark, Flink, Hadoop, множеством реляционных и NoSQL-СУБД (Cassandra, HBase, Hive, PostgreSQL, Elasticsearch, Google Big Query, Mysql, MariaDB, Redshift), а также поддерживает различные языки программирования, популярные в области Big Data: Python, PySpark, R, Scala, SQL. Такая многофункциональность обеспечивается за счет интерпретаторов — плагинов для поддержки языка программирования, базы данных или фреймворка. С точки зрения работы с большими данными, отдельного упоминания заслуживает встроенная интеграция с Apache Spark, что дает общие контексты (SparkContext и SQLContext), загрузку jar-зависимостей из локальной файловой системы или репозитория maven во время выполнения задачи, а также возможность отмены задания и отображение хода его выполнения. Также Цеппелин поддерживает работу с REST-API Apache Spark – Livy, о котором мы подробно писали здесь. Благодаря интерпретатору Python, Apache Zeppelin предоставляет все возможности этого языка, ориентированные на Data Science, например, специализированные библиотеки (Matplotlib, Conda, Pandas и пр.) для аналитики больших данных и визуализации. Это позволяет автоматически построить круговые, столбчатые и прочие наглядные диаграммы, чтобы визуализировать статистику датасета или результатов исследования. Также в Zeppelin можно создавать интерактивные дэшборды с формами ввода данных, которые будут выглядеть как веб-страницы, чтобы поделиться их URL-адресами для совместной работы. Для многопользовательского режима Zeppelin поддерживает LDAP-авторизацию с настройками доступа [1].     Впрочем, при всех этих достоинствах, на практике можно столкнуться со следующими ограничениями Apache Zeppelin, которые могут рассматриваться как недостатки [2]: ·       нестабильная работа под высокой нагрузкой; ·       интерактивный веб-интерфейс требует много оперативной памяти; ·       отсутствие полного набора возможностей современных специализированных IDE; ·       меньшая «зрелость» и популярность по сравнению с Jupyter Notebook. Тем не менее, Apache Zeppelin завоевывает свою нишу, конкурируя с Jupyter Notebook в некоторых кейсах работы с большими данными. В каких случаях аналитику Big Data или Data Scientist’у следует предпочесть Apache Zeppelin вместо Jupyter Notebook, мы рассмотрим далее.  

Apache Zeppelin vs Jupyter Notebook: что и когда выбирать для аналитики Big Data

В этой статье разберем кейс построения экосистемы управления Big Data с озером данных на примере федеральной фармацевтической сети - российской Ассоциации независимых аптек (АСНА). Читайте в этом материале, зачем фармацевтическому ритейлеру большие данные, с какими трудностями столкнулся этот проект цифровизации и как открытые технологии (Arenadata Hadoop, Apache Spark, NiFi и Hive), взаимодействуют с проприетарными решениеми Informatica Big Data Management и 1С, а также облачными сервисами Azure.

Постановка задачи от бизнеса: проблемы, возможности и ограничения

АСНА позиционирует себя не просто информационным агрегатором по локальным и сетевым аптекам, а высокотехнологичной data-driven компанией с уникальной бизнес-моделью, в которой ключевую роль играет управление данными. Поэтому обеспечение их чистоты, качества и надежности является ключевой бизнес-задачей. Поскольку количество партнеров и поставщиков постоянно увеличивается, требование к быстрой масштабируемости Big Data системы стало одним из самых приоритетных [1]. Чтобы пояснить масштаб работы с большими данными, перечислим некоторые бизнес-показатели [2]: ·       250+ контрактов с различными производителями фармпрепаратов; ·       более 9000 товарных идентификаторов в маркетинговых контрактах, где следует контролировать ассортимент на уровне маркетингового соглашения; ·       около 2500 контрактов с разными аптечными сетями по всей России; ·       около 10000 аптек, в каждой из которых необходим контроль маркетингового ассортимента. Необходимость комплексной Big Data экосистемы управления данными вызвана следующими проблемами [2]: ·       ежедневная загрузка и обработка данных длилась порядка 18 часов на 6500 аптек; ·       интеграция и валидация данных также занимали чрезмерно много времени; ·       процедуры закрытия контрольного периода и сверки данных были слишком сложными, постоянно какая-то информация не загружалась; ·       добавление новых аптек длилось очень долго; ·       при однодневном сбое некоторых процессов обработки данных, приходилось стабилизировать работу всех систем за недельный период. Кроме решения этих проблем, перед новой экосистемой управления большими данными также стояла задача оперативного формирования бизнес-отчетности для конечных пользователей: поставщиков, аптек-партнеров и сотрудников АСНА (менеджеры, аналитики) [3].

Ход проекта: Data Lake и DWH

Проект стартовал в 2016 году, когда данные компании находились в двух хранилищах данных MS SQL Server. При этом качество данных было не высоким (так называемые «грязные» данные), отсутствовали мастер-данные и единая подсистема НСИ. Для анализа и обработки данных использовался набор разнородных приложений. В 2017 году была выбрана архитектура DWH и Data Lake на базе открытых Big Data технологий: Apache Hadoop, Hive, Spark, NiFi, а также коммерческое решение по управлению данными от Informatica.  

В процессе практической эксплуатации выяснилось, что эта архитектура не совсем соответствует требованиям к быстроте получения и удобстве представления аналитических отчетов. В частности, были выявлены следующие ограничения [2]: ·       данные, над которыми постоянно выполняются транзакционные изменения (операции удаления, обновления, вставки), лучше оставить в реляционной СУБД, а не в NoSQL-хранилище (КХД на базе Apache Hadoop); ·       так называемые «горячие» данные, которые нужны для быстрых операционных расчетов тоже целесообразнее оставить в реляционной СУБД; ·       NoSQL-инструменты отлично справляются с обработкой неструктурированных данных, но они не заменят традиционные OLAP-кубы для BI-аналитики; ·       высокая доступность выборки данных в HDFS компенсируется низкой скоростью выдачи результатов конечному потребителю. Поэтому в 2018 году озеро данных было модифицировано и стало выполнять роль операционного слоя первичных данных или стейджинга (Staging Layer) для реляционного КХД, которое используется как слой фактов (Fact Layer) для планирования бюджетов. Подробнее про послойную архитектуру КХД мы рассказывали здесь. Также в 2018-2019 годах была внедрена интегрированная обработка данных с помощью Hadoop и Informatica, разработана единая подсистема НСИ и мастер-данных в условиях постоянных изменений. Еще появилась валидация данных и стали использоваться методы машинного обучения (Machine Learning) для задач консолидации НСИ. Data Lake сети АСНА до сих совершенствуется: в частности, ведется разработка архивного хранения и API-интерфейсов [2].  

Архитектура Big Data решения

Рассмотрим архитектуру созданного Big Data решения подробнее. «Сырые» данные хранятся в Data Lake на базе отечественного дистрибутива Arenadata Hadoop. Потоковые ETL/ELT-процессы организованы с помощью Apache NiFif. В качестве SQL-on-Hadoop инструмента используется Apache Hive, что позволяет обращаться к данным в HDFS через SQL-подобные запросы (HiveQL). С помощью Apache Spark выполняется аналитика больших данных. Через облачные сервисы Azure организовано взаимодействие между внутренними системами 1С, которые поддерживают учетные функции бэк-офиса АСНА, и экосистемой управления Big Data (Data Lake И КХД). В рамках этой системы Informatica Big Data Management выполняет парсинг файлов данных, проверяет форматы данных на соответствие заранее определенным требованиям, удаляет дубликаты строк, преобразует данные в единую структуру, проводит обогащение данных внутренней подсистемой НСИ и отвечает за архивирование. Также Informatica Big Data Management обеспечивает распределение вычислительной нагрузки между разными платформами с помощью технологии pushdown. Она позволяет превратить маппинг в скрипт и выбрать среду для его запуска, чтобы комбинировать сильные стороны разных платформ и достигать их максимальной производительности. Еще одним полезным свойством Informatica Big Data Management является эффективная работа с отображениями (маппингами) данных. Это позволило формировать adhoc-отчёты на кластере с использованием Apache Spark, а затем ставить маппинг на периодическую загрузку в аналитическое хранилище на Greenplum без привлечения новых дополнительных технологий [3]. Таким образом, такая комплексная Big Data система помогает конечному пользователю получать результаты аналитических запросов через визуальные дашборды, формировать отчетность, определять эффективность бизнес-инициатив и управленческие решения, основанные на объективных данных. В частности, это позволяет рассчитывать фактические показатели по продажам, возвратам, остаткам, закупкам, маркетинговым мероприятиям и отслеживать прочие важные бизнес-метрики [1].  

Вчера мы рассматривали интеграцию ClickHouse с Apache Kafka с помощью встроенного движка. Сегодня поговорим про проблемы, которые могут возникнуть при его практическом использовании и разберем способы их решения для корректной связи этих Big Data систем.

Почему случаются тайм-ауты: многопоточность и безопасность

Напомним, интеграцию ClickHouse и Kafka обеспечивает встроенный движок (engine), который позволяет публиковать потоки данных и подписываться на них, организовать отказоустойчивое хранилище и обрабатывать потоки по мере их появления через использование таблиц с указанием специальных параметров [1]. На практике при подключении ClickHouse к Kafka с помощью этого движка может возникнуть проблема, связанная с криптографическим протоколом SSL, который обеспечивает защищенное соединение. Из-за того, что данный протокол не входит в настройки подключения, которые заданы по умолчанию, библиотека librdkafka теряет связь с брокером Кафка (Local: timed out). Аналогичный случай возникает с пакетом kafkacat, который также использует эту библиотеку. Проблема решается изменением параметров подключения: в конфигурационный файл /etc/clickhouse-server/config.xml следует добавить строку с указанием защищенного протокола SSL: ssl [2]. Примечательно, что ранее исходящие SSL-соединения вообще не работали в Docker-образе ClickHouse из-за отсутствия сертификатов. Эта проблема была исправлена в 19.16.19.85, 20.1.12, 20.3.9 и более поздних релизах. При этом движок Kafka в Кликхаус поддерживает аутентификацию на уровне сервера, что указывается в конфигурационном файле config.xml [3]. Вообще стоит отметить, что при наличии в ClickHouse большого числа таблиц типа «движок Kafka» система генерирует тайм-ауты из-за того, что библиотека librdkafka создает один поток для каждого брокера и служебных данных. Много потоков в ClickHouse приводит к большому количеству переключений контекста, в результате чего возникает ошибка Local: timed out. Сократить число таблиц Kafka, использующих разные топики, можно с помощью материализованных представлений, о которых мы рассказывали здесь. При этом несколько материализованных представлений могут фильтровать данные по виртуальному столбцу «_topic». Также стоит учитывать, что таблица Kafka использует поток из пула потоков "background_schedule". Если их слишком много, имеет смысл увеличить параметр background_schedule_pool_size, который задает количество потоков для выполнения фоновых операций в движках таблиц [3].

Параллельная обработка топиков Кафка

В случае распределенного Кликхаус, следует создать таблицу движка Kafka на каждом сервере кластера. При интеграции с Кафка нужно, что все таблицы движков Kafka использовали одно и то же имя группы потребителей (обязательный параметр kafka_group_name), чтобы параллельно использовать один и тот же топик. Если количество потребителей (серверов ClickHouse с таблицами движков Kafka) больше числа разделов (партиций) топика, некоторые потребители не смогут ничего считать. Эта проблема решается корректным заданием партиций топика Кафка с учетом числа серверов ClickHouse. Также можно сбросить данные из таблицы движка Kafka в распределенную таблицу с определенным ключом шардинга. Распределенные таблицы ClickHouse будут повторять вставки одного и того же блока, и могут быть дедуплицированы – записаны лишь 1 раз при многократной записи [3].

Проблема искаженных сообщений из Kafka: форматы и виртуальные столбцы ClickHouse

Иногда корректность переноса данных из топиков Кафка в Кликхаус зависит от формата сообщений. Напомним, ClickHouse может принимать (INSERT) и отдавать (SELECT) данные в различных форматах: Parquet, JSON, CSV, ORC, XML и еще 20 видов [4]. При заливке данных в ClickHouse из Kafka формат сообщений задается при создании таблицы «движок Кафка». Его следует точно указать в обязательном параметре kafka_format. При том в опциональных параметрах можно определить максимальное количество некорректных сообщений в блоке (kafka_skip_broken_messages). По умолчанию kafka_skip_broken_messages = 0, в случае положительного целочисленного значения движок отбросит столько сообщений Кафки, которые не получилось обработать. Одно сообщение Kafka в точности соответствует одной записи (строке) ClickHouse. Также в опциональных параметрах можно задать символ-разделитель записей (строк), которым завершается сообщение (kafka_row_delimiter) и определение схемы для тех форматов, где она необходима (kafka_schema). Например, формат бинарных сообщений Cap’n Proto требует путь к файлу со схемой и название корневого объекта schema.capnp:Message  [1]. Важно, что параметр kafka_skip_broken_messages не работает для всех форматов одинаково. Например, если одно сообщение содержит несколько строк, одна из которых искажена, движок пропустит именно искаженную строку, а не все строки в самом сообщении, как можно подумать из-за названия параметра. В JSON, TSV и подобных текстовых форматах можно анализировать все поля как строки, преобразуя их к соответствующим типам на уровне материализованного представления. Это позволяет ClickHouse помещать искаженные сообщения в отдельный поток со специальными виртуальными столбцами, такими как _error или _raw_message. Затем для таких данных можно использовать материализованное представление, чтобы отфильтровать или сохранить их отдельно [3]. Здесь стоит уточнить, что в Кликхаус виртуальный столбец является неотъемлемой частью движка таблиц, определенный в его исходном коде. Виртуальные столбцы не указываются в запросе создания таблицы (CREATE TABLE) и не отображаются в результатах запросов на отображение данных (SHOW CREATE TABLE и DESCRIBE TABLE). Виртуальные столбцы доступны только для чтения, в них нельзя записать данные. Имя виртуального столбца не должно совпадать с именем «реального» столбца в таблице Кликхаус, иначе виртуальный столбец станет недоступным. Чтобы избежать этого, имена виртуальных столбцов обычно пишутся с нижним подчеркиванием впереди [5]. В следующей статье мы рассмотрим альтернативные варианты для организации загрузки потоковых данных из Apache Kafka в ClickHouse и разберем, почему крупный сервис такси «Ситимобил» подключил к этим Big Data системам еще одну – реляционную аналитическую In-Memory MPP-СУБД Exasol. А как самостоятельно обеспечить эффективную интеграцию Apache Kafka с другими внешними источниками для потоковой обработки больших данных вы узнаете на практических курсах по Кафка в нашем лицензированном учебном центре повышения квалификации и обучения руководителей и ИТ-специалистов (разработчиков, архитекторов, инженеров и аналитиков Big Data) в Москве: Kafka интеграция для разработчиков Kafka Streams для разработчиков

 Смотреть расписание занятий

В этой статье рассмотрим интеграцию ClickHouse с Apache Kafka: когда и зачем она нужна, как связать эти две Big Data системы, каковы ограничения и недостатки существующих способов и каким образом их можно обойти. Также разберем, почему кластер Кликхаус использует Zookeeper и что такое материализованное представление таблицы Кафка.

Big Data маркетинг, непрерывный мониторинг и IoT: зачем нужна интеграция ClickHouse с Apache Kafka

Напомним, ClickHouse – это колоночная СУБД от отечественной компании «Яндекс» с открытым кодом для быстрой обработки аналитических запросов в режиме реального времени на структурированных больших данных [1]. Изначально разработчики ClickHouse предусмотрели возможность потоковой заливки данных из Apache Kafka. На практике такая задача может возникнуть, например, при маркетинговой аналитике, когда необходимо оперативно оценить успешность рекламного блока с минимальной задержкой. Вычисления производятся на основе данных о множестве событий около 100 миллионов в час. Apache Kafka отлично справляется с таким потоком сообщений, собирая и агрегируя их, однако она не подходит для интерактивной маркетинговой аналитики [2]. ClickHouse превосходно решает подобные задачи благодаря столбцовой организации хранения данных, когда значения из разных столбцов хранятся отдельно, а данные одного столбца - вместе. Это обусловливает высокую скорость считывания данных, что особенно важно для аналитики Big Data в режиме онлайн [3]. Похожий кейс с маркетинговой аналитикой больших данных мы описывали здесь, рассматривая применение Arenadata QuickMarts, основанной на ClickHouse, в Х5 Retail Group. Другой реальный пример использования ClickHouse и Kafka для оперативной аналитики больших данных – это централизованная система сбора и обработки системных логов для оценки качества сервисов и ускорения реакции на скрытые проблемы при эксплуатации корпоративных ИТ-активов [4]. Также подобная связка может быть востребована в системах интернета вещей (Internet of Things, IIoT), особенно промышленного (Industrial IoT, IIoT), когда требуется в реальном времени анализировать различные показатели технологических процессов по множеству срезов. Интересен также опыт американской ИТ-компании Cloudflare, которая предоставляет услуги CDN, защиту от DDoS-атак, безопасный доступ к ресурсам и серверы DNS. На базе Apache Kafka и ClickHouse Cloudflare построила собственный конвейер данных для аналитики пользовательских логов и высокоскоростной обработки около 6 миллионов запросов в секунду [5]. Конвейер данных для аналитики пользовательских логов компании Cloudflare на базе Apache Kafka и ClickHouse В высоконагруженных Big Data проектах для обеспечения отказоустойчивости и масштабируемости ClickHouse разворачивается в кластерном режиме. Как и в случае Apache Kafka, для координации процесса репликации используется ZooKeeper [1]. Зачем Кафка использует Зукипер и можно ли обойтись без него, мы разбирали в этой статье. Напомним, Zookeeper выполняет роль координации распределенных сервисов, не обеспечивая передачу данных из одного источника в другой. За интеграцию Кликхаус с Кафка отвечают соответствующие средства, о которых мы поговорим далее.

Движок интеграции и материализованные представления таблиц Кафка

Связь ClickHouse с другими системами хранения и обработки данных обеспечивают специальные интеграционные движки (engines). В частности, движок Kafka позволяет публиковать потоки данных и подписываться на них, организовать отказоустойчивое хранилище и обрабатывать потоки по мере их появления через использование таблиц с указанием специальных параметров. Например, следующий код показывает заливку данных из топика Kafka в таблицу Кликхаус под названием queue (очередь) [6]: CREATE TABLE queue (timestamp UInt64, level String, message String) ENGINE = Kafka('localhost:9092', 'topic', 'group1', 'JSONEachRow');  На практике стоит учитывать следующие ограничения встроенного движка для интеграции ClickHouse с Apache Kafka [6]: ·       на один раздел (партицию) топика Кафка может быть назначен лишь один потребитель; ·       общее число потребителей не должно превышать количество партиций в топике Кафка; ·       в пределах одной группы сообщений каждое из них считывается только один раз, что реализует автоматическое отслеживание, но делает неэффективным применение запроса SELECT. Обойти это ограничение, чтобы организовать многократное чтение потока данных в реальном времени, позволяет специальная концепция ClickHouse - материализованное представление (MATERIALIZED VIEW). Материализованное представление таблицы Kafka в ClickHouse позволяет СУБД непрерывно получать сообщения от Кафка в фоновом режиме. Преобразование данных с помощью SELECT в нужный формат выполняется следующим образом: поток данных реализуется через потребителя Kafka, созданного с помощью движка; материализованное представление преобразует данные от движка и помещает их в заранее созданную таблицу с нужной структурой; Одна таблица Кафка в Кликхаус может иметь любое число материализованных представлений, поскольку они не считывают данные непосредственно, а блоками получают новые записи. Материализованные представления в ClickHouse не читают Kafka-таблицы, а получают данные от них при срабатывании триггера на INSERT. Поэтому одну Kafka-таблицу можно использовать как для сырых запросов, так и для агрегированных и сгруппированных данных по ним [4]. Благодаря этому данные можно записать в несколько таблиц с разным уровнем детализации. Пример программного кода будет выглядеть следующим образом [6]: CREATE TABLE queue (timestamp UInt64, level String, message String) ENGINE = Kafka('localhost:9092', 'topic', 'group1', 'JSONEachRow');   CREATE TABLE daily (day Date, level String, total UInt64) ENGINE = SummingMergeTree(day, (day, level), 8192);   CREATE MATERIALIZED VIEW consumer TO daily     AS SELECT toDate(toDateTime(timestamp)) AS day, level, count() as total     FROM queue GROUP BY day, level;    SELECT level, sum(total) FROM daily GROUP BY level; На практике интеграция ClickHouse с Kafka посредством встроенного движка не всегда проходит гладко. Завтра мы рассмотрим, какие проблемы могут при этом возникнуть и как их решить. А технические подробности по интеграции Apache Kafka с другими внешними источниками для потоковой обработки больших данных вы узнаете на практических курсах по Кафка в нашем лицензированном учебном центре повышения квалификации и обучения руководителей и ИТ-специалистов (разработчиков, архитекторов, инженеров и аналитиков Big Data) в Москве: Kafka интеграция для разработчиков Kafka Streams для разработчиков

 Смотреть расписание занятий

Мы уже рассказывали про интеграцию Tarantool с Apache Kafka на примере Arenadata Grid. Сегодня рассмотрим, как интегрировать Кафка с MPP-СУБД Greenplum и каковы ограничения каждого из существующих способов. Читайте в сегодняшнем материале, что такое GPSS, PXF и при чем тут Docker-контейнер с коннектором Кафка для Arenadata DB.

IoT и не только или зачем интегрировать Greenplum с Apache Kafka

Прежде всего поясним, почему вообще возникает задача интеграции MPP-СУБД Greenplum с брокером сообщений Apache Kafka. Представьте, что есть множество входящих потоков данных, например, от устройств интернета вещей (Internet of Things, IoT), которые необходимо проанализировать в реальном времени. Или нужна оперативная аналитика биржевых показателей на платформе онлайн-трейдинга, где миллионы клиентов со всего мира торгуют валютой и ценными бумагами в режиме онлайн. Технология массивно-параллельной обработки данных (Massive Parallel Processing, MPP), реализованная в Greenplum, позволяет быстро и надежно решить подобные проблемы событийно-потоковой обработки (event streaming). Однако, загрузка этой информации происходит в Greenplum не напрямую, а опосредовано, через надежную Big Data систему сбора и агрегации потоковых данных – распределенную стриминговую платформу Apache Kafka. 6-я версия Greenplum, выпущенная осенью 2019 года, включает 2 следующих способа интеграции с Apache Kafka [1]: ·       Greenplum Stream Server (GPSS) – средство ETL-процессов, которое принимает потоковые данные от одного или нескольких клиентов, используя механизм внешних таблиц базы Гринплам для преобразования и вставки данных в целевую таблицу этой СУБД; ·       PXF (Platform eXtension Framework) – специализированный Java-фреймворк параллельного обмена данными со сторонними системами, который обеспечивает одновременное взаимодействие всех сегментов кластера Гринплам с внешним источником данных. Также интересен способ интеграции Arenadata DB, основанной на Greenplum, и Кафка с помощью специализированного коннектора. Каждый из перечисленных способов мы подробнее рассмотрим далее.

Что такое Greenplum Stream Server и как он работает

Greenplum Stream Server – это потоковый сервер gRPC, система удаленного вызова процедур с открытым исходным кодом, изначально разработанная в Google в 2015 году. Он включает операции и сообщения, необходимые для подключения к Greenplum и для записи данных от внешней системы в таблицу MPP-СУБД [2]: ·       утилита командной строки gpss, которая позволяет запустить экземпляр GPSS, бесконечно ожидающий данных клиента; ·       утилита командной строки gpsscli для отправки заданий загрузки данных из Kafka в экземпляр GPSS и управления этими заданиями. Типичная последовательность выполнения ETL-задачи с использованием GPSS выглядит следующим образом [2]: ·       пользователь инициирует одно или несколько ETL-заданий через клиентское приложение; ·       клиентское приложение использует gRPC-протокол для отправки и запуска заданий на загрузку данных в работающий экземпляр службы GPSS; ·       Экземпляр службы GPSS передает каждую транзакцию запроса на загрузку в главный экземпляр кластера Greenplum, с помощью протокола gpfdist для хранения данных во внешних таблицах, которые создаются вновь или используются повторно. ·       Экземпляр службы GPSS записывает данные, полученные от клиента, непосредственно в сегменты кластера базы данных Greenplum. Важным ограничением этого способа является то, что Greenplum Stream Server не поддерживает загрузку данных из нескольких топиков Kafka в одну и ту же таблицу СУБД Гринплам. В этом случае все задания будут зависать [3].

Интеграция Гринплам и Кафка с помощью Greenplum Stream Server

Как устроен PXF-фреймворк

PXF появился еще в 5-ой версии Greenplum в 2017 году. Этот Java-фреймворк реализован в виде отдельного процесса на сервере, который общается с сегментами Greenplum через REST API с одной стороны, а с другой – использует сторонние Java-клиенты и библиотеки, чтобы через JDBC связать Гринплам с Apache HDFS, Hbase и Hive, а также внешними СУБД. Например, именно так X5 Retail Group в 2019 году построила собственную аналитическую платформу на базе Arenadata DB с кластером Hadoop, о чем мы подробно рассказывали здесь. PXF предоставляет коннекторы для доступа к данным, хранящимся во внешних источниках. Коннекторы сопоставляют внешний источник данных с определением внешней таблицы базы данных Greenplum. Один процесс PXF-агента на каждом хосте сегмента Greenplum выделяет рабочий поток для каждого экземпляра СУБД, который участвует в запросе к внешней таблице. Агенты PXF на хостах с несколькими сегментами взаимодействуют с внешним хранилищем данных параллельно. PXF содержит встроенные коннекторы для Hadoop (HDFS, Hive, HBase), хранилищ объектов (Azure, Google Cloud Storage, Minio, S3) и баз данных SQL (через JDBC), а также поддерживает форматы данных text, Avro, JSON, RCFile, Parquet, SequenceFile и ORC [4]. Именно PXF-коннекторы лежат в основе интеграции Arenadata DB c Apache Kafka, которую мы рассмотрим далее.

Platform eXtension Framework Greenplum для интеграции с внешними системами

Big Data интеграция: коннектор к Кафка в Arenadata DB

Установка коннектора выполняется с помощью специального пакета adb_kafka_external_connector-.rpm на все сервера-сегменты кластера Greenplum. Этот пакет добавляет на каждый сервер Гринплам исполняемые файлы [5]: ·       /usr/lib/adbkafka/adb-kafka-consumer; ·       /usr/lib/adbkafka/adb-kafka-producer. С архитектурной точки зрения коннектор Arenadata DB (ADB) к Apache Kafka – это кластер из отдельных синхронизированных процессов, запущеных в Docker, которые, с одной стороны, являются потребителями Кафка (consumer), а с другой — вставляют данные из топиков напрямую в сегменты Greenplum. Коннектор может работать с Kafka Registry и обеспечивает полную консистентность переносимых данных даже в случае аппаратных сбоев [6]. При использовании этого решения необходимо учитывать следующую специфику: ··       число разделов (partition) в топике Kafka должно быть больше или равно числу сегментов ADB, которые запрашивают данные. Иначе нужно корректировать число сегментов при создании внешней таблицы [7]; ·       все сегменты-хосты кластера ADB должны иметь сетевой доступ до всех брокеров и Zookeeper-узлов кластера Kafka, а также иметь о них записи (hostname и IP-адрес) в своих директориях /etc/hosts [8]; ·       создание внешних таблиц коннектора доступно только суперпользователям СУБД, при том использование таблиц не ограничено [7]. Больше подробностей по настройке и эксплуатации Greenplum на примере Arenadata DB рассматривается на специализированном курсе Эксплуатация Arenadata DB в нашем лицензированном учебном центре повышения квалификации «Школа Больших Данных», который является единственным авторизованным партнером компании Arenadata по сертификации специалистов и обучению в Москве. А особенности интеграции Apache Kafka с другими внешними источниками для потоковой обработки Big Data вы узнаете на практических курсах по Кафка для программистов, архитекторов, инженеров и аналитиков больших данных: Kafka интеграция для разработчиков Kafka Streams для разработчиков

 Смотреть расписание занятий