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

Realtime Compute for Apache Flink reads log data from Simple Log Service by creating a source table. This topic covers source table creation and attribute field extraction.

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

Only Ververica Runtime (VVR) 11.1 and later supports using SLS as a synchronous source for data ingestion defined in YAML.
The SLS connector guarantees only at-least-once semantics.
Do not set the source parallelism higher than the number of shards, as this wastes resources. Additionally, for Ververica Runtime (VVR) 8.0.5 and earlier, a change in the number of shards can cause the automatic failover feature to fail, preventing some shards from being consumed.

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 options

General

Parameter	Description	Type	Required	Default	Remarks
connector	The connector to use.	String	Yes	None	Set this to `sls`.
endPoint	The endpoint of Log Service (SLS).	String	Yes	None	Specify the VPC access address of Log Service (SLS). For more information, see Service endpoints. Note By default, Realtime Compute for Apache Flink cannot access the internet. To enable internet access from your Virtual Private Cloud (VPC), use a NAT Gateway. For more information, see How do I access the Internet?. We recommend that you do not access SLS over the internet. If you must do so, use HTTPS and enable transfer acceleration. For more information, see Manage transfer acceleration.
project	The name of the SLS project.	String	Yes	None	N/A
logStore	The name of the Logstore or MetricStore.	String	Yes	None	Data in a Logstore is consumed in the same way as data in a MetricStore.
accessId	The AccessKey ID of your Alibaba Cloud account.	String	Yes	None	For more information, see Obtain an AccessKey pair. Important To prevent your AccessKey pair from being exposed, we recommend that you use variables to specify the AccessKey ID and AccessKey secret. For more information, see Project variables.
accessKey	The AccessKey secret of your Alibaba Cloud account.	String	Yes	None
akless.assume-role.role-name	The name of the RAM role that is used to access SLS.	String	No	None	If accessId and accessKey are not specified, you must specify this parameter to access SLS resources by using a RAM role. The RAM role must be referenced in advance under `Access Security/RAM Roles`. Note This parameter is supported only in VVR 11.9.0 and later.

Source-specific

Parameter	Description	Type	Required	Default	Remarks
enableNewSource	Specifies whether to use the new data source that implements the FLIP-27 interface.	Boolean	No	false	The new source can automatically adapt to shard changes and distribute shards as evenly as possible across all source subtasks. Important This option is supported only in VVR 8.0.9 and later. The default value is `true` for VVR 11.1 and later. If you change this option's value, you cannot restore jobs from a saved state. To resume consumption from a historical offset, first start the job with the `consumerGroup` option to record the consumption progress in the SLS consumer group. Then, set the `consumeFromCheckpoint` option to `true` and restart the job without a state. If a Logstore has read-only shards, some subtasks might continue requesting data from other shards after finishing their own. This can cause an imbalanced workload, affecting performance. To mitigate this issue, you can adjust the parallelism, optimize the scheduling strategy, or merge small shards to reduce the number of shards and simplify task allocation.
shardDiscoveryIntervalMs	The interval for dynamic shard discovery.	Long	No	60000	Set this option to a negative value to disable dynamic discovery. Unit: milliseconds. Note The value must be greater than or equal to 60,000 milliseconds (1 minute). This option takes effect only when `enableNewSource` is set to `true`. This option is supported only in VVR 8.0.9 and later.
startupMode	The startup mode for the source table.	String	No	timestamp	`timestamp` (default): Consume logs starting from the specified start time. `latest`: Start consuming logs from the latest offset. `earliest`: Starts consuming logs from the earliest offset. `consumer_group`: Starts log consumption from the offset recorded by the consumer group. If the consumer group has not recorded a consumption offset for a shard, consumption starts from the earliest offset. Important For VVR versions earlier than 11.1, the consumer_group value is not supported. You must set `consumeFromCheckpoint` to `true`. In this case, log consumption starts from the offset recorded by the specified consumer group, and the startup mode setting will not take effect.
startTime	The start time for log consumption.	String	No	Current time	The format is `yyyy-MM-dd hh:mm:ss`. This takes effect only when `startupMode` is set to `timestamp`. Note The `startTime` and `stopTime` options are based on the `__receive_time__` attribute in SLS, rather than the `__timestamp__` attribute.
stopTime	The end time for log consumption.	String	No	None	The format is `yyyy-MM-dd hh:mm:ss`. Note This option is used only for consuming historical logs and must be set to a time in the past. If you set it to a future time, consumption may stop prematurely if no new logs are written, resulting in a data flow interruption without any error messages. If you want the Flink job to exit after all logs are consumed, you must also set `exitAfterFinish` to `true`.
consumerGroup	The name of the consumer group.	String	No	None	A consumer group is used to record consumption progress. You can specify a custom name without a fixed format. Note Different Flink jobs must use different consumer groups. If multiple Flink jobs use the same consumer group, they do not coordinate and each job consumes all the data. This is because Flink does not use the SLS consumer group for partition assignment when it consumes data from SLS. As a result, each consumer consumes messages independently, even if they share the same consumer group.
consumeFromCheckpoint	Specifies whether to consume from a consumer group checkpoint.	String	No	false	`true`: You must also specify a consumer group. The Flink program starts consuming logs from the checkpoint saved in the consumer group. If the consumer group does not have a corresponding checkpoint, consumption starts from the startTime configuration value. `false` (default value): Does not start consuming logs from the checkpoint saved for the specified consumer group. Important This parameter is no longer supported in VVR 11.1 and later. For these versions, you must set the `startupMode` option to `consumer_group`.
maxRetries	The number of retries after a failed attempt to read from SLS.	String	No	3	N/A
batchGetSize	The number of log groups to read in a single request.	String	No	100	The `batchGetSize` setting cannot exceed 1000, or an error is reported.
exitAfterFinish	Specifies whether the Flink job exits after all data is consumed.	String	No	false	`true`: The Flink program exits after all data is consumed. `false` (default): The Flink program does not exit after data consumption is complete.
query Important This option was deprecated in VVR 11.3, but later versions remain compatible.	The query statement for preprocessing data before consumption.	String	No	None	Use this option to filter SLS data before Flink consumes it. This reduces costs and improves processing speed. For example, `'query' = '\| where request_method = ''GET'''` indicates that before Flink reads data from SLS, it first matches data where the value of the `request_method` field is 'GET'. Note* This option uses the SPL language of Log Service (SLS). For more information, see SPL syntax. Important This option is supported only in VVR 8.0.1 and later. This feature incurs Log Service (SLS) fees. For more information, see Billing of Log Service.
processor	The name of the SLS processor for data preprocessing. If both `query` and `processor` are specified, `query` takes precedence and `processor` is ignored.	String	No	None	This option filters SLS data before Flink consumes it, which reduces costs and improves processing speed. We recommend that you use `processor` instead of `query`. For example, `'processor' = 'test-filter-processor'` indicates that the SLS processor filters data before Flink reads the data from SLS. Note This option uses the SPL language of Log Service (SLS). For more information, see SPL syntax. For information about how to create or update an SLS processor, see Manage processors. Important This option is supported only in VVR 11.3 and later. This feature incurs Log Service (SLS) fees. For more information, see Billing of Log Service.
preserveRawBytes	Specifies whether to directly fetch and preserve the raw bytes carried by SLS when `enableNewSource` is set to `true`.	Boolean	No	false	When this option is enabled, BINARY/VARBINARY fields read the raw byte[] directly, and CHAR/VARCHAR fields construct string data from the raw bytes. Other fields still use the default conversion logic. The behavior is consistent with that of the old source. Note This option is supported only in VVR 11.8 and later. This option takes effect only when `enableNewSource` is set to `true`.

Sink-specific

Parameter	Description	Type	Required	Default	Remarks
topicField	Specifies a field whose value overwrites the `__topic__` attribute, which indicates the log topic.	String	No	None	The value of this option must be an existing field in the table.
timeField	Specifies a field whose value overwrites the `__timestamp__` attribute, which indicates the log write time.	String	No	Current time	The value of this option must be an existing `INT` field in the table. If this option is not specified, the current time is used.
sourceField	Specifies a field whose value overwrites the `__source__` attribute, which indicates the log source, such as the IP address of the machine that generated the log.	String	No	None	The value of this option must be an existing field in the table.
partitionField	Specifies a field for partitioning. A hash of this field's value determines which shard receives the data, ensuring records with the same hash go to the same shard.	String	No	None	If this option is not specified, each record is randomly written to an available shard.
buckets	If `partitionField` is specified, this option defines the number of buckets for mapping hash values.	String	No	64	The value must be a power of 2 in the range [1, 256]. The number of buckets must be greater than or equal to the number of shards. Otherwise, some shards may not receive any data.
flushIntervalMs	The data write interval.	String	No	2000	Unit: milliseconds.
writeNullProperties	Specifies whether to write null values as empty strings to SLS.	Boolean	No	true	`true` (default value): Writes null values to the log as empty strings. `false`: Fields that evaluate to null are not written to the log. Note This option is supported only in VVR 8.0.6 and 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.