Learn how to use the Elasticsearch connector, including limits, syntax, parameters, and examples. - Realtime Compute for Apache Flink

Contexto

Alibaba Cloud Elasticsearch es compatible con Elasticsearch de código abierto e incluye funciones comerciales como Security, Machine Learning, Graph y APM para el análisis y la búsqueda de datos. Proporciona servicios de nivel empresarial, que incluyen control de acceso, monitoreo y alertas de seguridad, así como generación automatizada de informes.

La siguiente tabla describe las capacidades del conector de Elasticsearch.

Elemento	Descripción
Tipo de tabla	Tabla de origen, tabla de dimensiones y tabla de destino
Modo de ejecución	Modo por lotes y modo streaming
Formato de datos	JSON
Métrica	Métrica Tabla de origen pendingRecords numRecordsIn numRecordsInPerSecond numBytesIn numBytesInPerSecond Tabla de dimensiones Ninguna Tabla de destino (para Ververica Runtime (VVR) 6.0.6 y versiones posteriores) numRecordsOut numRecordsOutPerSecond Nota Para obtener más información sobre estas métricas, consulte Metrics.
Tipo de API	DataStream API y SQL
Actualización o eliminación de datos en una tabla de destino	Compatible

Requisitos previos

Ha creado un índice de Elasticsearch. Para obtener más información, consulte Getting started.
Ha configurado una lista de permitidos de direcciones IP públicas o privadas para la instancia de Elasticsearch. Para obtener más información, consulte Manage IP address whitelists.

Límites

Las tablas de origen y las tablas de dimensiones admiten Elasticsearch 6.8.x o versiones posteriores.

Nota
El uso de Elasticsearch 8.x con tablas de origen y de dimensiones requiere VVR 11.6 o versiones posteriores.
Las tablas de destino solo admiten Elasticsearch 6.x, 7.x y 8.x.
Solo se admiten tablas de origen completas de Elasticsearch; no se admiten las incrementales.

Sintaxis

Tabla de origen

Elasticsearch 8.x

CREATE TABLE elasticsearch_source(
  name STRING,
  location STRING,
  value FLOAT
) WITH (
  'connector' ='elasticsearch-8',
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>'
);

Otras versiones

CREATE TABLE elasticsearch_source(
  name STRING,
  location STRING,
  value FLOAT
) WITH (
  'connector' ='elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'indexName' = '<yourIndexName>'
);

Tabla de dimensiones
Elasticsearch 8.x
```
CREATE TABLE es_dim(
  field1 STRING, -- Must be of the STRING type when used as a key for a JOIN.
  field2 FLOAT,
  field3 BIGINT,
  PRIMARY KEY (field1) NOT ENFORCED
) WITH (
  'connector' ='elasticsearch-8',
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>'
);
```
Otras versiones
```
CREATE TABLE es_dim(
  field1 STRING, -- Must be of the STRING type when used as a key for a JOIN.
  field2 FLOAT,
  field3 BIGINT,
  PRIMARY KEY (field1) NOT ENFORCED
) WITH (
  'connector' ='elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'indexName' = '<yourIndexName>'
);
```
Nota
- Si se especifica una clave primaria, solo un campo puede ser la clave de unión, y debe ser el ID del documento en el índice de Elasticsearch correspondiente.
- Si no se especifica ninguna clave primaria, puede usar uno o más campos como claves de unión. Estas claves deben ser campos presentes en los documentos de Elasticsearch correspondientes.
- Para los campos STRING, el conector agrega el sufijo .keyword a los nombres de campo de forma predeterminada por motivos de compatibilidad. Si esto impide la coincidencia con campos TEXT en Elasticsearch, establezca la opción ignoreKeywordSuffix en true.
Tabla de destino
```
CREATE TABLE es_sink(
  user_id   STRING,
  user_name   STRING,
  uv BIGINT,
  pv BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'elasticsearch-7', -- If you use Elasticsearch 6.x, set this to 'elasticsearch-6'.
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>'
);
```
Nota
- Una tabla de destino de Elasticsearch opera en upsert mode o en append mode, dependiendo de si se define una clave primaria.
  - Si se define una clave primaria, su valor se utiliza como ID del documento. La tabla de destino opera entonces en upsert mode y puede procesar operaciones UPDATE y DELETE.
  - Si no se define ninguna clave primaria, Elasticsearch genera automáticamente un ID de documento aleatorio. La tabla de destino opera entonces en append mode y solo puede consumir mensajes INSERT.
- Los tipos de datos como BYTES, ROW, ARRAY y MAP no tienen una representación de cadena correspondiente. Por lo tanto, no puede utilizar campos de estos tipos de datos como clave primaria.
- Los campos en la DDL corresponden a campos en un documento de Elasticsearch. No puede escribir metadatos, como el ID del documento, en la tabla de destino porque el clúster de Elasticsearch mantiene dichos metadatos.

Opciones WITH

Tabla de origen

Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
connector	El tipo de la tabla de origen.	String	Sí	Ninguna	Valores válidos: `elasticsearch` o `elasticsearch-8`. Nota Solo VVR 11.6 o versiones posteriores admiten el valor `elasticsearch-8`.
endPoint	La dirección del servidor del clúster de Elasticsearch.	String	Sí	Ninguna	Nombre de opción heredado.
hosts	La dirección del servidor del clúster de Elasticsearch.	String	Sí	Ninguna	Se usa con `elasticsearch-8`.
indexName	El nombre del índice.	String	Sí	Ninguna	Nombre de opción heredado.
index	El nombre del índice.	String	Sí	Ninguna	Para usar con `elasticsearch-8`.
accessId	El nombre de usuario para la autenticación.	String	No	Ninguna	De forma predeterminada, este parámetro está vacío y no se realiza la autenticación. Si especifica accessId, debe especificar un valor no vacío para accessKey. Nota `elasticsearch-8` se actualizó para usar los parámetros `username` y `password`. Importante Para evitar la exposición de su nombre de usuario y contraseña, se recomienda utilizar variables de proyecto. Para obtener más información, consulte project variables.
username	El nombre de usuario para la autenticación.	String	No	Ninguna
accessKey	La contraseña para la autenticación.	String	No	Ninguna
password	La contraseña para la autenticación.	String	No	Ninguna
typeNames	El nombre del tipo.	String	No	_doc	Se recomienda no configurar esta opción para Elasticsearch 7.0 o versiones posteriores.
batchSize	El número máximo de documentos que se recuperan del clúster de Elasticsearch por cada solicitud de desplazamiento (scroll).	Int	No	2000	Ninguna
keepScrollAliveSecs	El tiempo máximo para mantener activo el contexto de desplazamiento (scroll).	Int	No	3600	Unidad: segundos.

Tabla de destino

Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
connector	El tipo de la tabla de destino.	String	Sí	Ninguna	El valor debe ser `elasticsearch-6`, `elasticsearch-7` o `elasticsearch-8`. Nota Solo VVR 8.0.5 o versiones posteriores admiten el valor `elasticsearch-8`.
hosts	La dirección del servidor del clúster de Elasticsearch.	String	Sí	Ninguna	Ejemplo: `127.0.0.1:XXXX`.
index	El nombre del índice.	String	Sí	Ninguna	La tabla de destino admite índices estáticos y dinámicos: Para un índice estático, el valor debe ser una cadena simple, como `myusers`. Todos los registros se escriben en el índice `myusers`. Para un índice dinámico, puede usar `{field_name}` para hacer referencia a valores de campo del registro y generar dinámicamente el índice de destino. También puede usar `{field_name\|date_format_string}` para convertir valores de campo de los tipos de datos TIMESTAMP, DATE y TIME al formato especificado por `date_format_string`. El parámetro `date_format_string` es compatible con DateTimeFormatter de Java. Por ejemplo, si establece el índice en `myusers-{log_ts\|yyyy-MM-dd}`, un registro con el valor de campo log_ts `2020-03-27 12:25:55` se escribe en el índice `myusers-2020-03-27`.
document-type	El tipo de documento.	String	`elasticsearch-6`: Sí `elasticsearch-7`: No compatible	Ninguna	Cuando el tipo de conector es `elasticsearch-6`, el valor de este parámetro debe coincidir con el valor del parámetro type en Elasticsearch.
username	El nombre de usuario para la autenticación.	String	No	Ninguna	De forma predeterminada, la autenticación está deshabilitada. Si especifica `username`, también debe especificar una `password` no vacía. Importante Para evitar la exposición de su nombre de usuario y contraseña, se recomienda utilizar variables de proyecto. Para obtener más información, consulte project variables.
password	La contraseña para la autenticación.	String	No	Ninguna
document-id.key-delimiter	El delimitador para el ID del documento.	String	No	_	El conector utiliza la clave principal para generar el ID del documento. El conector concatena todos los campos de clave principal en el orden definido en el DDL, mediante el delimitador especificado por document-id.key-delimiter, para crear una cadena de ID de documento para cada fila. Nota Un ID de documento es una cadena de hasta 512 bytes que no contiene espacios.
failure-handler	La directiva de manejo de errores para solicitudes de Elasticsearch con errores.	String	No	fail	Directivas válidas: `fail` (predeterminado): Falla el trabajo si una solicitud presenta errores. `ignore`: Ignora el error y descarta la solicitud. `retry-rejected`: Vuelve a agregar solicitudes que fallaron debido a una cola llena. Un nombre de clase personalizado: Usa una subclase ActionRequestFailureHandler para el manejo de errores.
sink.flush-on-checkpoint	Especifica si se debe vaciar (flush) al guardar un punto de control.	Boolean	No	true	`true`: Valor predeterminado. `false`: Si se deshabilita, el conector no espera a que se confirmen todas las solicitudes pendientes durante un punto de control. Esto significa que el conector no garantiza la entrega de al menos una vez.
sink.bulk-flush.backoff.strategy	Si la operación de vaciado falla debido a un error temporal de solicitud, establezca sink.bulk-flush.backoff.strategy para especificar la estrategia de reintento.	Enum	No	DISABLED	`DISABLED` (predeterminado): No se realizan reintentos. El trabajo falla ante el primer error de solicitud. `CONSTANT`: Estrategia de retroceso constante donde el tiempo de espera entre reintentos siempre es el mismo. `EXPONENTIAL`: Estrategia de retroceso exponencial donde el tiempo de espera entre reintentos aumenta exponencialmente.
sink.bulk-flush.backoff.max-retries	El número máximo de reintentos.	Int	No	Ninguna	Ninguna
sink.bulk-flush.backoff.delay	El retraso entre intentos de reintento.	Duration	No	Ninguna	Para una estrategia de `constant backoff`, este valor representa el retraso entre cada reintento. Para una estrategia de `exponential backoff`, este valor es el retraso base inicial.
sink.bulk-flush.max-actions	El número máximo de acciones almacenadas en búfer para cada solicitud masiva (bulk).	Int	No	1000	Un valor de 0 deshabilita esta función.
sink.bulk-flush.max-size	El tamaño máximo de memoria del búfer de solicitudes.	String	No	2 MB	La unidad es MB. El valor predeterminado es 2 MB. Un valor de 0 deshabilita esta función.
sink.bulk-flush.interval	El intervalo de vaciado.	Duration	No	1s	La unidad es segundos. El valor predeterminado es 1s. Un valor de 0s deshabilita esta función.
connection.path-prefix	Una cadena que se antepone a cada ruta de comunicación REST.	String	No	Ninguna	Ninguna
retry-on-conflict	El número máximo de reintentos para una operación de actualización si ocurre un conflicto de versión. Si el número de reintentos supera este valor, el trabajo falla con una excepción.	Int	No	0	Nota Esta opción solo es compatible con VVR 4.0.13 o versiones posteriores. Esta opción solo surte efecto cuando se define una clave principal.
routing-fields	Especifica uno o más nombres de campos de Elasticsearch utilizados para enrutar un documento a un shard específico.	String	No	Ninguna	Separe varios nombres de campos con punto y coma (;). Si los datos de un campo están vacíos, el campo se establece en null. Nota Esta opción solo es compatible con VVR 8.0.6 o versiones posteriores, para `elasticsearch-7` y `elasticsearch-8`.
sink.delete-strategy	Configura cómo maneja el destino un mensaje de retracción (-D para DELETE o -U para UPDATE_BEFORE).	Enum	No	DELETE_ROW_ON_PK	Estrategias válidas: `DELETE_ROW_ON_PK` (predeterminado): Ignora los mensajes -U, pero elimina la fila (documento) correspondiente a la clave principal cuando recibe un mensaje -D. `IGNORE_DELETE`: Ignora tanto los mensajes -U como los -D. No se produce ninguna retracción en el destino de Elasticsearch. NON_PK_FIELD_TO_NULL: Ignora los mensajes -U. Sin embargo, cuando se recibe un mensaje -D, esta configuración modifica la fila (documento) de la clave principal: el valor de la clave principal permanece igual y todos los demás valores que no son de clave principal en el esquema de la tabla se establecen en NULL. Esto se usa principalmente para actualizaciones parciales cuando varios destinos escriben en la misma tabla de Elasticsearch simultáneamente. `CHANGELOG_STANDARD`: Similar a `DELETE_ROW_ON_PK`, pero también elimina la fila (documento) correspondiente a la clave principal cuando se recibe un mensaje -U. Nota Esta opción solo es compatible con VVR 8.0.8 o versiones posteriores.
sink.ignore-null-when-update	Al actualizar datos, especifica si se debe actualizar un campo a `null` o dejarlo sin cambios si el valor del campo entrante es `null`.	BOOLEAN	No	false	Valores válidos: `true`: El campo no se actualiza. Este valor solo es compatible cuando se establece una clave principal para la tabla de Flink y el formato de datos de Elasticsearch es JSON. `false`: El campo se actualiza a null. Nota Esta opción solo es compatible con VVR 11.1 o versiones posteriores.
connection.request-timeout	El tiempo de espera para solicitar una conexión desde el administrador de conexiones.	Duration	No	Ninguna	Ejemplo: `'connection.request-timeout' = '1 min' -- 1 minute 'connection.request-timeout' = '500ms' -- 500 milliseconds` Nota Esta opción solo es compatible con VVR 11.7 o versiones posteriores.
connect.timeout	El tiempo de espera para establecer una conexión.	Duration	No	Ninguna	Ejemplo: `'connect.timeout' = '1 min' -- 1 minute 'connect.timeout' = '500ms' -- 500 milliseconds` Nota Esta opción solo es compatible con VVR 11.7 o versiones posteriores.
socket.timeout	El tiempo de espera para recibir datos, que corresponde al periodo máximo de inactividad entre dos paquetes de datos consecutivos.	Duration	No	Ninguna	Ejemplo: `'socket.timeout' = '1 min' -- 1 minute 'socket.timeout' = '500ms' -- 500 milliseconds` Nota Esta opción solo es compatible con VVR 11.7 o versiones posteriores.
connection.keep-alive	La duración máxima que una conexión puede permanecer inactiva antes de que el sistema la cierre. Si no se establece esta opción, el encabezado de respuesta `Keep-Alive` del servidor determina la duración. Si el servidor no envía un encabezado `Keep-Alive`, la conexión permanece activa indefinidamente.	Duration	No	Ninguna	Ejemplo: `'connection.keep-alive' = '1 min' -- 1 minute 'connection.keep-alive' = '500ms' -- 500 milliseconds` Nota Esta opción solo es compatible con VVR 11.7 o versiones posteriores.
sink.bulk-flush.update.doc_as_upsert	Especifica si se debe tratar el documento como un documento upsert en una solicitud de actualización.	BOOLEAN	No	false	Valores válidos: `true`: Establece el campo `doc_as_upsert` de la solicitud de actualización en true. `false`: Rellena el campo `upsert` de la solicitud de actualización con el documento. Según https://github.com/elastic/elasticsearch/issues/105804, los pipelines de ingesta de Elasticsearch no admiten actualizaciones parciales para solicitudes de actualización masiva. Si desea usar un pipeline de ingesta, establezca esta opción en true. Nota Esta opción solo es compatible con VVR 11.5 o versiones posteriores.

Tabla de dimensiones

Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
connector	El tipo de la tabla de dimensiones.	String	Sí	Ninguna	Valores válidos: `elasticsearch` o `elasticsearch-8`. Nota Solo VVR 11.6 o versiones posteriores admiten el valor `elasticsearch-8`.
endPoint	La dirección del servidor del clúster de Elasticsearch.	String	Sí	Ninguna	Nombre de opción heredado.
hosts	La dirección del servidor del clúster de Elasticsearch.	String	Sí	Ninguna	Para usar con `elasticsearch-8`.
indexName	El nombre del índice.	String	Sí	Ninguna	Nombre de opción heredado.
index	El nombre del índice.	String	Sí	Ninguna	Para usar con `elasticsearch-8`.
accessId	El nombre de usuario para la autenticación.	String	No	Ninguna	De forma predeterminada, este parámetro está vacío y no se realiza la autenticación. Si especifica accessId, debe especificar un valor no vacío para accessKey. Nota `elasticsearch-8` ahora usa los parámetros `username` y `password`. Importante Para evitar la exposición de su nombre de usuario y contraseña, se recomienda utilizar variables de proyecto. Para obtener más información, consulte project variables.
username	El nombre de usuario para la autenticación.	String	No	Ninguna
accessKey	La contraseña para la autenticación.	String	No	Ninguna
password	La contraseña para la autenticación.	String	No	Ninguna
typeNames	El nombre del tipo.	String	No	_doc	Se recomienda no configurar esta opción para Elasticsearch 7.0 o versiones posteriores.
maxJoinRows	El número máximo de filas para unir (join) en una sola búsqueda.	Integer	No	1024	Ninguna
cache	La estrategia de caché.	String	No	Ninguna	Valores válidos: `ALL`: Almacena en caché todos los datos de la tabla de dimensiones. Antes de que comience el trabajo, el sistema carga todos los datos de la tabla de dimensiones en la caché. Las búsquedas posteriores se atienden directamente desde la caché. Si no se encuentra una clave, el sistema considera que no existe. El sistema recarga toda la caché cuando expira el TTL. `LRU`: Almacena en caché una parte de los datos de la tabla de dimensiones. Cuando llega un registro de la tabla de origen, el sistema primero busca los datos en la caché. Si hay un fallo de caché, consulta la tabla de dimensiones física. `None`: Sin almacenamiento en caché.
cacheSize	El tamaño de la caché, especificado como número de filas.	Long	No	100000	El parámetro cacheSize solo surte efecto cuando se selecciona la directiva de caché LRU para cache.
cacheTTLMs	El tiempo de vida (TTL) de la caché.	Long	No	Long.MAX_VALUE	Unidad: milisegundos. El comportamiento de cacheTTLMs depende de la configuración de cache: Cuando cache es `LRU`, cacheTTLMs es el TTL para las entradas de la caché. De forma predeterminada, las entradas no caducan. Cuando cache es `ALL`, cacheTTLMs es el intervalo para recargar la caché. De forma predeterminada, la caché no se recarga.
ignoreKeywordSuffix	Especifica si se debe ignorar el sufijo .keyword que se agrega automáticamente a los campos STRING.	Boolean	No	false	Por motivos de compatibilidad, Flink convierte los tipos `Text` de Elasticsearch a `STRING` y agrega un sufijo .keyword al nombre del campo de forma predeterminada. Valores válidos: `true`: Ignora el sufijo. Si el sufijo impide la coincidencia con campos de tipo `Text` en Elasticsearch, establezca esta opción en `true`. `false`: No ignora el sufijo.
cacheEmpty	Especifica si se deben almacenar en caché los resultados vacíos de las búsquedas en la tabla de dimensiones física.	Boolean	No	true	El parámetro cacheEmpty solo es efectivo cuando cache utiliza la directiva de caché LRU.
queryMaxDocs	Para tablas de dimensiones sin clave principal, este es el número máximo de documentos que devuelve el servidor de Elasticsearch para cada consulta de búsqueda.	Integer	No	10000	El valor predeterminado de 10,000 coincide con el número máximo de documentos que un servidor de Elasticsearch puede devolver por consulta. Este valor no puede superar dicho límite. Nota Esta opción solo es compatible con VVR 8.0.8 o versiones posteriores. Esta opción solo surte efecto para tablas de dimensiones sin clave principal, ya que los datos en las tablas con clave principal son únicos. Un valor predeterminado alto ayuda a garantizar la corrección de la consulta, pero aumenta el uso de memoria durante las consultas de Elasticsearch. Si experimenta problemas de memoria, puede reducir este valor para optimizar el uso de memoria.

Mapeo de tipos

Flink analiza los datos de Elasticsearch como JSON. Para obtener más información, consulte mapeo de tipos de datos.

Ejemplos

Ejemplo de tabla de origen

CREATE TEMPORARY TABLE elasticsearch_source (
  name STRING,
  location STRING,
  `value` FLOAT
) WITH (
  'connector' ='elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'indexName' = '<yourIndexName>',
  'typeNames' = '<yourTypeName>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  name STRING,
  location STRING,
  `value` FLOAT
) WITH (
  'connector' ='blackhole'
);

INSERT INTO blackhole_sink
SELECT name, location, `value`
FROM elasticsearch_source;

Ejemplo de tabla de dimensiones

CREATE TEMPORARY TABLE datagen_source (
  id STRING, 
  data STRING,
  proctime as PROCTIME()
) WITH (
  'connector' = 'datagen' 
);

CREATE TEMPORARY TABLE es_dim (
  id STRING,
  `value` FLOAT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' ='elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'indexName' = '<yourIndexName>',
  'typeNames' = '<yourTypeName>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  id STRING,
  data STRING,
  `value` FLOAT
) WITH (
  'connector' = 'blackhole' 
);

INSERT INTO blackhole_sink
SELECT e.*, w.*
FROM datagen_source AS e
JOIN es_dim FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;

Ejemplo de tabla de destino 1

Este ejemplo escribe contenido de texto en Elasticsearch después de text vectorization.

Nota

Cree un mapeo de índices en Elasticsearch con antelación. Establezca el tipo de datos del campo embedding como dense_vector y especifique las dimensiones. De lo contrario, Elasticsearch podría inferirlo como un tipo de arreglo regular.

    CREATE TEMPORARY TABLE datagen_source (
      id STRING,
      content STRING,
      embedding ARRAY<FLOAT>
    ) WITH (
      'connector' = 'datagen'
    );

    CREATE TEMPORARY TABLE es_sink (
      id STRING,
      content STRING,
      embedding ARRAY<FLOAT>,
      PRIMARY KEY (id) NOT ENFORCED -- The primary key is optional. If you define a primary key, its value becomes the document ID. Otherwise, a random document ID is generated.
    ) WITH (
      'connector' = 'elasticsearch-8',
      'hosts' = '<yourHosts>',
      'index' = '<yourIndex>',
      'username' ='${secret_values.ak_id}',
      'password' ='${secret_values.ak_secret}'
    );

    INSERT INTO es_sink
    SELECT id, content, embedding
    FROM datagen_source;

Ejemplo de tabla de destino 2

El conector admite la escritura de tipos complejos como ROW, ARRAY y MAP en Elasticsearch.

CREATE TEMPORARY TABLE datagen_source(  
  id STRING,
    details ROW<  
        name STRING,  
        ages ARRAY<INT>,  
        attributes MAP<STRING, STRING>  
    >
) WITH (  
    'connector' = 'datagen'
);

CREATE TEMPORARY TABLE es_sink (
  id STRING,
    details ROW<  
        name STRING,  
        ages ARRAY<INT>,  
        attributes MAP<STRING, STRING>  
    >, 
  PRIMARY KEY (id) NOT ENFORCED  -- The primary key is optional. If you define a primary key, its value becomes the document ID. Otherwise, a random document ID is generated.
) WITH (
  'connector' = 'elasticsearch-6',
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>',
  'document-type' = '<yourElasticsearch.types>',
  'username' ='${secret_values.ak_id}',
  'password' ='${secret_values.ak_secret}'
);

INSERT INTO es_sink
SELECT id, details
FROM datagen_source;