Realtime Compute for Apache Flink:Simple Log Service (SLS)

Syntax

CREATE TABLE sls_table(
  a INT,
  b INT,
  c VARCHAR
) WITH (
  'connector' = 'sls',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'logStore' = '<yourLogStoreName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}'
);

WITH parameters

• General

  Parameter	Description	Data type	Required	Default value	Remarks
  connector	The table type.	String	Yes	None	Set this parameter to sls.
  endPoint	The endpoint.	String	Yes	None	Enter the private network endpoint of SLS. For more information, see Endpoints. Note By default, Realtime Compute for Apache Flink cannot access the public network. However, Alibaba Cloud provides NAT Gateway to enable communication between a VPC and the public network. For more information, see How do I access the public network?. We do not recommend that you access SLS over the public network. If you must access SLS over the public network, use the HTTPS protocol and enable Global Accelerator (GA) for SLS. For more information, see Manage transfer acceleration.
  project	The name of the SLS project.	String	Yes	None	None.
  logStore	The name of the SLS Logstore or metricstore.	String	Yes	None	Data in a Logstore is consumed in the same way as in a metricstore.
  accessId	The AccessKey ID of your Alibaba Cloud account.	String	Yes	None	For more information, see How do I view the AccessKey ID and AccessKey secret?. Important To prevent your AccessKey information from being leaked, we recommend that you use variables to specify the AccessKey pair. For more information, see Manage variables.
  accessKey	The AccessKey secret of your Alibaba Cloud account.	String	Yes	None

• Source table-specific

  Parameter	Description	Data type	Required	Default value	Remarks
  enableNewSource	Specifies whether to enable the new data source that implements the FLIP-27 interface.	Boolean	No	false	The new data source can automatically adapt to shard changes and ensure that shards are distributed as evenly as possible across all source concurrencies. Important This parameter is supported only in Realtime Compute for Apache Flink VVR 8.0.9 or later. This parameter is set to true by default from VVR 11.1. A job cannot be restored from a state after this configuration item is changed. To resolve this issue, you can first set the consumerGroup configuration item to start the job and record the consumption progress in the SLS consumer group. Then, set the consumeFromCheckpoint configuration item to true and start the job in a stateless manner. This way, the job can resume consumption from the historical progress. If read-only shards exist in SLS, some concurrent Flink tasks continue to request data from other unfinished shards after they finish consuming data from the read-only shards. This can lead to some concurrent tasks being assigned to multiple shards, causing an imbalance in shard distribution among them. This imbalance affects overall consumption efficiency and system performance. To mitigate this issue, adjust the concurrency, optimize the task scheduling policy, or merge small shards to reduce the number of shards and the complexity of task assignment.
  shardDiscoveryIntervalMs	The interval at which shard changes are dynamically detected. Unit: millisecond.	Long	No	60000	Set this parameter to a negative value to disable dynamic detection. Note The value of this parameter cannot be less than 1 minute (60,000 milliseconds). This parameter takes effect only when the enableNewSource parameter is set to true. This parameter is supported only in Realtime Compute for Apache Flink VVR 8.0.9 or later.
  startupMode	The startup mode of the source table.	String	No	timestamp	timestamp (default): Logs are consumed from the specified start time. latest: Logs are consumed from the latest offset. earliest: Logs are consumed from the earliest offset. consumer_group: Logs are consumed from the offset recorded in the consumer group. If the consumer group does not record the consumer offset of a shard, logs are consumed from the earliest offset. Important In VVR versions earlier than 11.1, the consumer_group value is not supported. You must set consumeFromCheckpoint to true. In this case, logs are consumed from the offset recorded in the specified consumer group, and the startup mode specified here does not take effect.
  startTime	The time when log consumption starts.	String	No	Current time	The format is yyyy-MM-dd hh:mm:ss. This parameter takes effect only when startupMode is set to timestamp. Note The startTime and stopTime parameters are based on the __receive_time__ attribute in SLS, not the __timestamp__ attribute.
  stopTime	The time when log consumption ends.	String	No	None	The format is yyyy-MM-dd hh:mm:ss. Note This parameter is used only to consume historical logs and should be set to a past point in time. If you set it to a future time, consumption may stop prematurely if no new logs are written. This appears as a data stream interruption with no error message. If you want the Flink program to exit after log consumption is complete, you must also set exitAfterFinish=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 for the consumer group. The format is not fixed. Note Multiple jobs cannot use the same consumer group for collaborative consumption. Different Flink jobs should have different consumer groups. If different Flink jobs use the same consumer group, they will consume all the data. This is because when Flink consumes data from SLS, it does not perform partition allocation through the SLS consumer group. As a result, each consumer consumes its own messages independently, even if the consumer group is the same.
  consumeFromCheckpoint	Specifies whether to consume logs from the checkpoint that is saved in the specified consumer group.	String	No	false	true: You must also specify a consumer group. The Flink program consumes logs from the checkpoint that is saved in the consumer group. If the consumer group does not have a corresponding checkpoint, logs are consumed from the time specified by the startTime parameter. false (default): Logs are not consumed from the checkpoint that is saved in the specified consumer group. Important This parameter is no longer supported from VVR 11.1. For VVR 11.1 and later, you must set startupMode to consumer_group.
  maxRetries	The number of retries after a failed attempt to read data from SLS.	String	No	3	None.
  batchGetSize	The number of log groups to read in a single request.	String	No	100	The value of batchGetSize cannot exceed 1,000. Otherwise, an error is reported.
  exitAfterFinish	Specifies whether the Flink program exits after data consumption is complete.	String	No	false	true: The Flink program exits after data consumption is complete. false (default): The Flink program does not exit after data consumption is complete.
  query Important This parameter is deprecated in VVR 11.3 and later, but remains compatible with subsequent versions.	The pre-processing statement for SLS data consumption.	String	No	None	Using the query parameter lets you filter data before it is consumed from SLS. This prevents all data from being consumed into Flink, which saves costs and improves processing speed. For example, 'query' = '*| where request_method = ''GET''' indicates that before Flink reads data from SLS, it matches data where the value of the request_method field is 'get'. Note The query parameter requires the use of Structured Process Language (SPL) for Simple Log Service. For more information, see SPL syntax. Important This parameter is supported only in Realtime Compute for Apache Flink VVR 8.0.1 or later. This feature incurs fees from Simple Log Service. For more information, see Billing.
  processor	The SLS Consum Processor. If both this parameter and `query` are configured, `query` takes precedence and `processor` does not take effect.	String	No	None	Using the processor parameter lets you filter data before it is consumed by Flink. This helps save costs and improve processing speed. We recommend that you use the processor parameter instead of the query parameter. For example, 'processor' = 'test-filter-processor' indicates that data is filtered by the SLS Consum Processor before it is read by Flink. Note The processor must use the Structured Process Language (SPL) of Simple Log Service (SLS). For more information, see SPL syntax. To create and update SLS Consum Processors, see Manage Consum Processors. Important This parameter is supported only in Realtime Compute for Apache Flink VVR 11.3 or later. This feature incurs fees from Simple Log Service. For more information, see Billing.

• For sink tables only

  Parameter	Description	Data type	Required	Default value	Remarks
  topicField	Specifies a field name. The value of this field overwrites the value of the __topic__ attribute field and indicates the topic of the log.	String	No	None	The value of this parameter must be one of the existing fields in the table.
  timeField	Specifies a field name. The value of this field overwrites the value of the __timestamp__ attribute field and indicates the time when the log is written.	String	No	Current time	The value of this parameter must be one of the existing fields in the table, and the field type must be INT. If this parameter is not specified, the current time is used by default.
  sourceField	Specifies a field name. The value of this field overwrites the value of the __source__ attribute field and indicates the source of the log, such as the IP address of the machine that generates the log.	String	No	None	The value of this parameter must be one of the existing fields in the table.
  partitionField	Specifies a field name. When data is written, a hash value is calculated based on the value of this column. Data with the same hash value is written to the same shard.	String	No	None	If this parameter is not specified, each piece of data is randomly written to a currently available shard.
  buckets	The number of groups that are re-grouped based on the hash value when partitionField is specified.	String	No	64	The value of this parameter must be in the range of [1, 256] and must be an integer power of 2. The number of buckets must be greater than or equal to the number of shards. Otherwise, no data is written to some shards.
  flushIntervalMs	The period that triggers data writes.	String	No	2000	Unit: millisecond.
  writeNullProperties	Specifies whether to write null values as empty strings to SLS.	Boolean	No	true	true (default): Null values are written as empty strings to logs. false: Fields with null values are not written to logs. Note This parameter is supported only in Realtime Compute for Apache Flink VVR 8.0.6 or later.

Type mapping

Limits

This feature is supported only in Realtime Compute for Apache Flink VVR 11.1 or later.

Syntax

source:
   type: sls
   name: SLS Source
   endpoint: <endpoint>
   project: <project>
   logstore: <logstore>
   accessId: <accessId>
   accessKey: <accessKey>

Configuration items

Type mapping

The following table shows the type mapping for data ingestion.

Table schema inference and change synchronization

• Shard data pre-consumption and table schema initialization

  The SLS connector maintains the schema of the Logstore that is being read. Before it reads data from the Logstore, the SLS connector attempts to pre-consume a maximum of maxPreFetchLogGroups log groups from each shard. The connector then parses the schema of each log in these log groups and merges the schemas to initialize the table schema. Before the actual data consumption starts, a corresponding table creation event is generated based on the initialized schema.

  Note

  For each shard, the SLS connector attempts to consume data and parse the log schema starting from one hour before the current time.

• Primary key information

  SLS logs do not contain primary key information. You can manually add primary keys to a table using transform rules:

  transform:
    - source-table: <project>.<logstore>
      projection: \*
      primary-keys: key1, key2

• Schema inference and schema evolution

  After the table schema is initialized, if schema.inference.strategy is set to static, the SLS connector parses each log based on the initial table schema and does not generate schema change events. If schema.inference.strategy is set to continuous, the SLS connector parses the data of each log, infers the physical columns, and compares them with the currently recorded schema. If the inferred schema is inconsistent with the current schema, the schemas are merged based on the following rules:

  • If the inferred physical columns contain fields that are not in the current schema, these fields are added to the schema, and an event for adding nullable columns is generated.

  • If the inferred physical columns do not contain fields that are already in the current schema, these fields are retained, their data is filled with NULL, and no event for deleting columns is generated.

  The SLS connector infers the type of all fields in each log as String. Currently, only adding columns is supported. New columns are added to the end of the current schema and are set as nullable columns.