Elasticsearch

背景信息

阿里云Elasticsearch兼容开源Elasticsearch的功能，以及Security、Machine Learning、Graph、APM等商业功能，致力于数据分析、数据搜索等场景服务。为您提供企业级权限管控、安全监控告警、自动报表生成等场景服务。

前提条件

• 已创建Elasticsearch索引，详情请参见创建示例。

• 已配置Elasticsearch公网或私网访问白名单，详情请参见配置实例公网或私网访问白名单。

使用限制

• 源表和维表支持大于等于6.8.x，但小于8.x版本的Elasticsearch。

• 结果表仅支持Elasticsearch 6.x、7.x和8.x版本。

• 仅支持全量Elasticsearch源表，不支持增量Elasticsearch源表。

语法结构

• 源表

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

• 维表

  CREATE TABLE es_dim(
    field1 STRING, --作为JOIN时的Key，必须为STRING类型。
    field2 FLOAT,
    field3 BIGINT,
    PRIMARY KEY (field1) NOT ENFORCED
  ) WITH (
    'connector' ='elasticsearch',
    'endPoint' = '<yourEndPoint>',
    'indexName' = '<yourIndexName>'
  );

  说明

  • 如果指定主键，则维表JOIN时的Key（字段）有且只能有一个，且必须为Elasticsearch对应索引的文档ID。

  • 如果不指定主键，则维表JOIN时的Key可以有一个或多个，需要为Elasticsearch对应索引的文档中的字段。

  • 对于String类型，为了保持兼容性，默认会对表中字段名增加.keyword后缀。如果因此无法匹配到Elasticsearch中的Text字段，可以将配置项ignoreKeywordSuffix配置为true。

• 结果表

  CREATE TABLE es_sink(
    user_id   STRING,
    user_name   STRING,
    uv BIGINT,
    pv BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
  ) WITH (
    'connector' = 'elasticsearch-7', -- 如果是Elasticsearch 6.x版本,填写elasticsearch-6
    'hosts' = '<yourHosts>',
    'index' = '<yourIndex>'
  );

  说明

  • Elasticsearch结果表会根据是否定义了主键，确定是在upsert模式或append模式下工作。

    • 如果定义了主键，则主键必须为文档ID，Elasticsearch结果表将在upsert模式下工作，该模式可以处理包含UPDATE和DELETE的消息。

    • 如果未定义主键，Elasticsearch将自动生成随机的文档ID，Elasticsearch结果表将在append模式工作，该模式只能消费INSERT消息。

  • 某些类型（例如BYTES、ROW、ARRAY和MAP等）由于没有对应的字符串表示形式，所以不允许其作为主键字段。

  • DDL中的字段均对应Elasticsearch文档中的字段，不支持将文档ID等Meta信息写入Elasticsearch结果表中，因为文档ID等Meta信息由Elasticsearch实例侧维护。

WITH参数

类型映射

Flink以JSON来解析Elasticsearch数据，详情请参见数据类型映射关系。

使用示例

• 源表示例

  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;

• 维表示例

  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;

• 结果表示例1

  CREATE TEMPORARY TABLE datagen_source (
    id STRING, 
    name STRING,
    uv BIGINT
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE es_sink (
    user_id STRING,
    user_name STRING,
    uv BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED -- 主键可选，如果定义了主键，则作为文档ID，否则文档ID将为随机值。
  ) 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, name, uv
  FROM datagen_source;

• 结果表示例2

  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  -- 主键可选，如果定义了主键，则作为文档ID，否则文档ID将为随机值。
  ) 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;