CREATE TABLE elasticsearch6_sink_table (
    id INT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED -- Corresponds to _id in Elasticsearch
) WITH (
    'connector' = 'elasticsearch-6', --Output to Elasticsearch 6
    'username' = '$username', -- optional username
    'password' = '$password', -- optional password
    'hosts' = 'http://10.28.28.94:9200', -- Elasticsearch connection address
    'index' = 'my-index', -- Elasticsearch index name
    'document-type'='_doc', --Document type of Elasticsearch
    'format' = 'json' -- Output data format, currently only support 'json'
);

CREATE TABLE elasticsearch7_sink_table (
    id INT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED -- Corresponds to the _id in Elasticsearch
) WITH (
    'connector' = 'elasticsearch-7', -- output to Elasticsearch 7
    'username' = '$username', -- optional username
    'password' = '$password', -- optional password
    'hosts' = 'http://10.28.28.94:9200', -- Elasticsearch connection address
    'index' = 'my-index', --Index name of Elasticsearch
    'format' = 'json' -- Output data format, currently only support 'json'
);

CREATE TABLE elasticsearch7_source_table (
  id bigint,
  event_date int,
  app int,
  primary key (id) not enforced
) with (
  -required parameter
  'connector' = 'es-source',
  'endPoint' = '127.0.0.1', -- Elasticsearch connection ip
  'accessId' = 'elastic', -- Username
  'accessKey' = 'PASSWORD', -- Password
  'indexName' = 'my-index', -- Elasticsearch index name
  'format' = 'json', -- data format, support only 'json'
  -optional parameter
  'scheme' => 'http', // connection protocol
  'port' = '9200', -- Port
  'batchSize' = '2000', -- Maximum number of documents retrieved per scroll request from the Elasticsearch cluster
  'keepScrollAliveSecs' = '60' -- maximum duration for scroll context retention, in minutes
);

CREATE TABLE datagen_source_table (
    id INT, 
    name STRING 
) WITH ( 
    'connector' = 'datagen',
    'rows-per-second'='1'  -- Number of data records generated per second
);

CREATE TABLE elasticsearch7_sink_table (
    id INT,
    name STRING
) WITH (
    'connector' = 'elasticsearch-7',       -- output to Elasticsearch 7
    'username' = '$username',              -- optional username
    'password' = '$password',              -- optional password
    'hosts' = 'http://10.28.28.94:9200',   -- Elasticsearch connection address
    'index' = 'my-index',                  -- Name of the Elasticsearch index
    'sink.bulk-flush.max-actions' = '1000',   -- Data refresh frequency
    'sink.bulk-flush.interval' = '1s'      -- Data refresh interval
    'format' = 'json'                      -- Output data format, currently only support 'json'
);

INSERT INTO elasticsearch7_sink_table select * from datagen_source_table;

CREATE TABLE elasticsearch7_source_table (
  id bigint,
  event_date int,
  app int,
  primary key (id) not enforced
) with (
  -required parameter
  'connector' = 'es-source',
  'endPoint' = '127.0.0.1', -- Elasticsearch connection ip
  'accessId' = 'elastic', -- Username
  'accessKey' = 'PASSWORD', -- Password
  'indexName' = 'my-index', -- Elasticsearch index name
  ': 'format' = 'json', -- Data format, supports only 'json'
  -optional parameter
  'scheme' => 'http', // connection protocol
  'port' = '9200', -- Port
  'batchSize' = '2000', -- Maximum number of documents retrieved per scroll request from the ES cluster
  'keepScrollAliveSecs' = '60' -- maximum duration for scroll context retention, in minutes
);

CREATE TABLE logger_sink (
  id bigint,
  event_date int,
  app int,
  primary key (id) not enforced
) with (
  'connector' = 'logger'
);

INSERT INTO logger_sink SELECT * from elasticsearch7_source_table;