Elasticsearch Catalog

Elasticsearch Catalog не только поддерживает автоматическое сопоставление метаданных ES, но и объединяет возможности распределённого планирования запросов Doris с возможностями полнотекстового поиска ES (Elasticsearch), предоставляя более комплексное решение для OLAP-анализа:

1. Распределённые запросы Join к нескольким индексам ES.

2. Совместные запросы к таблицам Doris и ES с более сложной фильтрацией на основе полнотекстового поиска.

Предварительные требования

Поддерживается Elasticsearch 5.x и выше.

Подключение к Elasticsearch

CREATE CATALOG es_catalog PROPERTIES (
    'type' = 'es',
    'hosts' = 'http://127.0.0.1:9200'
    {ElasticsearchProperties}
);

• {ElasticsearchProperties}

Параметр	Обязательный	По умолчанию	Описание
hosts	Да		Один или несколько адресов ES либо адрес балансировщика нагрузки ES
user	Нет	Пусто	Имя пользователя ES
password	Нет	Пусто	Пароль соответствующего пользователя
doc_value_scan	Нет	true	Получать ли значения полей запроса из столбцового хранилища ES/Lucene
keyword_sniff	Нет	true	Определять ли text.fields для строковых типов с токенизацией в ES и выполнять запросы через keyword. При значении false сопоставление выполняется по токенизированному содержимому
nodes_discovery	Нет	true	Включать ли обнаружение узлов ES. По умолчанию true. В изолированных сетевых средах задайте false, чтобы подключаться только к указанным узлам
ssl	Нет	false	Включён ли в ES доступ по https. В текущей реализации fe/be доверяют всем сертификатам
mapping_es_id	Нет	false	Сопоставлять ли поле _id в индексе ES
like_push_down	Нет	true	Преобразовывать ли like в wildcard и передавать ли его в ES для выполнения. Это увеличивает потребление CPU на стороне ES
include_hidden_index	Нет	false	Включать ли скрытые индексы. По умолчанию false

Примечания:

1. В настоящее время поддерживается только аутентификация HTTP Basic. Убедитесь, что у пользователя есть доступ к путям /_cluster/state/, _nodes/http и права на чтение индексов. Если аутентификация для защиты кластера не включена, имя пользователя и пароль не требуются.

2. В версиях 5.x и 6.x, если в одном индексе несколько типов, по умолчанию используется первый тип.

Сопоставление иерархии

Поскольку в Elasticsearch нет понятия базы данных, после подключения к ES автоматически создаётся единственная база данных: default_db.

При переключении на ES Catalog с помощью команды SWITCH автоматически выбирается default_db, поэтому выполнять команду USE default_db не требуется.

Сопоставление типов столбцов

Тип ES	Тип Doris	Комментарий
null	null
boolean	boolean
byte	tinyint
short	smallint
integer	int
long	bigint
unsigned_long	largeint
float	float
half_float	float
double	double
scaled_float	double
date	date	Поддерживаются только форматы default/yyyy-MM-dd HH:mm:ss/yyyy-MM-dd/epoch_millis
keyword	string
text	string
ip	string
constant_keyword	string
wildcard	string
nested	json
object	json
flattened	json	Поддерживается начиная с версий 3.1.4, 4.0.3
other	UNSUPPORTED

Типы массивов

В Elasticsearch нет отдельного типа массива, но его поля могут содержать 0 или более значений.

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

Для Elasticsearch 6.x и более ранних версий см. _meta.

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

{
  "array_int_field": [1, 2, 3, 4],
  "array_string_field": ["doris", "is", "the", "best"],
  "id_field": "id-xxx-xxx",
  "timestamp_field": "2022-11-12T12:08:56Z",
  "array_object_field": [
    {
      "name": "xxx",
      "age": 18
    }
  ]
}

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

# ES 7.x and above
curl -X PUT "localhost:9200/doc/_mapping?pretty" -H 'Content-Type:application/json' -d '
{
    "_meta": {
        "doris":{
            "array_fields":[
                "array_int_field",
                "array_string_field",
                "array_object_field"
            ]
        }
    }
}'

# ES 6.x and before
curl -X PUT "localhost:9200/doc/_mapping/_doc?pretty" -H 'Content-Type: application/json' -d '
{
    "_meta": {
        "doris":{
            "array_fields":[
                "array_int_field",
                "array_string_field",
                "array_object_field"
            ]
        }
    }
}

array_fields: указывает поля, имеющие тип массива.

Тип flattened

Для типа flattened, когда свойство enable_docvalue_scan имеет значение false, считываемые данные JSON представлены в плоском формате. Если свойство enable_docvalue_scan имеет значение true, считывается исходный формат JSON. Пример приведён ниже:

Определение индекса:

"mappings": {
   "properties": {
      "column1": {
      "type": "keyword"
      },
      "extra": {
      "type": "flattened"
      }
   }
}

Данные:

{
  "column1": 1,
  "extra": {
    "subcol1": "abc",
    "sub_array": [
      {"k1": "element1"},
      {"k2": "element2"},
      {"k3": "element3"}
    ]
  }
}

• Если свойство enable_docvalue_scan имеет значение false

  Результат запроса для столбца extra:

  {
     "subcol1": "abc",
     "sub_array": [
        {"k1": "element1"},
        {"k2": "element2"},
        {"k3": "element3"}
     ]
  }

• Если свойство enable_docvalue_scan имеет значение true

  Результат запроса для столбца extra:

  ["abc","element1","element2","element3"]

Выполнение запросов

После настройки Catalog можно запрашивать данные таблиц в Catalog следующими способами:

-- 1. switch to catalog, use database and query
SWITCH es_ctl;
SELECT * FROM es_tbl LIMIT 10;

-- 2. use default_db directly
USE es_ctl.default_db;
SELECT * FROM es_tbl LIMIT 10;

-- 3. use full qualified name to query
SELECT * FROM es_ctl.default_db.es_tbl LIMIT 10;

Подробнее о расширенных возможностях запросов см. в разделе [Рекомендации по использованию].

Рекомендации по использованию

Передача предикатов фильтрации

ES Catalog поддерживает передачу предикатов фильтрации: условия фильтрации передаются в ES, поэтому возвращаются только данные, которые действительно соответствуют условиям. Это позволяет значительно повысить производительность запросов и сократить использование CPU, памяти и ресурсов ввода-вывода в Doris и Elasticsearch.

Следующие операторы преобразуются в соответствующие запросы ES:

Синтаксис SQL	Синтаксис ES 5.x+
=	запрос term
in	запрос terms
> , < , >= , <=	запрос range
and	bool.filter
or	bool.should
not	bool.must_not
not in	bool.must_not + запрос terms
is_not_null	запрос exists
is_null	bool.must_not + запрос exists
esquery	QueryDSL в нативном для ES формате JSON

Включение столбцового сканирования для ускорения запросов (enable_docvalue_scan=true)

Установите "enable_docvalue_scan" = "true"

После включения Doris будет получать данные из ES в соответствии со следующими двумя принципами:

• По возможности: автоматически определяется, включено ли столбцовое хранение для считываемых полей (doc_value: true). Если столбцовое хранение включено для всех полей, Doris получает значения всех полей из столбцового хранилища.

• Автоматический переход к резервному способу: если хотя бы для одного из запрашиваемых полей столбцовое хранение не включено, значения всех полей извлекаются путём разбора данных из строкового хранилища _source.

Преимущества:

По умолчанию Doris On ES получает все необходимые столбцы из строкового хранилища, то есть _source. В хранилище _source используется строковый формат с JSON, который уступает столбцовому хранению по производительности пакетного чтения, особенно когда требуется лишь несколько столбцов. В таких случаях производительность docvalue примерно в десять раз выше, чем у _source.

Примечания:

1. Для полей типа text в ES нет столбцового хранилища, поэтому, если хотя бы одно из запрашиваемых значений относится к типу text, автоматически выполняется переход к получению данных из _source.

2. Если запрашивается слишком много полей (>= 25), производительность получения значений полей из docvalue будет практически такой же, как при получении значений из _source.

3. Поля типа keyword могут отображаться пустыми из-за ограничения параметра ignore_above, если длина текстового поля превышает этот предел. В таком случае нужно отключить enable_docvalue_scan и получать результаты из _source.

Обнаружение полей типа keyword

Установите "enable_keyword_sniff" = "true"

В ES можно импортировать данные напрямую, не создавая индекс заранее. В этом случае ES автоматически создаст новый индекс. Для строковых полей ES создаёт поля одновременно с типом text и типом keyword. Это возможность ES под названием multi fields; сопоставление выглядит следующим образом:

"k4": {
   "type": "text",
   "fields": {
      "keyword": {   
         "type": "keyword",
         "ignore_above": 256
      }
   }
}

При фильтрации по k4, например с помощью =, Doris On ES преобразует запрос в TermQuery ES.

Условие фильтрации SQL:

k4 = "Doris On ES"

После преобразования в DSL запроса ES:

"term" : {
    "k4": "Doris On ES"

}

Поскольку первый тип поля k4 — text, при импорте данных его содержимое разбивается на токены в соответствии с токенизатором, настроенным для k4 (если он не задан, используется стандартный токенизатор). В результате получаются три терма: doris, on, es, как показывает анализ с помощью analyze API ES:

POST /_analyze
{
  "analyzer": "standard",
  "text": "Doris On ES"
}

Результат токенизации:

{
   "tokens": [
      {
         "token": "doris",
         "start_offset": 0,
         "end_offset": 5,
         "type": "<ALPHANUM>",
         "position": 0
      },
      {
         "token": "on",
         "start_offset": 6,
         "end_offset": 8,
         "type": "<ALPHANUM>",
         "position": 1
      },
      {
         "token": "es",
         "start_offset": 9,
         "end_offset": 11,
         "type": "<ALPHANUM>",
         "position": 2
      }
   ]
}

В запросе используется:

"term" : {
    "k4": "Doris On ES"
}

Терм Doris On ES не совпадает ни с одним термом в словаре, поэтому запрос не вернёт результатов. Включение enable_keyword_sniff: true автоматически преобразует k4 = "Doris On ES" в k4.keyword = "Doris On ES", обеспечивая полное соответствие семантике SQL. Преобразованный DSL запроса ES:

"term" : {
    "k4.keyword": "Doris On ES"
}

Поле k4.keyword имеет тип keyword, и данные записываются в ES как единый терм, поэтому совпадение возможно.

Включение автоматического обнаружения узлов (nodes_discovery=true)

Установите "nodes_discovery" = "true"

При значении true Doris находит в ES все доступные соответствующие узлы данных, на которых размещены указанные выше шарды. Если Doris BE не может обращаться к адресам узлов данных ES, установите false. Например, кластер ES развёрнут во внутренней сети, изолированной от общедоступного Интернета, а пользователи обращаются к нему через прокси.

Включение доступа по HTTPS к кластеру ES

Установите "ssl" = "true"

В текущей реализации FE/BE доверяют всем сертификатам. Это временное решение; в дальнейшем будут использоваться сертификаты, настроенные пользователем.

Расширенная функция esquery()

Функция esquery(field, QueryDSL) позволяет передавать в ES для фильтрации запросы, которые невозможно выразить в SQL, например match_phrase, geoshape и другие. Первый параметр esquery, содержащий имя столбца, используется для привязки к index, а второй параметр представляет собой базовый Query DSL ES в формате JSON, заключённый в фигурные скобки {}. В JSON должен быть ровно один корневой ключ root key, например match_phrase, geo_shape, bool и т. д.

Запрос match_phrase:

select * from es_table where esquery(k4, '{
        "match_phrase": {
           "k4": "doris on es"
        }
    }');

Запросы, связанные с geo:

select * from es_table where esquery(k4, '{
      "geo_shape": {
         "location": {
            "shape": {
               "type": "envelope",
               "coordinates": [
                  [
                     13,
                     53
                  ],
                  [
                     14,
                     52
                  ]
               ]
            },
            "relation": "within"
         }
      }
   }');

Запрос bool:

select * from es_table where esquery(k4, ' {
         "bool": {
            "must": [
               {
                  "terms": {
                     "k1": [
                        11,
                        12
                     ]
                  }
               },
               {
                  "terms": {
                     "k2": [
                        100
                     ]
                  }
               }
            ]
         }
      }');

Рекомендации по использованию полей даты и времени

Применимо только к внешним таблицам ES. В ES Catalog типы дат автоматически сопоставляются с Date или Datetime.

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

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

 "dt": {
     "type": "date",
     "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
 }

При создании этого поля в Doris рекомендуется задать тип date или datetime; также можно использовать тип varchar. Следующие SQL-запросы позволяют передавать условия фильтрации непосредственно в ES:

select * from doe where k2 > '2020-06-21';

select * from doe where k2 < '2020-06-21 12:00:00'; 

select * from doe where k2 < 1593497011; 

select * from doe where k2 < now();

select * from doe where k2 < date_format(now(), '%Y-%m-%d');

Примечание:

• Если для полей даты и времени в ES не задан format, по умолчанию используется следующий формат:

  strict_date_optional_time||epoch_millis

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

Получение поля ID метаданных ES

При импорте документов без указания _id ES присваивает каждому документу глобально уникальный _id (первичный ключ). При импорте пользователи также могут указать _id, имеющий особое значение для бизнес-логики.

Если нужно получать значение этого поля во внешних таблицах ES, при создании таблицы можно добавить поле _id типа varchar:

CREATE EXTERNAL TABLE `doe` (
  `_id` varchar COMMENT "",
  `city`  varchar COMMENT ""
) ENGINE=ELASTICSEARCH
PROPERTIES (
"hosts" = "http://127.0.0.1:8200",
"user" = "root",
"password" = "root",
"index" = "doe"
}

Если нужно получать значение этого поля в ES Catalog, установите "mapping_es_id" = "true".

Примечание:

1. В условиях фильтрации для поля _id поддерживаются только = и in.

2. Поле _id должно иметь тип varchar.

Получение глобально упорядоченных результатов запроса

В таких сценариях, как ранжирование по релевантности и приоритетное отображение важного содержимого, полезно сортировать результаты запросов ES по оценке релевантности. Doris запрашивает ES, получая данные в соответствии с распределением шардов индексов ES, чтобы в полной мере использовать преимущества архитектуры MPP. Для получения глобально упорядоченных результатов запросы к ES нужно выполнять из одной точки. Этим можно управлять с помощью переменной сеанса enable_es_parallel_scroll (по умолчанию true). Если установлено enable_es_parallel_scroll=false, Doris отправляет в кластер ES запросы scroll без информации shard_preference и sort, получая таким образом глобально упорядоченные результаты. Примечание: используйте с осторожностью, если набор результатов запроса велик.

Изменение размера пакета запросов scroll

Размер batch для запросов scroll по умолчанию составляет 4064. Его можно изменить с помощью переменной сеанса batch_size.

Часто задаваемые вопросы

1. Поддерживаются ли кластеры ES с аутентификацией X-Pack?

   Поддерживаются все кластеры ES, использующие аутентификацию HTTP Basic.

2. Некоторые запросы выполняются значительно медленнее, чем при прямом обращении к ES

   Да, например запросы, связанные с _count. ES напрямую считывает внутренние метаданные о количестве документов, соответствующих условиям, без фильтрации самих данных.

3. Можно ли передавать операции агрегации в ES?

   В настоящее время Doris On ES не поддерживает передачу в ES операций агрегации, таких как sum, avg, min/max и т. д. Вычисления выполняются следующим образом: все документы, соответствующие условиям, поступают из ES потоком пакетами, после чего обрабатываются в Doris.

Приложение

Принцип выполнения запросов Doris к ES

+----------------------------------------------+
|                                              |
| Doris      +------------------+              |
|            |       FE         +--------------+-------+
|            |                  |  Request Shard Location
|            +--+-------------+-+              |       |
|               ^             ^                |       |
|               |             |                |       |
|  +-------------------+ +------------------+  |       |
|  |            |      | |    |             |  |       |
|  | +----------+----+ | | +--+-----------+ |  |       |
|  | |      BE       | | | |      BE      | |  |       |
|  | +---------------+ | | +--------------+ |  |       |
+----------------------------------------------+       |
   |        |          | |        |         |          |
   |        |          | |        |         |          |
   |    HTTP SCROLL    | |    HTTP SCROLL   |          |
+-----------+---------------------+------------+       |
|  |        v          | |        v         |  |       |
|  | +------+--------+ | | +------+-------+ |  |       |
|  | |               | | | |              | |  |       |
|  | |   DataNode    | | | |   DataNode   +<-----------+
|  | |               | | | |              | |  |       |
|  | |               +<--------------------------------+
|  | +---------------+ | | |--------------| |  |       |
|  +-------------------+ +------------------+  |       |
|   Same Physical Node                         |       |
|                                              |       |
|           +-----------------------+          |       |
|           |                       |          |       |
|           |      MasterNode       +<-----------------+
| ES        |                       |          |
|           +-----------------------+          |
+----------------------------------------------+

1. FE обращается к хостам, указанным при создании таблицы, чтобы получить информацию об HTTP-портах всех узлов, распределении шардов индексов и другие сведения. Если запрос завершается ошибкой, FE последовательно перебирает список хостов до успешного выполнения запроса или исчерпания списка.

2. При выполнении запроса на основе полученных FE сведений об узлах и метаданных индексов формируется план запроса, который отправляется соответствующим узлам BE.

3. Узлы BE сначала обращаются к локально развёрнутым узлам ES, руководствуясь правилом proximity principle. BE параллельно получает данные из каждого шарда индекса ES методом HTTP Scroll, считывая их потоком из _source или docvalue.

4. После завершения вычислений Doris возвращает результаты пользователю.