Consume log data with Realtime Compute for Apache Flink - Simple Log Service

Supported capabilities

Realtime Compute for Apache Flink supports the following log consumption capabilities.

Category	Description
Supported type	Source table and result table.
Running mode	Streaming mode only.
Metric	Not supported.
Data format	None.
API type	SQL.
Result table operations	Insert only. Updates and deletions are not supported.

Start consuming log data by following Getting started with a Flink SQL deployment.

Prerequisites

To use a RAM user or RAM role, ensure it has the required permissions on the Realtime Compute for Apache Flink console. For more information, see Permissions.
A Realtime Compute for Apache Flink workspace is created. For more information, see Activate Realtime Compute for Apache Flink.
A project and a Logstore are created. Create a project and a Logstore.

Limitations

Data ingestion YAML requires Ververica Runtime (VVR) 11.1 or later.
The SLS connector provides at-least-once delivery guarantees — records are never lost, but may be delivered more than once (for example, after a job restart from a checkpoint). Design your downstream systems to handle duplicates.
Keep source concurrency at or below the number of shards. Higher concurrency wastes resources without improving throughput.
In VVR 8.0.5 and earlier, automatic failover may stop working if the shard count changes. After a shard change, verify that all shards are still being consumed.
The sink table supports insert operations only. Updating or deleting existing data is not supported.

Create a source table and a result table

Important

Develop a complete SQL draft that contains a source table and a result table before consuming log data. After processing, results are inserted into the result table by using the INSERT INTO statement.

For more information about how to develop an SQL draft in Realtime Compute for Apache Flink, see Job development overview.

Simple Log Service stores log data in real time, and Realtime Compute for Apache Flink reads it in streaming mode. Example log entry:

__source__:  11.85.*.199
__tag__:__receive_time__:  1562125591
__topic__:  test-topic
request_method:  GET
status:  200

Sample code

The following SQL draft example consumes Simple Log Service data in Realtime Compute for Apache Flink.

Important

Enclose table, column, or reserved field names in backticks (`) if they conflict with each other.

CREATE TEMPORARY TABLE sls_input(
  request_method STRING,
  status BIGINT,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` BIGINT METADATA VIRTUAL,
   __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'starttime' = '2023-08-30 00:00:00',
  'project' ='sls-test',
  'logstore' ='sls-input'
);

CREATE TEMPORARY TABLE sls_sink(
  request_method STRING,
  status BIGINT,
  `__topic__` STRING,
  `__source__` STRING,
  `__timestamp__` BIGINT ,
  receive_time BIGINT
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = '${ak_id}',
  'accessKey' = '${ak_secret}',
  'project' ='sls-test',
  'logstore' ='sls-output'
);

INSERT INTO sls_sink
SELECT 
  request_method,
  status,
  `__topic__` ,
  `__source__` ,
  `__timestamp__` ,
  cast(__tag__['__receive_time__'] as bigint) as receive_time
FROM sls_input;

WITH parameters

Common parameters

Parameter	Data type	Required	Default	Description
`connector`	String	Yes	—	Fixed value: `sls`
`endPoint`	String	Yes	—	Private network endpoint for SLS. See Service endpoint. Note Realtime Compute for Apache Flink does not support public network access by default. Use Alibaba Cloud NAT Gateway for VPC-to-internet communication. If public access is required, use HTTPS and enable Global Accelerator (GA) for SLS.
`project`	String	Yes	—	SLS project name
`logStore`	String	Yes	—	Logstore or Metricstore name. Logstores and Metricstores use the same consumption method.
`accessId`	String	Yes	—	AccessKey ID. To avoid exposing credentials, use project variables. See How do I view my AccessKey ID and AccessKey secret? and Project variables.
`accessKey`	String	Yes	—	AccessKey secret

Source parameters

Start reading position

Use startupMode to control where consumption begins. The available modes are:

Mode	Behavior
`timestamp` (default)	Start from the time specified by `startTime`. If `startTime` is not set, consumption starts from the current time.
`latest`	Start from the latest available offset
`earliest`	Start from the earliest available offset
`consumer_group`	Resume from the offset saved in the consumer group. If no offset is recorded for a shard, falls back to `earliest`.

Related parameters:

startTime (String, default: current time) — Start time in yyyy-MM-dd hh:mm:ss format. Takes effect only when startupMode is timestamp. Times are based on the SLS __receive_time__ attribute, not __timestamp__.
stopTime (String, default: none) — End time in yyyy-MM-dd hh:mm:ss format. Set this only for historical log consumption. If set to a future time, the stream may end prematurely when no new logs arrive — without an error message. To exit Flink after consumption finishes, also set exitAfterFinish=true.
consumerGroup (String, default: none) — Records consumption progress. Each Flink job must use a unique consumer group; sharing one across jobs does not coordinate offsets because Flink does not use the SLS consumer group for partition assignment — each consumer reads all data independently.
consumeFromCheckpoint (String, default: false) — Not supported in VVR 11.1 and later. In earlier versions, set to true to resume from the offset saved in the specified consumer group; if no checkpoint exists, falls back to startTime. For VVR 11.1 and later, use startupMode=consumer_group instead.

VVR versions earlier than 11.1 do not support startupMode=consumer_group. Use consumeFromCheckpoint=true with a consumerGroup to resume from a saved offset.

Other source parameters

Parameter	Data type	Required	Default	Description
`enableNewSource`	Boolean	No	`false` (VVR 11.1+: `true`)	Enable the FLIP-27 source interface. The new source adapts automatically to shard changes and distributes shards evenly. Requires VVR 8.0.9 or later. Important Changing this parameter prevents the job from resuming its previous state. To resume from a historical offset, first run the job with `consumerGroup` set, then restart without state and use `startupMode=consumer_group`. If read-only shards exist, some tasks may request additional unprocessed shards after completing read-only shard consumption, causing uneven distribution. Adjust concurrency, optimize task scheduling, or merge small shards to reduce the effect.
`shardDiscoveryIntervalMs`	Long	No	`60000`	How often (in milliseconds) to detect shard changes. Minimum: 60,000 ms. Set to a negative value to disable. Takes effect only when `enableNewSource=true`. Requires VVR 8.0.9 or later.
`maxRetries`	String	No	`3`	Number of retries after a failed SLS read
`batchGetSize`	String	No	`100`	Log groups to fetch per request. Maximum: 1,000. Exceeding this limit causes an error.
`exitAfterFinish`	String	No	`false`	Whether to exit the Flink job after consumption completes. Set to `true` when consuming historical logs with `stopTime`.
`query`	String	No	—	Deprecated in VVR 11.3 (still compatible). SPL filter applied before Flink reads the data, reducing data volume and cost. Example: `'query' = '* \| where request_method = ''GET'''`. Use SPL syntax. See SPL syntax. Requires VVR 8.0.1 or later. This feature incurs SLS charges. For more information, see Pricing.
`processor`	String	No	—	SLS consumer processor. Takes precedence over `query` when both are set. Filters data before Flink reads it. Example: `'processor' = 'test-filter-processor'`. Use SPL syntax. See SPL syntax and Manage consumer processors. Requires VVR 11.3 or later. This feature incurs SLS charges. For more information, see Pricing.

Sink parameters

Parameter	Data type	Required	Default	Description
`topicField`	String	No	—	Name of a table field whose value overrides the `__topic__` metadata. The field must exist in the table.
`timeField`	String	No	Current time	Name of a table field whose value overrides the `__timestamp__` metadata. The field must exist in the table and have the INT type.
`sourceField`	String	No	—	Name of a table field whose value overrides the `__source__` metadata (for example, the machine IP). The field must exist in the table.
`partitionField`	String	No	—	Name of a field used to route records to shards by hash value. Records with the same hash value go to the same shard. If unset, records are written to shards randomly.
`buckets`	String	No	`64`	Number of buckets for hash-based routing when `partitionField` is set. Valid values: integers from 1 to 256 that are powers of two. Must be greater than or equal to the shard count — otherwise some shards receive no data.
`flushIntervalMs`	String	No	`2000`	How often (in milliseconds) data is written to SLS
`writeNullProperties`	Boolean	No	`true`	Whether to write null values as empty strings (`true`) or skip null fields (`false`). Requires VVR 8.0.6 or later.

Extract attribute fields

In addition to log fields and custom fields, Realtime Compute for Apache Flink can extract the following attribute fields.

Field	Type	Description
__source__	STRING METADATA VIRTUAL	The message source.
__topic__	STRING METADATA VIRTUAL	The message topic.
__timestamp__	BIGINT METADATA VIRTUAL	The log time.
__tag__	MAP<VARCHAR, VARCHAR> METADATA VIRTUAL	The message tag. For the `"__tag__:__receive_time__":"1616742274"` attribute, the `__receive_time__` and 1616742274 fields are recorded as key-value pairs in a map. You can include `__tag__['__receive_time__']` in an SQL statement to query the tag.

To extract attribute fields, define headers in your SQL statement. Example:

create table sls_stream(
  __timestamp__ bigint HEADER,
  __receive_time__ bigint HEADER
  b int,
  c varchar
) with (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou.log.aliyuncs.com',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'starttime' = '2023-08-30 00:00:00',
  'project' ='sls-test',
  'logstore' ='sls-input'
);

References

For more information about how to use the DataStream API of Realtime Compute for Apache Flink to consume log data, see DataStream API.