Потоковые данные

Процесс адаптации передает данные из источника и сохраняет их в таблицах Iceberg. Изменения из потоковых источников данных непрерывно применяются к таблицам хранения почти в реальном времени.

Адаптация данных

Данные адаптируются в рамках проекта конвейера, а наборы данных сохраняются в местоположении S3, определенном в настройках проекта.

1. В проекте нажмите Создать, а затем Адаптировать данные.

2. Добавьте Имя задачи и необязательное Описание для адаптации.

   Нажмите Далее.

3. Выберите подключение к источнику.

   Можно выбрать существующее подключение к потоковому источнику или создать новое подключение к источнику.

   Для получения дополнительной информации см. раздел Подключение к потокам данных

   Нажмите Далее и следуйте инструкциям ниже для вашего источника данных.

Выбор данных

Apache Kafka

В списке отображаются доступные топики Kafka из кластера, определенного в подключении к источнику.

При выборе топиков можно выбрать определенные наборы данных. Также можно использовать правила выбора для включения или исключения групп наборов данных:

• Используйте % как знак подстановки для определения критериев выбора для наборов данных.

Если топики выбраны с использованием правил выбора, можно выбрать, загружать ли все наборы данных в одну целевую таблицу или создавать отдельную целевую таблицу для каждого исходного топика:

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

• Если правила выбора используются для загрузки нескольких топиков в одну таблицу, необходимо указать целевое имя.

• Если используются правила выбора и данные загружаются в отдельные таблицы (один набор данных на топик), целевыми именами по умолчанию являются имена топиков. На этом этапе нельзя изменить имена в мастере, но это можно сделать позже в задаче промежуточного хранения.

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

Выберите один или несколько наборов данных или используйте правило выбора и нажмите Добавить. Нажмите Далее.

Amazon Kinesis

В списке отображаются доступные потоки Kinesis, определенные в подключении к источнику.

Выберите один или несколько наборов данных и нажмите Добавить. Добавленные наборы данных можно увидеть в разделе Выбранные потоки. Нажмите Далее.

Amazon S3

В браузере каталогов отображается список всех каталогов, расположенных в блоке S3 вашего подключения к источнику. 

• Выберите каталоги для включения при промежуточном хранении данных:

  • Для каждого каталога в поле Добавить путь введите путь и шаблон имени файла:

    • Используйте * как знак подстановки для соответствия любому символу.

    • Чтобы ввести шаблон даты, используйте <yyyy> в качестве четырехзначного заполнителя года, <MM> в качестве двузначного заполнителя месяца, <dd> в качестве двузначного заполнителя дня и <HH> в качестве двузначного заполнителя часа. Например:

      • MyDir3/<yyyy>_<MM>_<dd>_<HH>_orders.csv

      • MyDir3/<yyyy>/<MM>/<dd>/<HH>_orders.csv

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

• Нажмите Проверить, чтобы убедиться, что пути и шаблоны имен файлов правильные и рабочие.

• В поле Определить имя целевого набора данных укажите имя, чтобы сопоставить топик с целевой таблицей Iceberg. Нажмите Далее.

Выбор типа контента

Выберите тип контента исходных событий.

• Выберите тип принимаемых событий в разделе Выберите тип событий данных.

• Для получения дополнительной информации см. раздел Подключение к потокам данных.

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

• Разверните раздел Проверить правильность загрузки событий, чтобы подтвердить, что данные могут быть проанализированы. Рекомендуется убедиться в правильности данных на этом этапе, иначе потребуется воссоздать конвейер и загрузить данные снова. Используйте Выбрать набор данных для изучения конкретных наборов данных и проверки любых предупреждений, которые могут повлиять на загрузку данных. Нажмите на значок глаза рядом с любыми столбцами структуры, чтобы просмотреть данные.

• Нажмите Далее.

Настройка свойств приема

Настройте параметры для вашего конвейера:

• Читать данные из

  • Начать с самого раннего события: прием всех исторических данных.

  • Начать с текущего момента: прием новых данных, поступающих с момента запуска конвейера.

• Развертывание столбцов

  • Сохранить вложенные столбцы: преобразования не применяются.

  • Развернуть в отдельные столбцы: данные разделяются на отдельные столбцы.

• Настройки загрузки для новых наборов данных

  • Только добавление: как правило, лучший вариант для данных событий, поскольку они обычно имеют короткий срок жизни и не обновляются, например, Заказы.

  • Применить изменения: это лучше всего подходит для данных, которые обновляются со временем, например, Клиенты. Обновляет существующие записи и вставляет новые записи на основе ключевых полей. Потребуется указать ключевые поля позже при определении задачи.

• Раздел целевой таблицы

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

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

  • Без раздела: таблицы создаются без какого-либо секционирования.

  • Секционировать по дате приема события: таблицы секционируются по дате приема событий.

    Примечание к информацииЕсли этот параметр выбран вместе с параметром столбца заголовка hdr__from_timestamp, hdr__from_timestamp будет использоваться в качестве столбца раздела по умолчанию. Для получения информации о добавлении столбца заголовка hdr__from_timestamp в стандартные представления см. раздел Определения таблиц.

• Обработка изменений данных

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

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

  • Создать хранилище исторических данных (Тип 2): Это позволит сохранить предыдущие версии измененных записей.

• Нажмите Далее.

Сводка

На экране сводки отображается визуальное представление вашего конвейера:

• При необходимости для задачи потокового промежуточного хранения и потокового преобразования можно нажать Изменить имя и описание, чтобы указать новые значения.

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

• После настройки всех параметров нажмите Создать, чтобы создать потоковый конвейер.

• Когда проект отобразится, можно подготовить и запустить каждую задачу, чтобы начать прием данных.

  • Подготовьте и запустите задачу потокового промежуточного хранения.

    Для получения дополнительной информации см. раздел Промежуточное хранение потоковых данных в Открытое озеро данных Qlik.

  • Подготовьте и запустите задачу потокового преобразования.

    Для получения дополнительной информации см. раздел Хранение потоковых наборов данных.

Сопоставления типов данных

Начальная исходная схема основана на данных образца, взятых до этапа ПОДГОТОВКИ при создании проекта конвейера, а эволюция схемы обрабатывается во время чтения. Задачи зеркалирования и другие последующие задачи, которые не поддерживают STRUCT и ARRAY, используют тип JSON. Данные можно анализировать с помощью SQL.

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

• Типы данных выводятся из данных образца, которые адаптируются. Например, если поле содержит только целочисленные значения в образце, оно создается как INT8 в задачах потокового промежуточного хранения и преобразования. Если последующие данные включают дробные значения двойной прецизионности, файлы промежуточного хранения содержат эти значения; однако в задаче потокового преобразования, если для параметра Изменить тип данных поля установлено значение Игнорировать, столбец остается INT8, а дробные значения усекаются. Чтобы избежать непреднамеренного усечения, убедитесь, что данные образца включают полный диапазон ожидаемых значений перед адаптацией, или настройте Изменить тип данных поля на Остановить задачу на ранних этапах и при необходимости скорректируйте типы данных.

• Если поле добавляется в структуру в источнике, оно всегда добавляется в цель промежуточного хранения. Для потокового преобразования поведение применяется в соответствии с параметром, выбранным в разделе Настройки задачи потокового преобразования > Эволюция схемы > Добавить поля в структуру (Применить к цели, Игнорировать, Остановить задачу).

• Если поле отсутствует в определенной записи или массив пуст, они рассматриваются как null.

• Если набор данных сглаживается массивом, и поступает запись, в которой этот массив пуст или имеет значение null, система создает одну строку, а сглаженное поле имеет значение null. Оно не исключается автоматически. Если вы хотите исключить эти строки, вручную добавьте фильтр, например, array_element IS NOT NULL.

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

• Новый атрибут не может быть добавлен внутрь структуры во вложенном поле JSON, только на корневом уровне.

• В задачах потокового преобразования сглаживание поддерживается только для одного уровня массива. Когда сглаживание применяется к многоуровневому массиву, например, ARRAY<ARRAY<STRUCT>>, сглаживается только внешний массив, что приводит к ARRAY<STRUCT>, а не к полностью сглаженной STRUCT. Кроме того, текущий пользовательский интерфейс позволяет настраивать сглаживание только на уровне столбца. В результате выбор многоуровневого массива неявно применяет сглаживание только к первому уровню массива.

• При обращении к массиву примитивов используется тип данных элемента, если детализацией является массив. В противном случае используется тип данных массива.

  В этом примере OrderDetails имеет массив CustomerID с типом данных INT. OrderDetails.CustomerID означает INT, если детализацией является OrderDetails.CustomerID, и ARRAY<INT>, если детализацией является OrderDetails.

JSON

В файлах JSON числовое значение в источнике определяет целевой тип данных:

• INT8 используется для целочисленных значений, которые вписываются в поддерживаемый диапазон целых чисел и не включают дробную часть.

• REAL8 (DOUBLE) используется, когда значение содержит дробную часть (число с плавающей запятой).

• STRING используется, когда числовое значение превышает максимальный поддерживаемый диапазон целых чисел.

Типы данных сопоставляются следующим образом:

CSV, TSV, REGEX и SPLIT

По умолчанию все исходные типы данных принимаются в строку. Используйте параметр Автоматически выводить типы, чтобы сопоставить исходные и целевые типы следующим образом:

Parquet

Файлы Parquet поддерживают физические и логические типы данных. Физические типы данных определяют, как значения хранятся на диске, например INT32, DOUBLE или BYTE_ARRAY. Логические типы данных обеспечивают семантическое значение поверх физического представления, например, определяя, представляет ли целочисленное значение дату. Если логический тип прикреплен к столбцу Parquet и поддерживается в Открытое озеро данных Qlik (как указано ниже), задача потокового промежуточного хранения использует логический тип при определении целевой схемы, а не базовый физический тип. Это гарантирует правильную интерпретацию данных, сохраняет предполагаемую семантику, такую как прецизионность, масштаб и временное значение, и приводит к более точным схемам при записи данных в последующие форматы.

Данные, полученные из файлов Parquet, сопоставляются следующим образом:

Avro

Следующие сопоставления применяются к файлам Avro с реестром схем.

ORC

Следующие сопоставления применяются к файлам ORC.

Ограничения и рекомендации

• Если структура или массив изменяются в результате автоматической эволюции схемы при промежуточном хранении, последующие представления, которые не были созданы потоковой задачей Qlik Talend Cloud, могут потребовать обновления, чтобы не устареть.

• Если в задаче есть ошибки анализа, она не перейдет в состояние ошибки и не будет отмечена как требующая внимания. Поскольку ошибки анализа являются постоянно растущей метрикой, критерия выхода для состояния ошибки не существует.

• Удаление возможности кластера допускается только в том случае, если нет задач, использующих эту возможность.

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

• Если источник содержит большое количество столбцов, в задачах и в каталоге отображаются только 500 самых популярных столбцов по частоте. Все столбцы сохраняются в файлах Avro при промежуточном хранении S3, но только 500 самых популярных столбцов сохраняются в таблицах Iceberg. При эволюции схемы, если добавляется новый столбец, он не будет добавлен в самые популярные столбцы, даже если он часто встречается.