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

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}'
);

Connector options in the WITH clause

• General

  Option	Description	Data type	Required?	Default value	Remarks
  connector	The connector to use.	String	Yes	No default value	Set it to sls.
  endPoint	The endpoint of SLS.	String	Yes	No default value	Enter the VPC access address of SLS. For more information, see Endpoints. Note By default, Realtime Compute for Apache Flink cannot access the Internet. However, Alibaba Cloud provides NAT gateways to enable the communication between VPCs and the Internet. For more information, see How does Realtime Compute for Apache Flink access the Internet?. We recommend that you do not access SLS over the Internet. If you do need to access SLS over the Internet, use HTTPS and enable transfer acceleration for SLS.
  project	The name of the SLS project.	String	Yes	No default value
  logStore	The name of a SLS Logstore or Metricstore.	STRING	Yes	No default value	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	No default value	For more information, see How do I view the AccessKey pair of an account? Important To protect your AccessKey pair, configure your AccessKey by using variables.
  accessKey	The AccessKey secret of your Alibaba Cloud account.	STRING	Yes	No default value

• Source-specific

  Option	Description	Data type	Required?	Default value	Remarks
  enableNewSource	Specifies whether to use the FLIP-27 refactor source interface.	BOOLEAN	No	false	Enable this option and the source automatically adapts to shard changes and distribute shards across source subtasks as evenly as possible. Note Only VVR 8.0.9 or later supports this option. Important Starting from VVR 11.1, this option is set to true by default. If the option's value changes, your job cannot resume from a specific state. To resolve this, configure the consumerGroup option to record the current consumer offset and start your job. Then, set consumeFromCheckpoint to true and start the job without states. When source subtasks finish reading from read-only shards, they continue to request consuming other shards. This can cause uneven shard consumption among source subtasks, affecting overall job performance. To alleviate this issue, consider adjusting the source parallelism, optimizing your scheduling strategy, or merging small shards to simplify shard assignment.
  shardDiscoveryIntervalMs	The interval at which shard changes are dynamically detected.	LONG	No	60000	Unit: milliseconds. To disable dynamic detection, set the option to a negative value. Note The value of this option cannot be less than 1 minute (or 60,000 milliseconds). This option takes effect only if the enableNewSource option is set to true. Only VVR 8.0.9 or later supports this option.
  startupMode	The startup mode of the source table.	STRING	No	timestamp	timestamp: 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 consumption offset of a shard, logs are consumed from the earliest offset. Important In VVR versions earlier than 11.1, consumer_group is no longer supported. To consume data from a recorded offset in a specified consumer group, set consumeFromCheckpoint to true.
  startTime	The time at which logs start to be consumed.	STRING	No	Current time	The value of this option is in the yyyy-MM-dd hh:mm:ss format. This option takes effect only if startupMode is set to timestamp. Note The startTime and stopTime parameters are configured based on the __receive_time__ field in a SLS source table rather than on the __timestamp__ field.
  stopTime	The time at which log consumption is stopped.	String	No	No default value	The value of this option is in the yyyy-MM-dd hh:mm:ss format. Note To consume only historical logs, set this option to a specific historical time point. Using a future time point may cause consumption to stop unexpectedly if new log ingestion is temporarily disrupted. The observable symptom is the disruption of data streams without any accompanying error messages or exceptions. If you want Realtime Compute for Apache Flink programs to exit after log consumption is complete, you must also configure the exitAfterFinish option and set the exitAfterFinish option to true.
  consumerGroup	The name of the consumer group.	STRING	No	No default value	A consumer group records the consumption progress. You can specify a custom consumer group name. The format of the name is not fixed. Note A consumer group cannot be shared by multiple jobs for collaborative consumption. We recommend that you specify different consumer groups for different jobs. If you specify the same consumer group for different jobs, all data is consumed. When Realtime Compute for Apache Flink consumes data from SLS, the data is not sharded in a consumer group. Therefore, if multiple jobs share the same consumer group, all messages in the consumer group are consumed by each job.
  consumeFromCheckpoint	Specifies whether to consume logs from the checkpoint that is stored in the specified consumer group.	STRING	No	false	true: If you set this option to true, you must also specify a consumer group. Flink consumes logs from the checkpoint that is stored in the consumer group. If no checkpoint exists in the consumer group, Flink consumes logs from the time specified by the startTime option. false : Flink does not consume logs from the checkpoint that is stored in the specified consumer group. Important Starting from VVR 11.1, this option is no longer supported. You need to set startupMode to consumer_group.
  maxRetries	The number of retries that are allowed when data fails to be read from SLS.	String	No	3
  batchGetSize	The number of log groups to read in a request.	String	No	100	To prevent errors, set batchGetSize to a value less than 1000.
  exitAfterFinish	Specifies whether Realtime Compute for Apache Flink programs exit after data consumption is complete.	String	No	false	true false
  query Important This option was deprecated in VVR 11.3, but subsequent versions remain compatible.	The query statement that is used to preprocess data before data consumption.	STRING	No	No default value	Configuring this option to filter data from SLS before data consumption begins, reducing costs and improving data processing efficiency. For example, if you specify 'query' = '*| where request_method = ''GET''', Realtime Compute for Apache Flink filters data where its request_method field values are equal to GET before data consumption. Note Use SPL syntax when configuring this option. Important Only VVR 8.0.1 or later supports this option. This feature incurs fees from SLS. For details, see Billing.
  processor	The SLS processor. If both this option and query are set, query takes precedence.	STRING	No	No default value	This option is functionally equivalent to query, but we recommend using this option. For example, setting 'processor' = 'test-filter-processor' indicates that data will be filtered by the SLS processor before being consumed by Flink. Note Use SPL syntax when configuring this option. Important Only VVR 8.0.1 or later supports this option. This feature incurs fees from SLS. For details, see Billing.

• Sink-specific

  Option	Description	Data type	Required?	Default value	Remarks
  topicField	Specifies a field name. The value of this option overwrites the value of the __topic__ field to indicate the topic of the log.	String	No	No default value	The value of this option must be an existing field in the table.
  timeField	Specifies a field name. The value of this option overwrites the value of the __timestamp__ field to indicate the log write time.	String	No	Current time	The option must be set to an existing INT field. If no field is specified, the current time is used.
  sourceField	Specifies a field name. The value of this option overwrites the value of the __source__ attribute field to indicate the origin of the log. For example, the value is the IP address of the machine that generates the log.	STRING	No	No default value	The value of this option must be an existing field in the table.
  partitionField	Specifies a field name. A hash value is calculated based on the value of this parameter when data is written to SLS. Data that includes the same hash value is written to the same shard.	STRING	No	No default value	If you do not specify this option, each data entry is randomly written to an available shard.
  buckets	The number of buckets that are regrouped based on the hash value when the partitionField option is specified.	String	No	64	Valid value range: [1,256]. The value of this option 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 specific shards.
  flushIntervalMs	The interval at which data writing is triggered.	STRING	No	2000	Unit: milliseconds.
  writeNullProperties	Specifies whether to write null values as empty strings to SLS.	BOOLEAN	No	true	true false Note Only VVR 8.0.6 or later supports this option.

Data type mappings

Limits

Only VVR 11.1 or later supports data ingestion from SLS.

Syntax

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

Configuration options

Data type mappings

The data type mappings for data ingestion are as follows:

Schema inference and evolution

• Data pre-consumption and schema initialization

  The SLS connector maintains the schema of the current Logstore. Before reading data from the Logstore, the connector attempts to pre-consume up to maxPreFetchLogGroups log groups from each shard and initializes the schema by parsing and merging the schema of each log. Subsequently, before data consumption begins, a table creation event is generated based on the initialized schema.

  Note

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

• Primary key

  SLS logs do not contain primary keys. Manually add primary keys to the table in the transform module:

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

• Schema inference and evolution

  After schema initialization, if schema.inference.strategy is set to static, the connector parses each log entry based on the the 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 schema contains physical columns not in the current schema, the missing columns are added to the current schema, and nullable column addition events are generated.

  • If the inferred schema does not contain specific columns in the current schema, these columns are retained and their values are set to NULL.

  The SLS connector infers all fields as string ones. Currently, only column addition is supported. New columns are appended to the current schema as nullable columns.