Realtime Compute for Apache Flink:Tair (Enterprise)

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,   -- document content
  p STRING    -- mapping definition
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  index_name STRING,  -- TairSearch index name
  doc_id     STRING,  -- document ID
  doc        STRING,  -- document body (JSON)
  mapping    STRING,  -- index mapping
  PRIMARY KEY (index_name) NOT ENFORCED
) WITH (
  'connector' = 'tair',
  'mode'      = 'tairsearch',
  'host'      = '${tairHost}',
  'port'      = '${tairPort}',
  'password'  = '${password}'
);

INSERT INTO tair_output
SELECT
  'index' AS index_name,
  v AS doc_id,
  p AS doc,
  '{"mappings":{"_source":{"enabled":true},"properties":{"product_id":{"type":"keyword","ignore_above":128},"product_name":{"type":"text"}}}}' AS mapping
FROM datagen_stream;

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,   -- key
  p STRING    -- increment value per record
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  key  STRING,  -- TairString key
  step STRING,  -- increment value (column referenced by incrValue)
  PRIMARY KEY (key) NOT ENFORCED
) WITH (
  'connector' = 'tair',
  'mode'      = 'tairstring',
  'host'      = '${tairHost}',
  'port'      = '${tairPort}',
  'password'  = '${password}',
  'incrMode'  = 'dynamic_float',
  'incrValue' = 'step'
);

INSERT INTO tair_output
SELECT *
FROM datagen_stream;

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,
  p STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  key STRING,  -- TairString key
  PRIMARY KEY (key) NOT ENFORCED
) WITH (
  'connector' = 'tair',
  'mode'      = 'tairstring',
  'host'      = '${tairHost}',
  'port'      = '${tairPort}',
  'password'  = '${password}',
  'incrMode'  = 'float',
  'incrValue' = '11.11'
);

INSERT INTO tair_output
SELECT v
FROM datagen_stream;

CREATE TEMPORARY TABLE datagen_stream (
  v STRING,
  p STRING,
  s STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE tair_output (
  key   STRING,  -- TairHash key
  field STRING,  -- hash field
  value STRING,  -- field value
  PRIMARY KEY (key) NOT ENFORCED
) WITH (
  'connector'        = 'tair',
  'mode'             = 'tairhash',
  'host'             = '${tairHost}',
  'port'             = '${tairPort}',
  'password'         = '${password}',
  'fieldExpireMode'  = 'millisecond',
  'fieldExpireValue' = '1000'
);

INSERT INTO tair_output
SELECT v, p, s
FROM datagen_stream;