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

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

Подключить данные

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

1. В проекте щелкните Создать, затем Подключить данные.

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

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

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

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

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

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

Выбор данных

Apache Kafka

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

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

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

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

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

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

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

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

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

Amazon Kinesis

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

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

Amazon S3

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

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

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

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

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

      • MyDir3/<yyyy>_<MM>_<dd>_<HH>_заказы.csv

      • MyDir3/<yyyy>/<MM>/<dd>/<HH>_заказы.csv

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

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

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

Выбор типа содержимого

Выберите тип содержимого событий источника.

• Выберите тип событий, которые вы загружаете в Выберите тип событий данных.

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

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

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

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

Настройка свойств приема данных

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

• Прочитать данные из

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

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

• Развертка по столбцам

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

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

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

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

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

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

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

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

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

  • Разделение по дате приема событий: таблицы разбиваются на разделы по дате появления событий в системе.

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

• Изменение данных

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

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

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

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

Сводка

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

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

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

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

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

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

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

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

    Для получения дополнительной информации см. раздел Storing streaming datasets.

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

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

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

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

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

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

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

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

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

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

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

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

JSON

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

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

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

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

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

CSV, TSV, REGEX и SPLIT

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

Parquet

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

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

Авро

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

ORC

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

Ограничения и замечания

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

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

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

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

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

  Примечание к информацииПри приеме разреженных столбцов из SQL Server, количество 500 включает Qlik столбцы заголовков.