Elasticsearch:Use Real-time Computing to process and sync data to ES

Step 1: Create a Flink job

1. Log on to the Realtime Compute for Apache Flink console.

2. Create a Realtime Compute for Apache Flink job. For more information, see the Job Development > Development section of the Blink SQL Development Guide in the Alibaba Cloud Blink Exclusive Mode (Phased-Out for Alibaba Cloud) document.

Step 2: Create the SLS source table

Define a source table that reads from SLS:

create table sls_stream(
  a int,
  b int,
  c VARCHAR
)
WITH (
  type ='sls',
  endPoint ='<yourEndpoint>',
  accessId ='<yourAccessId>',
  accessKey ='<yourAccessKey>',
  startTime = '<yourStartTime>',
  project ='<yourProjectName>',
  logStore ='<yourLogStoreName>',
  consumerGroup ='<yourConsumerGroupName>'
);

Replace the placeholders with your actual values:

Step 3: Create the Elasticsearch sink table

Define a sink table that writes to Elasticsearch:

CREATE TABLE es_stream_sink(
  a int,
  cnt BIGINT,
  PRIMARY KEY(a)
)
WITH(
  type ='elasticsearch-7',
  endPoint = 'http://<instanceid>.public.elasticsearch.aliyuncs.com:<port>',
  accessId = '<yourAccessId>',
  accessKey = '<yourAccessSecret>',
  index = '<yourIndex>',
  typeName = '<yourTypeName>'
);

Replace the placeholders with your actual values:

Step 4: Write the business logic

Insert the aggregated results into the sink table:

INSERT INTO es_stream_sink
SELECT
  a,
  count(*) as cnt
FROM sls_stream GROUP BY a

Step 5: Publish and start the job

Publish and start the job. Data from SLS is aggregated and written to Elasticsearch.