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 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

• 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.

• 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.

Type mappings

Limitations

This feature is supported only by Realtime Compute for Apache Flink versions 11.1 and later.

Syntax

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

Parameters

Reuse an existing catalog

Starting from Realtime Compute for Apache Flink version 11.5, you can reference a built-in SLS Catalog created on the Data Management page in Flink CDC data ingestion jobs. This avoids the need to manually specify connection properties.

source:
  type: sls
  using.built-in-catalog: sls_catalog

Currently, data ingestion jobs can automatically reuse the following parameters from the built-in SLS Catalog:

• endpoint

• project

• accessId

• accessKey

To override these automatically reused parameters, you can explicitly define them in the YAML configuration. Parameters defined in the YAML file take precedence.

Data type mapping

If fixed-types is not configured, the following data type mapping applies:

If fixed-types is configured, the system parses data by using the specified types.

Schema inference and evolution

• Shard data pre-consumption and schema initialization

  The SLS connector maintains the schema of the Logstore it is reading from. Before reading data from the Logstore, the connector pre-consumes up to maxPreFetchLogGroups log groups from each shard. It parses the schema of each log entry and merges them to initialize the schema for the table. A table creation event is then generated based on this initial schema before data consumption begins.

  Note

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

• Primary key information

  Log Service (SLS) logs do not contain primary key information. You can manually add a primary key to the table by using transform rules:

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

• Schema inference and schema changes

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

  • If the inferred physical columns contain fields that are not in the current schema, the connector adds these fields to the schema and generates an event to add nullable columns.

  • If the inferred physical columns are missing fields that exist in the current schema, the connector retains these fields, populates their data with NULL, and does not generate a column deletion event.

  The SLS connector infers the data type of all fields in each log entry as String. Currently, only adding new columns is supported. The connector appends new columns to the end of the schema and sets them to nullable.