Elasticsearch 连接器提供了对 Elasticsearch 数据分析引擎的写入能力，仅支持做数据结果表。流式计算 Flink 版支持 Elasticsearch-6 和 Elasticsearch-7 两个版本，部分配置存在差异，请注意区分。

DDL 定义 #

CREATE TABLE elasticsearch_sink (
  user_id STRING,
  user_name STRING,
  uv BIGINT,
  pv BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED  
)  WITH (
  'connector' = 'elasticsearch-6',
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>',
  'document-type' = '<yourDocumentType>',
  'username' ='<yourUsername>',
  'password' ='<yourPassword>'
);

WITH 参数 #

示例代码 #

• Elasticsearch-6 结果表

  create table orders (
      order_id bigint,
      order_product_id bigint,
      order_customer_id bigint,
      order_status varchar,
      order_update_time as localtimestamp
    )
  WITH
    (
      'connector' = 'datagen',
      'rows-per-second' = '100',
      'fields.order_status.length' = '3',
      'fields.order_id.min' = '1',
      'fields.order_id.max' = '10000',
      'fields.order_product_id.min' = '1',
      'fields.order_product_id.max' = '100',
      'fields.order_customer_id.min' = '1',
      'fields.order_customer_id.max' = '1000'
    );
  
  
  create table es_sink (
      order_id bigint,
      order_product_id bigint,
      order_customer_id bigint,
      order_status varchar,
      order_update_time timestamp
    )
  WITH
    (
      'connector' = 'elasticsearch-6',
      'hosts' = 'https://elasticsearch-wu****l8e.escloud.ivolces.com:9200',
      'index' = 'test_orders',
      'document-type' = 'doc',
      'username' = 'admin', 
      'password' = 'cd****456' 
    );
  
  insert into es_sink
  select * from orders;
  

• Elasticsearch-7 结果表

  create table orders (
      order_id bigint,
      order_product_id bigint,
      order_customer_id bigint,
      order_status varchar,
      order_update_time as localtimestamp
    )
  WITH
    (
      'connector' = 'datagen',
      'rows-per-second' = '1',
      'fields.order_status.length' = '3',
      'fields.order_id.min' = '1',
      'fields.order_id.max' = '10000',
      'fields.order_product_id.min' = '1',
      'fields.order_product_id.max' = '100',
      'fields.order_customer_id.min' = '1',
      'fields.order_customer_id.max' = '1000'
    );
  
  
  create table es_sink (
      order_id bigint,
      order_product_id bigint,
      order_customer_id bigint,
      order_status varchar,
      order_update_time timestamp
    )
  WITH
    (
      'connector' = 'elasticsearch-7',
      'hosts' = 'https://elasticsearch-wu****l8e.escloud.ivolces.com:9200',
      'index' = 'test_orders',
      'username' = 'admin', 
      'password' = 'cd****456' 
    );
  
  insert into es_sink
  select * from orders;